Add optional Parallel Search MCP backend to tools gateway

huggingface/wechat-ai-tools/.env.example

@@ -16,6 +16,7 @@ UPSTREAM_TIMEOUT_MS=60000
 MAX_BODY_BYTES=1048576
 
 # Search
+# ddg (default), searxng, tavily, or parallel (no Parallel API key needed)
 SEARCH_PROVIDER=ddg
 # SEARXNG_URL=https://searx.example/search
 # TAVILY_API_KEY=

huggingface/wechat-ai-tools/README.md

@@ -50,6 +50,18 @@ UPSTREAM_DENY_PRIVATE=true
 
 鉴权：`Authorization: Bearer <TOOLS_API_KEY>` 或 `X-API-Key`。
 
+### 搜索后端 Search backends
+
+`SEARCH_PROVIDER` defaults to `ddg`. To use [Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp) for free, keyless web search, set this in the **tools service** environment and restart it:
+
+```env
+SEARCH_PROVIDER=parallel
+```
+
+Anonymous search uses Fast mode and needs no Parallel account or API key. It returns the same `title`, `url`, and `snippet` fields used by the main site's search tool and Chatflow search node. Existing `ddg`, `searxng` (`SEARXNG_URL`), and `tavily` (`TAVILY_API_KEY`) selections remain available. Requests still require the configured gateway `TOOLS_API_KEY`; the main site's global and persona search switches still apply.
+
+`SEARCH_TIMEOUT_MS` bounds the complete Parallel handshake and search; `SEARCH_MAX_RESULTS` caps returned results. Provider failures are reported without silently switching backends. The gateway's search API does not expose page fetching or conversation identifiers.
+
 ### `POST /v1/chat/completions`
 
 OpenAI 兼容子集。用户自定义模型示例：

huggingface/wechat-ai-tools/requirements.txt

@@ -2,6 +2,7 @@
 fastapi==0.115.12
 uvicorn[standard]==0.34.2
 httpx==0.28.1
+mcp==1.30.0
 pydantic==2.11.3
 pydantic-settings==2.8.1
 duckduckgo-search==8.0.1

huggingface/wechat-ai-tools/services/web_search.py

@@ -1,9 +1,12 @@
-"""Web search backends (DDG default; optional SearXNG / Tavily)."""
+"""Web search backends (DDG default; optional SearXNG / Tavily / Parallel)."""
 
 from __future__ import annotations
 
+import asyncio
+import json
 import logging
 from typing import Any
+from urllib.parse import urlsplit
 
 import httpx
 
@@ -37,6 +40,8 @@ async def search_web(
     limit = _clamp_max(max_results, settings)
     provider = (settings.search_provider or "ddg").strip().lower()
 
+    if provider == "parallel":
+        return await _search_parallel(q, limit, settings)
     if provider == "tavily":
         return await _search_tavily(q, limit, settings)
     if provider == "searxng":
@@ -122,3 +127,72 @@ async def _search_tavily(
             }
         )
     return out
+
+
+async def _search_parallel(
+    query: str, limit: int, settings: Settings
+) -> list[dict[str, str]]:
+    """Call anonymous Search MCP through the official Streamable HTTP client."""
+    from mcp import ClientSession
+    from mcp.client.streamable_http import streamable_http_client
+
+    timeout = max(1.0, settings.search_timeout_ms / 1000.0)
+    try:
+        # Bound the entire handshake, discovery and search, not each HTTP request.
+        async with asyncio.timeout(timeout):
+            async with httpx.AsyncClient(
+                timeout=timeout,
+                headers={"User-Agent": "wechat-ai-tools/1.0.0"},
+            ) as client:
+                async with streamable_http_client(
+                    "https://search.parallel.ai/mcp", http_client=client
+                ) as (read, write, _):
+                    async with ClientSession(read, write) as session:
+                        await session.initialize()
+                        tools = await session.list_tools()
+                        if not any(tool.name == "web_search" for tool in tools.tools):
+                            raise SearchError("Parallel web_search tool unavailable")
+                        # The gateway receives independent searches, without a
+                        # conversation identity. Leave optional session_id unset
+                        # rather than inventing a shared or per-call identity.
+                        result = await session.call_tool(
+                            "web_search",
+                            {"objective": query, "search_queries": [query]},
+                        )
+                        if result.isError:
+                            raise SearchError("Parallel search returned an error")
+                        data = result.structuredContent
+                        if data is None:
+                            text = "\n".join(
+                                part.text for part in result.content if part.type == "text"
+                            )
+                            data = json.loads(text)
+    except TimeoutError as exc:
+        raise SearchError("Parallel search timed out") from exc
+    except Exception as exc:  # SDK transport failures can be nested exception groups.
+        log.warning("parallel search failed: %s", type(exc).__name__)
+        raise SearchError("Parallel search failed") from exc
+
+    if not isinstance(data, dict) or not isinstance(data.get("results"), list):
+        raise SearchError("Invalid Parallel search response")
+    out: list[dict[str, str]] = []
+    for item in data["results"][:limit]:
+        if not isinstance(item, dict):
+            raise SearchError("Invalid Parallel search result")
+        url = item.get("url")
+        title = item.get("title")
+        excerpts = item.get("excerpts")
+        try:
+            parsed = urlsplit(url) if isinstance(url, str) else None
+            valid_url = parsed and parsed.scheme in {"http", "https"} and parsed.hostname
+        except ValueError:
+            valid_url = False
+        if (
+            not valid_url
+            or (title is not None and not isinstance(title, str))
+            or not isinstance(excerpts, list)
+            or any(not isinstance(excerpt, str) for excerpt in excerpts)
+        ):
+            raise SearchError("Invalid Parallel search result")
+        out.append({"title": title or "", "url": url, "snippet": "\n\n".join(excerpts)})
+    return out

