diff --git a/huggingface/wechat-ai-tools/.env.example b/huggingface/wechat-ai-tools/.env.example index f7528585..d6b818a6 100644 --- a/huggingface/wechat-ai-tools/.env.example +++ b/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= diff --git a/huggingface/wechat-ai-tools/README.md b/huggingface/wechat-ai-tools/README.md index 96a95dc9..aa3a550f 100644 --- a/huggingface/wechat-ai-tools/README.md +++ b/huggingface/wechat-ai-tools/README.md @@ -50,6 +50,18 @@ UPSTREAM_DENY_PRIVATE=true 鉴权:`Authorization: Bearer ` 或 `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 兼容子集。用户自定义模型示例: diff --git a/huggingface/wechat-ai-tools/requirements.txt b/huggingface/wechat-ai-tools/requirements.txt index ac0cde3a..9cb9dfea 100644 --- a/huggingface/wechat-ai-tools/requirements.txt +++ b/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 diff --git a/huggingface/wechat-ai-tools/services/web_search.py b/huggingface/wechat-ai-tools/services/web_search.py index 845122f2..e8b4131e 100644 --- a/huggingface/wechat-ai-tools/services/web_search.py +++ b/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 diff --git a/huggingface/wechat-ai-tools/tests/test_web_search.py b/huggingface/wechat-ai-tools/tests/test_web_search.py new file mode 100644 index 00000000..70e72b0d --- /dev/null +++ b/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"]