huggingface/wechat-ai-tools/tests/test_web_search.py

@@ -0,0 +1,178 @@
+"""Exercise the route, provider resolver, and real MCP SDK HTTP transport."""
+import asyncio
+import json
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+# Import before patching httpx.AsyncClient: SDK annotations reference its class.
+import mcp.client.streamable_http  # noqa: F401
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from config import Settings, get_settings
+from routers.search import router
+from services import web_search
+
+
+class McpFixture:
+    def __init__(self):
+        self.requests = []
+        self.results = [
+            {"title": "Current facts", "url": "https://example.org/news?a=1", "excerpts": ["First", "Second"]},
+            {"title": None, "url": "http://example.org/next", "excerpts": []},
+        ]
+        self.tool_error = False
+        self.http_error = False
+        self.missing_tool = False
+        self.text_only = False
+        self.sse = False
+        self.delay = False
+        self.started = asyncio.Event()
+        self.cancelled = False
+
+    async def handle(self, request):
+        payload = json.loads(request.content) if request.content else {}
+        self.requests.append({"url": str(request.url), "method": request.method,
+                              "headers": dict(request.headers), "payload": payload})
+        if request.method == "GET":
+            return httpx.Response(405)
+        method = payload.get("method")
+        if method == "initialize":
+            result = {"protocolVersion": payload["params"]["protocolVersion"],
+                      "capabilities": {"tools": {}},
+                      "serverInfo": {"name": "fixture", "version": "1"}}
+        elif method == "tools/list":
+            result = {"tools": [] if self.missing_tool else [{"name": "web_search", "inputSchema": {"type": "object"}}]}
+        elif method == "tools/call":
+            self.started.set()
+            if self.delay:
+                try:
+                    await asyncio.Event().wait()
+                except asyncio.CancelledError:
+                    self.cancelled = True
+                    raise
+            if self.http_error:
+                return httpx.Response(503, text="unavailable")
+            data = {"results": self.results}
+            result = {"content": [{"type": "text", "text": json.dumps(data)}], "isError": self.tool_error}
+            if not self.text_only:
+                result["structuredContent"] = data
+        else:
+            return httpx.Response(202)
+        body = {"jsonrpc": "2.0", "id": payload["id"], "result": result}
+        if self.sse and method == "tools/call":
+            return httpx.Response(200, headers={"content-type": "text/event-stream"},
+                                  text="event: message\ndata: " + json.dumps(body) + "\n\n")
+        return httpx.Response(200, json=body)
+
+
+@pytest.fixture
+def backend(monkeypatch):
+    fixture = McpFixture()
+    real_client = httpx.AsyncClient
+    monkeypatch.setattr(web_search.httpx, "AsyncClient", lambda **kwargs: real_client(
+        **kwargs, transport=httpx.MockTransport(fixture.handle)))
+    return fixture
+
+
+def route_client(settings):
+    app = FastAPI()
+    app.include_router(router)
+    app.dependency_overrides[get_settings] = lambda: settings
+    return TestClient(app)
+
+
+def test_actual_route_anonymous_parallel(backend):
+    settings = Settings(_env_file=None, SEARCH_PROVIDER=" Parallel ", TOOLS_API_KEY="gateway-only", SEARCH_MAX_RESULTS=1)
+    with route_client(settings) as client:
+        response = client.post("/v1/web-search", headers={"Authorization": "Bearer gateway-only"},
+                               json={"query": " latest facts ", "max_results": 8})
+    assert response.status_code == 200
+    assert response.json() == {"query": "latest facts", "results": [
+        {"title": "Current facts", "url": "https://example.org/news?a=1", "snippet": "First\n\nSecond"}]}
+    assert {r["payload"].get("method") for r in backend.requests} >= {"initialize", "tools/list", "tools/call"}
+    for request in backend.requests:
+        assert request["url"] == "https://search.parallel.ai/mcp"
+        assert request["headers"]["user-agent"] == "wechat-ai-tools/1.0.0"
+        assert "authorization" not in request["headers"]
+    call = next(r["payload"] for r in backend.requests if r["payload"].get("method") == "tools/call")
+    assert call["params"] == {"name": "web_search", "arguments": {
+        "objective": "latest facts", "search_queries": ["latest facts"]}}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("text_only,sse", [(True, False), (False, True)])
+async def test_sdk_json_text_and_sse(backend, text_only, sse):
+    backend.text_only, backend.sse = text_only, sse
+    hits = await web_search.search_web("facts", max_results=10, settings=Settings(_env_file=None, SEARCH_PROVIDER="parallel"))
+    assert len(hits) == 2
+    assert hits[1] == {"title": "", "url": "http://example.org/next", "snippet": ""}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("provider", [None, "ddg", "searxng", "tavily"])
+async def test_existing_provider_choices(monkeypatch, provider):
+    settings = Settings(_env_file=None, **({"SEARCH_PROVIDER": provider} if provider else {}))
+    expected = provider or "ddg"
+    for name in ["ddg", "searxng", "tavily", "parallel"]:
+        stub = AsyncMock(return_value=[{"title": name, "url": "https://example.org", "snippet": "ok"}])
+        monkeypatch.setattr(web_search, "_search_" + name, stub)
+    result = await web_search.search_web("q", max_results=3, settings=settings)
+    assert result[0]["title"] == expected
+    for name in ["ddg", "searxng", "tavily", "parallel"]:
+        assert getattr(web_search, "_search_" + name).await_count == (1 if name == expected else 0)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["tool_error", "http_error", "missing_tool", "bad_shape", "bad_url", "bad_excerpts"])
+async def test_parallel_failures_never_fallback(backend, monkeypatch, failure):
+    fallback = AsyncMock(side_effect=AssertionError("must not fall back"))
+    monkeypatch.setattr(web_search, "_search_ddg", fallback)
+    if failure == "bad_shape":
+        backend.results = "invalid"
+    elif failure == "bad_url":
+        backend.results[0]["url"] = "javascript:alert(1)"
+    elif failure == "bad_excerpts":
+        backend.results[0]["excerpts"] = [None]
+    else:
+        setattr(backend, failure, True)
+    with pytest.raises(web_search.SearchError):
+        await web_search.search_web("facts", max_results=3, settings=Settings(_env_file=None, SEARCH_PROVIDER="parallel"))
+    fallback.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_parallel_total_timeout(backend):
+    backend.delay = True
+    with pytest.raises(web_search.SearchError, match="timed out"):
+        await web_search.search_web("facts", max_results=3, settings=Settings(_env_file=None, SEARCH_PROVIDER="parallel", SEARCH_TIMEOUT_MS=1000))
+    assert backend.cancelled
+
+
+@pytest.mark.asyncio
+async def test_parallel_cancellation(backend):
+    backend.delay = True
+    task = asyncio.create_task(web_search.search_web("facts", max_results=3, settings=Settings(_env_file=None, SEARCH_PROVIDER="parallel")))
+    await asyncio.wait_for(backend.started.wait(), timeout=2)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert backend.cancelled
+
+
+def test_gateway_auth_and_input_bounds(backend):
+    with route_client(Settings(_env_file=None, SEARCH_PROVIDER="parallel", TOOLS_API_KEY="private")) as client:
+        for headers in [{}, {"X-API-Key": "wrong"}]:
+            assert client.post("/v1/web-search", headers=headers, json={"query": "facts"}).status_code == 401
+        for body in [{"query": ""}, {"query": "x" * 501}, {"query": "facts", "max_results": 11}]:
+            assert client.post("/v1/web-search", headers={"X-API-Key": "private"}, json=body).status_code == 422
+    assert backend.requests == []
+
+
+def test_gateway_reports_parallel_error(backend):
+    backend.tool_error = True
+    with route_client(Settings(_env_file=None, SEARCH_PROVIDER="parallel", TOOLS_API_KEY="")) as client:
+        response = client.post("/v1/web-search", json={"query": "facts"})
+    assert response.status_code == 400
+    assert "Parallel search" in response.json()["detail"]