From 397a46f85018aaadfd1134daf249f6099eba393e Mon Sep 17 00:00:00 2001 From: host Date: Sun, 1 Nov 2026 08:16:48 +0000 Subject: [PATCH] websearch: add scope=code programming search + provider extract chain - search(..., scope=code) restricts to 15 programming domains (official docs, GitHub, registries, Stack Overflow, CVE/NVD): native include_domains/search_domain_filter where supported (Tavily, Exa, Perplexity), post-filtering for all providers (Brave, SearXNG, DuckDuckGo have no native filter). - web_fetch now uses fetch_best: Tavily extract, then Exa contents, then direct fetch; both-fail errors name both failures, direct-fetch 403 under the egress whitelist is fail-closed, not bypassed. Default general scope and direct fetch_url behaviour unchanged. - MCP web_search gains scope enum (general/code); docs updated. - Tests all mock HTTP, no real keys. --- cloud/tests/test_code_search.py | 358 ++++++++++++++++++++++++++++++ cloud/tests/test_websearch_mcp.py | 5 +- cloud/websearch.py | 315 ++++++++++++++++++++++++-- cloud/websearch_mcp.py | 32 ++- docs/web-search-mcp.md | 129 ++++++++++- 5 files changed, 803 insertions(+), 36 deletions(-) create mode 100644 cloud/tests/test_code_search.py diff --git a/cloud/tests/test_code_search.py b/cloud/tests/test_code_search.py new file mode 100644 index 0000000..784f58a --- /dev/null +++ b/cloud/tests/test_code_search.py @@ -0,0 +1,358 @@ +"""Programming search scope (scope="code") + provider extract chain. + +All HTTP is mocked; no network, no real keys. Fake key values are +distinctive so a leak into an error string or result is caught by +assertion. +""" + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud import websearch as _ws + from cloud.websearch import ( + CODE_SEARCH_DOMAINS, WebSearchError, _is_code_domain, + ensure_codex_home_websearch, fetch_best, fetch_via_provider, + search) + from cloud import websearch_mcp as _mcp + from cloud.websearch_mcp import handle_request +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud import websearch as _ws + from cloud.websearch import ( + CODE_SEARCH_DOMAINS, WebSearchError, _is_code_domain, + ensure_codex_home_websearch, fetch_best, fetch_via_provider, + search) + from cloud import websearch_mcp as _mcp + from cloud.websearch_mcp import handle_request + +_FAKE_KEYS = { + "PERPLEXITY_API_KEY": "pplx-SECRET-111", + "EXA_API_KEY": "exa-SECRET-222", + "TAVILY_API_KEY": "tvly-SECRET-333", + "BRAVE_API_KEY": "brave-SECRET-444", +} + + +class FakeHttp: + """post_json/get keyed by URL substring, in route order.""" + + def __init__(self, routes): + self.routes = routes + self.seen = [] + + def post_json(self, url, headers, payload): + self.seen.append(("POST", url, payload)) + return self._route(url) + + def get(self, url, headers=None): + self.seen.append(("GET", url, headers)) + return self._route(url) + + def _route(self, url): + for needle, body, status in self.routes: + if needle in url: + if status != 200: + raise WebSearchError("HTTP %s" % status) + if isinstance(body, dict): + return 200, json.dumps(body).encode(), \ + "application/json" + return 200, body.encode(), "text/html" + raise WebSearchError("HTTP 404") + + +def _tavily_body(urls): + return {"results": [{"title": "T %s" % u, "url": u, + "content": "snippet %s" % u} for u in urls]} + + +class CodeScopeTestCase(unittest.TestCase): + def test_tavily_code_sends_include_domains_and_filters(self): + http = FakeHttp([ + ("api.tavily.com", _tavily_body([ + "https://github.com/python/cpython", + "https://news.example.com/python-312", + "https://docs.python.org/3/library/pathlib.html", + ]), 200), + ]) + out = search("pathlib changes", provider="tavily", + scope="code", env=dict(_FAKE_KEYS), http=http) + payload = http.seen[0][2] + self.assertEqual(payload["include_domains"], + list(CODE_SEARCH_DOMAINS)) + self.assertEqual(out["scope"], "code") + self.assertEqual(out["code_domains"], list(CODE_SEARCH_DOMAINS)) + kept = [r["url"] for r in out["results"]] + self.assertEqual(kept, ["https://github.com/python/cpython", + "https://docs.python.org/3/library/" + "pathlib.html"]) + self.assertEqual(out["attempted"][0]["domain_filtered_out"], 1) + for secret in _FAKE_KEYS.values(): + self.assertNotIn(secret, json.dumps(out)) + + def test_exa_code_sends_include_domains(self): + http = FakeHttp([ + ("api.exa.ai", {"results": [ + {"title": "E", "url": "https://crates.io/crates/serde", + "text": "serde", "score": 0.9}]}, 200), + ]) + out = search("serde", provider="exa", scope="code", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(http.seen[0][2]["include_domains"], + list(CODE_SEARCH_DOMAINS)) + self.assertEqual(out["results"][0]["url"], + "https://crates.io/crates/serde") + + def test_perplexity_code_sends_domain_filter_within_cap(self): + self.assertLessEqual(len(CODE_SEARCH_DOMAINS), 20) + http = FakeHttp([ + ("api.perplexity.ai", + {"choices": [{"message": {"content": "answer"}}], + "search_results": [ + {"title": "P", "url": "https://go.dev/doc/", + "snippet": "go docs"}, + {"title": "Ad", "url": "https://shop.example.com/go", + "snippet": "buy go"}]}, 200), + ]) + out = search("go docs", provider="perplexity", scope="code", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(http.seen[0][2]["search_domain_filter"], + list(CODE_SEARCH_DOMAINS)) + self.assertEqual([r["url"] for r in out["results"]], + ["https://go.dev/doc/"]) + + def test_brave_code_has_no_native_filter_but_is_post_filtered(self): + http = FakeHttp([ + ("api.search.brave.com", + {"web": {"results": [ + {"title": "B1", + "url": "https://stackoverflow.com/questions/1", + "description": "so"}, + {"title": "B2", "url": "https://blog.example.com/x", + "description": "blog"}]}}, 200), + ]) + out = search("q", provider="brave", scope="code", + env=dict(_FAKE_KEYS), http=http) + # No native domain parameter was invented for Brave. + self.assertNotIn("include_domains", http.seen[0][1]) + self.assertNotIn("site:", http.seen[0][1]) + self.assertEqual([r["url"] for r in out["results"]], + ["https://stackoverflow.com/questions/1"]) + + def test_all_filtered_out_continues_chain_then_fails_clearly(self): + http = FakeHttp([ + ("api.search.brave.com", + {"web": {"results": [ + {"title": "B", "url": "https://blog.example.com/x", + "description": "blog"}]}}, 200), + ]) + with self.assertRaises(WebSearchError) as ctx: + search("q", provider="brave", scope="code", + env=dict(_FAKE_KEYS), http=http) + self.assertIn("code-domain filter removed all 1 results", + str(ctx.exception)) + + def test_default_scope_unchanged_no_filter_no_post_filter(self): + http = FakeHttp([ + ("api.tavily.com", _tavily_body([ + "https://example.com/general", + "https://github.com/python/cpython", + ]), 200), + ]) + out = search("q", provider="tavily", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(out["scope"], "general") + self.assertNotIn("code_domains", out) + self.assertNotIn("include_domains", http.seen[0][2]) + self.assertEqual(len(out["results"]), 2) + + def test_unknown_scope_raises_before_http(self): + http = FakeHttp([("api.tavily.com", _tavily_body([]), 200)]) + with self.assertRaises(ValueError): + search("q", provider="tavily", scope="shopping", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(http.seen, []) + + def test_domain_matching_is_dot_boundary(self): + self.assertTrue(_is_code_domain("https://gist.github.com/u/1")) + self.assertTrue(_is_code_domain( + "https://www.stackoverflow.com/questions/1")) + self.assertFalse(_is_code_domain("https://evilgithub.com/x")) + self.assertFalse(_is_code_domain("https://github.com.evil.com/x")) + self.assertFalse(_is_code_domain("https://notstackoverflow.com/")) + + def test_missing_key_in_code_scope_reports_unconfigured(self): + http = FakeHttp([]) + with self.assertRaises(WebSearchError) as ctx: + search("q", provider="tavily", scope="code", + env={}, http=http) + self.assertIn("missing API key (TAVILY_API_KEY unset)", + str(ctx.exception)) + self.assertEqual(http.seen, []) + + +class ExtractChainTestCase(unittest.TestCase): + _PAGE = ("Direct" + "

direct page text

") + + def test_tavily_extract_success(self): + http = FakeHttp([ + ("api.tavily.com/extract", + {"results": [{"url": "https://docs.python.org/3/", + "raw_content": "extracted docs text"}], + "failed_results": []}, 200), + ]) + out = fetch_via_provider("https://docs.python.org/3/", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(out["source"], "tavily-extract") + self.assertEqual(out["text"], "extracted docs text") + self.assertEqual(http.seen[0][1], "https://api.tavily.com/extract") + + def test_tavily_fails_then_exa_contents_used(self): + http = FakeHttp([ + ("api.tavily.com/extract", {}, 500), + ("api.exa.ai/contents", + {"results": [{"url": "https://docs.python.org/3/", + "text": "exa page text"}]}, 200), + ]) + out = fetch_via_provider("https://docs.python.org/3/", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(out["source"], "exa-contents") + self.assertEqual(out["text"], "exa page text") + + def test_both_extracts_fail_then_direct_fetch_fallback(self): + http = FakeHttp([ + ("api.tavily.com/extract", {}, 500), + ("api.exa.ai/contents", {}, 500), + ("docs.python.org", self._PAGE, 200), + ]) + out = fetch_best("https://docs.python.org/3/", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(out["source"], "direct-fetch") + self.assertIn("direct page text", out["text"]) + self.assertEqual( + [a["provider"] for a in out["extract_attempted"]], + ["tavily", "exa"]) + + def test_both_fail_and_direct_fails_reports_both_no_fabrication(self): + http = FakeHttp([ + ("api.tavily.com/extract", {}, 500), + ("api.exa.ai/contents", {}, 500), + ("docs.python.org", {}, 403), + ]) + with self.assertRaises(WebSearchError) as ctx: + fetch_best("https://docs.python.org/3/", + env=dict(_FAKE_KEYS), http=http) + msg = str(ctx.exception) + self.assertIn("provider extraction failed", msg) + self.assertIn("direct fetch failed", msg) + for secret in _FAKE_KEYS.values(): + self.assertNotIn(secret, msg) + + def test_no_keys_extract_errors_but_best_still_direct_fetches(self): + with self.assertRaises(WebSearchError) as ctx: + fetch_via_provider("https://docs.python.org/3/", + env={}, http=FakeHttp([])) + self.assertIn("no extract provider configured", str(ctx.exception)) + http = FakeHttp([("docs.python.org", self._PAGE, 200)]) + out = fetch_best("https://docs.python.org/3/", + env={}, http=http) + self.assertEqual(out["source"], "direct-fetch") + + def test_localhost_blocked_before_any_provider_call(self): + http = FakeHttp([]) + with self.assertRaises(WebSearchError): + fetch_best("http://127.0.0.1/admin", + env=dict(_FAKE_KEYS), http=http) + with self.assertRaises(WebSearchError): + fetch_via_provider("http://10.0.0.5/admin", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(http.seen, []) + + +class McpScopeTestCase(unittest.TestCase): + def test_tools_list_exposes_scope_enum(self): + resp = handle_request( + {"jsonrpc": "2.0", "id": 1, "method": "tools/list"}) + tool = [t for t in resp["result"]["tools"] + if t["name"] == "web_search"][0] + scope = tool["inputSchema"]["properties"]["scope"] + self.assertEqual(scope["enum"], ["general", "code"]) + self.assertEqual(scope["default"], "general") + self.assertIn("shopping", tool["description"]) + fetch_tool = [t for t in resp["result"]["tools"] + if t["name"] == "web_fetch"][0] + self.assertIn("Tavily extract", fetch_tool["description"]) + + def test_tools_call_passes_scope_through(self): + seen = {} + + def fake_search(query, limit=5, provider="auto", scope="general"): + seen.update(query=query, scope=scope) + return {"provider": "fake", "results": [], "answer": None, + "attempted": [], "scope": scope} + + resp = handle_request( + {"jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "serde", + "scope": "code"}}}, + search_fn=fake_search) + self.assertFalse(resp["result"]["isError"]) + self.assertEqual(seen["scope"], "code") + + def test_web_fetch_default_is_extract_first_chain(self): + # Wiring assertion: the MCP default fetch is fetch_best, not + # the direct-only fetch_url. + self.assertIs(_mcp.fetch_best, _ws.fetch_best) + + seen = {} + + def fake_fetch(url, char_limit=15000): + seen.update(url=url, char_limit=char_limit) + return {"url": url, "text": "t", "source": "fake"} + + resp = handle_request( + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "web_fetch", + "arguments": {"url": "https://example.com/x"}}}, + fetch_fn=fake_fetch) + self.assertFalse(resp["result"]["isError"]) + self.assertEqual(seen["url"], "https://example.com/x") + + +class SeededScriptTestCase(unittest.TestCase): + def test_seeded_flat_copy_exposes_code_scope(self): + with tempfile.TemporaryDirectory() as tmp: + home = os.path.join(tmp, "codex-home") + os.makedirs(home) + out = ensure_codex_home_websearch( + home, command_args=["python3", "/tmp/ws/websearch_mcp.py"]) + self.assertTrue(out["config_changed"]) + # Run the seeded flat copy standalone (no repo cloud + # package on its path, exactly like a tenant codex-home). + probe = ( + "import json, websearch_mcp\n" + "resp = websearch_mcp.handle_request(\n" + " {'jsonrpc': '2.0', 'id': 1, 'method': 'tools/list'})\n" + "tool = [t for t in resp['result']['tools']\n" + " if t['name'] == 'web_search'][0]\n" + "print(json.dumps(tool['inputSchema']['properties']" + "['scope']))\n") + env = {"PATH": os.environ.get("PATH", ""), + "PYTHONPATH": home} + proc = subprocess.run( + [sys.executable, "-c", probe], cwd=home, env=env, + capture_output=True, text=True, timeout=60) + self.assertEqual(proc.returncode, 0, proc.stderr) + scope = json.loads(proc.stdout.strip().splitlines()[-1]) + self.assertEqual(scope["enum"], ["general", "code"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_websearch_mcp.py b/cloud/tests/test_websearch_mcp.py index 8af3b1e..36673de 100644 --- a/cloud/tests/test_websearch_mcp.py +++ b/cloud/tests/test_websearch_mcp.py @@ -203,9 +203,10 @@ class McpFramingTestCase(unittest.TestCase): names = [t["name"] for t in resp["result"]["tools"]] self.assertEqual(names, ["web_search", "web_fetch"]) - def fake_search(query, limit=5, provider="auto"): + def fake_search(query, limit=5, provider="auto", + scope="general"): return {"provider": "fake", "results": [], "answer": None, - "attempted": []} + "attempted": [], "scope": scope} resp = handle_request( {"jsonrpc": "2.0", "id": 3, "method": "tools/call", diff --git a/cloud/websearch.py b/cloud/websearch.py index 2adc38c..7d77a99 100644 --- a/cloud/websearch.py +++ b/cloud/websearch.py @@ -32,6 +32,16 @@ Security notes: pretend otherwise. It does refuse non-http(s) URLs and literal localhost / private-IP URLs, which stops the most trivial SSRF accidents, but it is not a sandbox. + +Code scope (``search(..., scope="code")``): +- Only CODE_SEARCH_DOMAINS results survive, via native provider + filters where they exist (Perplexity search_domain_filter, + Exa/Tavily include_domains) plus post-filtering for every provider. + Default scope="general" is unchanged. +- Fetch chain (``fetch_best``, the web_fetch tool path): Tavily + extract, then Exa contents, then direct ``fetch_url``. Direct + fetch 403ing under the egress whitelist is fail-closed and is + reported, never bypassed. """ import html as _html @@ -47,12 +57,15 @@ from html.parser import HTMLParser __all__ = [ "DEFAULT_CHAIN", + "CODE_SEARCH_DOMAINS", "WebSearchError", "SearchResult", "available_providers", "resolve_provider_chain", "search", "fetch_url", + "fetch_via_provider", + "fetch_best", "ensure_codex_home_websearch", "DEFAULT_FETCH_CHAR_LIMIT", ] @@ -60,6 +73,54 @@ __all__ = [ DEFAULT_CHAIN = ("perplexity", "exa", "tavily", "brave", "searxng", "duckduckgo") DEFAULT_FETCH_CHAR_LIMIT = 15000 + +# Programming-only search scope (``search(..., scope="code")``). +# Fixed allowlist, deliberately closed: a result is kept only when its +# host is one of these domains or a subdomain of one (dot-boundary +# match, so evilgithub.com never matches github.com). Release notes +# and changelogs are in scope ONLY when hosted on a listed domain +# (e.g. a project's GitHub releases page); there is no generic +# "project official domain" detector, and inventing one would silently +# reopen the scope to arbitrary sites. The list has 15 entries, under +# Perplexity's search_domain_filter cap of 20. +CODE_SEARCH_DOMAINS = ( + # Official language / platform docs. + "docs.python.org", "nodejs.org", "go.dev", "pkg.go.dev", + "docs.rs", "doc.rust-lang.org", "developer.mozilla.org", + # Code hosting (githubusercontent.com is GitHub's raw host). + "github.com", "githubusercontent.com", + # Package registries. + "pypi.org", "npmjs.com", "crates.io", + # Q&A and security advisories. + "stackoverflow.com", "nvd.nist.gov", "cve.org", +) + + +def _is_code_domain(url): + """True when url's host is in (or under) CODE_SEARCH_DOMAINS.""" + try: + host = (urllib.parse.urlparse(url).hostname or "").lower() + except Exception: + return False + host = host.rstrip(".") + for domain in CODE_SEARCH_DOMAINS: + if host == domain or host.endswith("." + domain): + return True + return False + + +def _normalize_scope(scope): + """Map scope to "general" (default) or "code"; reject anything else. + + Unknown scopes raise before any HTTP happens, so a typo can never + silently degrade to an unscoped search. + """ + if scope is None or str(scope).strip() in ("", "general", "all"): + return "general" + if str(scope).strip() == "code": + return "code" + raise ValueError("unknown search scope: %r (expected 'general' " + "or 'code')" % (scope,)) _USER_AGENT = "codex-websearch-mcp/0.1 (+codex-app-server)" _KEY_ENV = { @@ -175,16 +236,22 @@ def _clamp_limit(limit, default=5, lo=1, hi=10): # ------------------------------------------------------------ providers -def _provider_perplexity(query, limit, env, http): +def _provider_perplexity(query, limit, env, http, domains=None): # Sonar is a synthesis API: one OpenAI-shaped chat call returns a # written, cited answer. We surface both the answer and the # underlying sources, so web_search keeps its "answer + citations" # shape instead of degrading to a link list. + # Native domain filter: top-level search_domain_filter (Sonar chat + # completions; allowlist mode, max 20 domains -- CODE_SEARCH_DOMAINS + # has 15, so it fits whole). + payload = {"model": "sonar", + "messages": [{"role": "user", "content": query}]} + if domains: + payload["search_domain_filter"] = list(domains) _status, body, _ctype = http.post_json( "https://api.perplexity.ai/chat/completions", _default_headers(env, "perplexity"), - {"model": "sonar", - "messages": [{"role": "user", "content": query}]}) + payload) data = json.loads(body.decode("utf-8", errors="replace")) choices = data.get("choices") or [] message = (choices[0].get("message") or {}) if choices else {} @@ -202,13 +269,17 @@ def _provider_perplexity(query, limit, env, http): return results, answer -def _provider_exa(query, limit, env, http): +def _provider_exa(query, limit, env, http, domains=None): + payload = {"query": query, "numResults": limit, "type": "auto", + "contents": {"text": {"maxCharacters": 2000}, + "highlights": True}} + if domains: + # Native domain filter. + payload["include_domains"] = list(domains) _status, body, _ctype = http.post_json( "https://api.exa.ai/search", _default_headers(env, "exa"), - {"query": query, "numResults": limit, "type": "auto", - "contents": {"text": {"maxCharacters": 2000}, - "highlights": True}}) + payload) data = json.loads(body.decode("utf-8", errors="replace")) results = [] for item in data.get("results") or []: @@ -226,13 +297,17 @@ def _provider_exa(query, limit, env, http): return results, None -def _provider_tavily(query, limit, env, http): +def _provider_tavily(query, limit, env, http, domains=None): + payload = {"query": query, "search_depth": "basic", + "max_results": limit, "include_answer": False, + "include_raw_content": False} + if domains: + # Native domain filter (max 300 domains). + payload["include_domains"] = list(domains) _status, body, _ctype = http.post_json( "https://api.tavily.com/search", _default_headers(env, "tavily"), - {"query": query, "search_depth": "basic", - "max_results": limit, "include_answer": False, - "include_raw_content": False}) + payload) data = json.loads(body.decode("utf-8", errors="replace")) results = [] for item in data.get("results") or []: @@ -245,7 +320,12 @@ def _provider_tavily(query, limit, env, http): return results, data.get("answer") or None -def _provider_brave(query, limit, env, http): +def _provider_brave(query, limit, env, http, domains=None): + # No native include_domains on Brave's web search endpoint (its + # site: query operator is not used for a 15-domain allowlist: + # multi-domain site: semantics are unproven, and a wrong rewrite + # would silently change the query). Code-scope filtering for + # Brave happens after the call, in search(). url = ("https://api.search.brave.com/res/v1/web/search?" + urllib.parse.urlencode( {"q": query, "count": limit, "extra_snippets": "true"})) @@ -263,7 +343,10 @@ def _provider_brave(query, limit, env, http): return results, None -def _provider_searxng(query, limit, env, http): +def _provider_searxng(query, limit, env, http, domains=None): + # No native include_domains is passed: SearXNG's site: handling + # depends on the pinned engines and is not relied on. Code-scope + # filtering happens after the call, in search(). base = (env.get("SEARXNG_URL") or "").rstrip("/") if not base: raise WebSearchError("searxng not configured (SEARXNG_URL unset)") @@ -352,7 +435,9 @@ def _unwrap_ddg(href): return href -def _provider_duckduckgo(query, limit, env, http): +def _provider_duckduckgo(query, limit, env, http, domains=None): + # No native domain filter; code-scope filtering happens after + # the call, in search(). url = ("https://html.duckduckgo.com/html/?" + urllib.parse.urlencode({"q": query})) _status, body, _ctype = http.get( @@ -416,8 +501,21 @@ def resolve_provider_chain(provider="auto", env=None): return names -def search(query, limit=5, provider="auto", env=None, http=None): - """Search the web. Returns {provider, results, answer, attempted}. +def search(query, limit=5, provider="auto", env=None, http=None, + scope="general"): + """Search the web. Returns {provider, results, answer, attempted, + scope} (plus code_domains when scope="code"). + + ``scope="code"`` restricts results to CODE_SEARCH_DOMAINS: + providers with a native domain filter (Perplexity + search_domain_filter, Exa/Tavily include_domains) get the list in + the request, and every provider's results are post-filtered to + the same list afterwards, so backends without a native filter + (Brave, SearXNG, DuckDuckGo) are scoped too. A provider whose + results are all filtered out counts as "no results" and the chain + continues; the filter count is recorded on its attempted entry. + Default scope="general" behaves exactly as before (no domain + filter, no post-filtering). Raises WebSearchError (with .attempted) when every provider in the chain fails or returns no results. Error strings never contain @@ -425,6 +523,8 @@ def search(query, limit=5, provider="auto", env=None, http=None): """ if not query or not str(query).strip(): raise ValueError("query must be a non-empty string") + scope = _normalize_scope(scope) + domains = list(CODE_SEARCH_DOMAINS) if scope == "code" else None env = env if env is not None else os.environ http = http or HttpClient( timeout=int(env.get("WEBSEARCH_TIMEOUT", "20") or 20)) @@ -439,7 +539,7 @@ def search(query, limit=5, provider="auto", env=None, http=None): continue try: results, answer = _PROVIDERS[name]( - str(query), limit, env, http) + str(query), limit, env, http, domains=domains) except Exception as exc: # noqa: BLE001 - chain must continue attempted.append({"provider": name, "ok": False, "error": _http_error_message(exc)}) @@ -448,11 +548,29 @@ def search(query, limit=5, provider="auto", env=None, http=None): attempted.append({"provider": name, "ok": False, "error": "no results"}) continue - attempted.append({"provider": name, "ok": True, "error": None}) - return {"provider": name, - "results": [dict(r) for r in results[:limit]], - "answer": answer, - "attempted": attempted} + if scope == "code": + kept = [r for r in results if _is_code_domain(r.get("url"))] + if not kept: + attempted.append( + {"provider": name, "ok": False, + "error": "code-domain filter removed all %d results" + % len(results)}) + continue + entry = {"provider": name, "ok": True, "error": None} + if len(kept) < len(results): + entry["domain_filtered_out"] = len(results) - len(kept) + attempted.append(entry) + results = kept + else: + attempted.append({"provider": name, "ok": True, "error": None}) + out = {"provider": name, + "results": [dict(r) for r in results[:limit]], + "answer": answer, + "attempted": attempted, + "scope": scope} + if scope == "code": + out["code_domains"] = domains + return out raise WebSearchError( "all search providers failed: %s" % "; ".join("%s (%s)" % (a["provider"], a["error"]) @@ -578,6 +696,159 @@ def fetch_url(url, char_limit=DEFAULT_FETCH_CHAR_LIMIT, env=None, "char_limit": char_limit} +# ------------------------------------------------- provider extract + +def _extract_tavily(url, char_limit, env, http): + """Tavily extract: POST /extract, raw_content is the page text.""" + _status, body, _ctype = http.post_json( + "https://api.tavily.com/extract", + _default_headers(env, "tavily"), + {"urls": [url], "extract_depth": "basic", "format": "text"}) + data = json.loads(body.decode("utf-8", errors="replace")) + items = data.get("results") or [] + for item in items: + if isinstance(item, dict): + text = item.get("raw_content") or item.get("content") or "" + if text: + return text + failed = data.get("failed_results") or [] + if failed: + raise WebSearchError("tavily extract failed for url") + raise WebSearchError("tavily extract returned no content") + + +def _extract_exa(url, char_limit, env, http): + """Exa contents: POST /contents with ids=[url], full text view.""" + # Our own single-page budget, not a claimed API cap: the fetch + # truncation below still applies, this just keeps the request + # from asking for an unbounded page. + budget = min(char_limit, 100000) + _status, body, _ctype = http.post_json( + "https://api.exa.ai/contents", + _default_headers(env, "exa"), + {"ids": [url], "text": {"maxCharacters": budget}}) + data = json.loads(body.decode("utf-8", errors="replace")) + items = data.get("results") or data.get("contents") or [] + for item in items: + if not isinstance(item, dict): + continue + text = item.get("text") or "" + if not text and item.get("highlights"): + text = " ".join(str(h) for h in item["highlights"][:3]) + if text: + return text + raise WebSearchError("exa contents returned no content") + + +def _fetch_result(url, text, char_limit, source): + text, truncated = _truncate(text, char_limit) + return {"url": str(url), "title": "", "text": text, + "links": [], "truncated": truncated, + "char_limit": char_limit, "source": source} + + +def fetch_via_provider(url, char_limit=DEFAULT_FETCH_CHAR_LIMIT, + env=None, http=None): + """Extract one URL through a search provider's own extraction + endpoint, never by fetching the target directly. + + Order: Tavily extract, then Exa contents; only backends with a + configured key are tried. Raises WebSearchError (with .attempted) + when no extract provider is configured or every tried one fails. + No result is fabricated: an empty extraction is a failure, not + an empty page. The localhost / private-IP refusal of fetch_url + applies here too, before any provider call. + """ + if not url or not str(url).strip(): + raise ValueError("url must be a non-empty string") + env = env if env is not None else os.environ + if env.get("WEBSEARCH_ALLOW_PRIVATE_URLS") != "1": + blocked = _is_blocked_fetch_url(str(url)) + if blocked: + raise WebSearchError(blocked) + http = http or HttpClient( + timeout=int(env.get("WEBSEARCH_TIMEOUT", "20") or 20)) + try: + char_limit = int(char_limit) + except (TypeError, ValueError): + char_limit = DEFAULT_FETCH_CHAR_LIMIT + char_limit = max(2000, min(500000, char_limit)) + attempted = [] + extractors = (("tavily", _extract_tavily), + ("exa", _extract_exa)) + tried = False + for name, fn in extractors: + if name in _KEY_ENV and not env.get(_KEY_ENV[name]): + attempted.append({"provider": name, "ok": False, + "error": "missing API key (%s unset)" + % _KEY_ENV[name]}) + continue + tried = True + try: + text = fn(str(url), char_limit, env, http) + except Exception as exc: # noqa: BLE001 - try next extractor + attempted.append({"provider": name, "ok": False, + "error": _http_error_message(exc)}) + continue + attempted.append({"provider": name, "ok": True, "error": None}) + out = _fetch_result(url, text, char_limit, + "tavily-extract" if name == "tavily" + else "exa-contents") + out["attempted"] = attempted + return out + if not tried: + raise WebSearchError( + "no extract provider configured (set TAVILY_API_KEY " + "and/or EXA_API_KEY); direct fetch is the fallback, " + "see fetch_best", + attempted=attempted) + raise WebSearchError( + "all extract providers failed: %s" + % "; ".join("%s (%s)" % (a["provider"], a["error"]) + for a in attempted), + attempted=attempted) + + +def fetch_best(url, char_limit=DEFAULT_FETCH_CHAR_LIMIT, env=None, + http=None): + """Fetch one URL: provider extraction first, direct fetch last. + + This is what the web_fetch MCP tool calls. When no extract + provider is configured it goes straight to direct fetch (same + behaviour as the old web_fetch); when extraction fails it falls + back to direct fetch; when both fail it raises one error naming + both failures, never a fabricated page. Direct fetch under the + egress proxy 403s for target domains not on the Tinyproxy + whitelist -- that is fail-closed, not a reason to bypass the + proxy, and the 403 surfaces in the error. The localhost / + private-IP refusal is checked before anything else. + """ + if not url or not str(url).strip(): + raise ValueError("url must be a non-empty string") + env = env if env is not None else os.environ + if env.get("WEBSEARCH_ALLOW_PRIVATE_URLS") != "1": + blocked = _is_blocked_fetch_url(str(url)) + if blocked: + raise WebSearchError(blocked) + extract_attempted = [] + try: + return fetch_via_provider(url, char_limit=char_limit, + env=env, http=http) + except WebSearchError as exc: + extract_attempted = list(exc.attempted or []) + extract_error = str(exc) + try: + out = fetch_url(url, char_limit=char_limit, env=env, http=http) + except WebSearchError as exc: + raise WebSearchError( + "provider extraction failed (%s) and direct fetch failed " + "(%s)" % (extract_error, exc)) + out = dict(out) + out["source"] = "direct-fetch" + out["extract_attempted"] = extract_attempted + return out + + # ------------------------------------------------------- codex-home glue def ensure_codex_home_websearch(codex_home_dir, command_args=None, diff --git a/cloud/websearch_mcp.py b/cloud/websearch_mcp.py index 33d395f..70c11a6 100644 --- a/cloud/websearch_mcp.py +++ b/cloud/websearch_mcp.py @@ -20,10 +20,10 @@ import os import sys try: - from cloud.websearch import WebSearchError, fetch_url, search + from cloud.websearch import WebSearchError, fetch_best, search except ImportError: # flat copy in a tenant codex-home sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - from websearch import WebSearchError, fetch_url, search # noqa: E402 + from websearch import WebSearchError, fetch_best, search # noqa: E402 SERVER_NAME = "websearch" SERVER_VERSION = "0.1.0" @@ -37,7 +37,14 @@ _TOOLS = [ "supplies one. Use web_fetch to read specific pages. " "provider may be 'auto' (walk Perplexity, Exa, Tavily, " "Brave, then keyless SearXNG / DuckDuckGo, falling back " - "on failure) or a single provider name to pin it." + "on failure) or a single provider name to pin it. " + "scope 'code' restricts results to programming sources " + "only (official language docs, GitHub, package " + "registries, Stack Overflow, CVE/NVD); anything else " + "is filtered out, never returned. Default scope " + "'general' is unfiltered. This tool does not do " + "shopping, social, places, travel, general news, " + "weather, finance, or browser checkout." ), "inputSchema": { "type": "object", @@ -48,6 +55,12 @@ _TOOLS = [ "description": "Max results, 1-10"}, "provider": {"type": "string", "default": "auto", "description": "auto or a provider name"}, + "scope": {"type": "string", "default": "general", + "enum": ["general", "code"], + "description": "general: unfiltered web " + "search. code: programming " + "sources only, see tool " + "description."}, }, "required": ["query"], }, @@ -57,7 +70,13 @@ _TOOLS = [ "description": ( "Fetch one URL and return clean text plus its links, " "truncated head+tail when over char_limit. " - "Only http(s) URLs; localhost / private IPs are blocked." + "Extraction first: Tavily extract, then Exa contents " + "(only backends with a configured key are tried), " + "direct fetch last. Under the egress proxy, direct " + "fetch of a domain outside the Tinyproxy whitelist " + "fails 403 fail-closed; that error is reported, not " + "bypassed. Only http(s) URLs; localhost / private IPs " + "are blocked. This is not a sandbox boundary." ), "inputSchema": { "type": "object", @@ -83,7 +102,7 @@ def handle_request(request, search_fn=None, fetch_fn=None): None for notifications (no id). Pure function for testability: the stdio loop passes the real search/fetch, tests pass fakes.""" search_fn = search_fn or search - fetch_fn = fetch_fn or fetch_url + fetch_fn = fetch_fn or fetch_best if not isinstance(request, dict): return {"jsonrpc": "2.0", "id": None, "error": _err(-32700, "request is not an object")} @@ -123,7 +142,8 @@ def handle_request(request, search_fn=None, fetch_fn=None): result = search_fn( args.get("query"), limit=args.get("limit", 5), - provider=args.get("provider", "auto")) + provider=args.get("provider", "auto"), + scope=args.get("scope", "general")) elif name == "web_fetch": result = fetch_fn( args.get("url"), diff --git a/docs/web-search-mcp.md b/docs/web-search-mcp.md index 085b31d..02121db 100644 --- a/docs/web-search-mcp.md +++ b/docs/web-search-mcp.md @@ -15,17 +15,23 @@ pi 的工具），只复用它的模式：统一工具 + provider 抽象 + stdout 是 JSON-RPC 流，绝不能混日志。 - `cloud/tests/test_websearch_mcp.py`：12 个单测，全部 mock HTTP，不联网、不需要真实 key。 +- `cloud/tests/test_code_search.py`：编程作用域与抽取回退链 + 的单测，同样全部 mock HTTP、不联网、不需要真实 key。 ## 工具 -- `web_search(query, limit=5, provider="auto")`：返回 - `{provider, results:[{title, url, snippet, score?}], answer?, attempted}`。 +- `web_search(query, limit=5, provider="auto", scope="general")`：返回 + `{provider, results:[{title, url, snippet, score?}], answer?, attempted, scope}`， + `scope="code"` 时另带 `code_domains`。 `answer` 只有 provider 本身给合成答案时才有（目前是 Perplexity Sonar）；其它 provider 的总结仍由主模型做。 -- `web_fetch(url, char_limit=15000)`：返回 - `{url, title, text, links, truncated}`。超长按头 75% + - 尾 25% 截断并标 `[TRUNCATED]`；只接受 http(s)，拒绝 - localhost 与字面私有 IP。 + `scope` 详见下面“编程搜索 scope=code”一节。 +- `web_fetch(url, char_limit=15000)`：走 `fetch_best` 抽取回退链 + （Tavily extract → Exa contents → 直接抓取），返回 + `{url, title, text, links, truncated, source}`（`source` 为 + `tavily-extract` / `exa-contents` / `direct-fetch`）。 + 超长按头 75% + 尾 25% 截断并标 `[TRUNCATED]`；只接受 http(s)，拒绝 + localhost 与字面私有 IP；抽取与直接抓取都失败时报错，不伪造页面。 ## Provider 链（auto 顺序） @@ -101,6 +107,111 @@ Docker 租户必须传容器内路径 字段与结果映射，未用真实 key 跑过；上生产前须用真实 key 在测试项目跑一遍，再谈默认链顺序是否合理。 +## 编程搜索 scope=code（v2 新增，只做编程任务搜索） + +选型：`web_search` 加 `scope="code"` 参数，不另立 +`code_search` 工具。理由：同一条 provider 链、同一套 +回退与 key 规则，只多一层域名限定；另立工具会复制 +schema、handler 与播种文档两套，且调用方要在两个 +工具间二选一。代码（`cloud/websearch.py` 的 +`CODE_SEARCH_DOMAINS` / `search(..., scope=...)`）、 +MCP 工具描述（`cloud/websearch_mcp.py` 的 +`web_search` 描述与 `scope` enum）与本文三处一致， +只有 `general` / `code` 两个值，未知 scope 在发任何 +HTTP 之前就报 `ValueError`，绝不静默退化成无限定搜索。 + +### 允许域清单（实际落代码的 15 个，含子域名） + +- 官方文档：`docs.python.org`、`nodejs.org`、`go.dev`、 + `pkg.go.dev`、`docs.rs`、`doc.rust-lang.org`、 + `developer.mozilla.org` +- 代码托管：`github.com`、`githubusercontent.com` + （GitHub 自己的 raw host，blob 页之外原始文件所用） +- 包仓库：`pypi.org`、`npmjs.com`、`crates.io` +- 问答与安全：`stackoverflow.com`、`nvd.nist.gov`、`cve.org` + +匹配规则是域名本身或其子域名（点边界匹配： +`gist.github.com` 算、`evilgithub.com` 不算）。 +changelog / release notes 只有落在上述域名时才在 +范围内（例如项目在 GitHub 的 Releases 页）；没有 +“项目官方域名自动识别”——不在清单里的项目官网 +（如某框架自己的 docs 站）本版不返回，这是明确的 +缺口，不是泛化成任意站的口子。 + +### 不在范围清单（scope=code 下一律过滤掉） + +通用新闻、购物、社交、地点、机票酒店、天气、行情、 +视频/图片站、博客聚合与其它任意站。`scope="code"` +不是“优先编程来源”，是硬过滤：非清单域的结果不 +返回、不降级展示，只在 `attempted` 里记 +`domain_filtered_out` 数量。 + +### 各后端如何限定（先核实请求字段再写代码） + +- 原生支持，直接传：Tavily `include_domains`、 + Exa `include_domains`、Perplexity Sonar chat + completions 的顶层 `search_domain_filter`（上限 + 20 个域，本清单 15 个，整单可传）。 +- 原生不支持，不编造参数：Brave 网页搜索端点没有 + `include_domains` 参数（多域 `site:` 改写语义未经 + 证实，不改写查询），SearXNG / DuckDuckGo 同样不传； + 这三家只靠 `search()` 返回后的统一事后过滤。 +- 双保险：原生限定的后端返回后同样再过一次事后 + 过滤，所以各后端在 code 作用域下的语义一致。 +- 链语义：某后端结果被全部滤掉时，该后端记为 + “no results”（`code-domain filter removed all N + results`），链继续走下一个后端；全部后端都空才 + 报错。 + +### 抓取回退链与 403 语义（web_fetch） + +`web_fetch` 实际调 `fetch_best`，顺序固定： + +1. Tavily extract（`POST https://api.tavily.com/extract`， + `urls:[url]`，取 `results[0].raw_content`；空结果或 + `failed_results` 命中都算失败）——只在配了 + `TAVILY_API_KEY` 时试。 +2. Exa contents（`POST https://api.exa.ai/contents`， + `ids:[url]` + `text.maxCharacters`，取 `results[0].text`， + 无 text 时取 highlights 拼接；空算失败）——只在 + 配了 `EXA_API_KEY` 时试。 +3. 直接抓取（旧 `fetch_url` 的 HTML 清洗与截断）。 + +抽取与直接都失败时，错误同时写明两边失败，不伪造 +页面。无任何抽取 key 时 `fetch_via_provider` 单独调会 +明确报“no extract provider configured”，但 `fetch_best` +会直接走直接抓取（与旧 web_fetch 行为一致），不是 +报错。localhost / 私有 IP 在抽取之前就拒绝，不会经 +抽取商去探内网。 + +403 语义：抽取走的是搜索商 API 域名（`tavily.com`、 +`exa.ai`，已在 Tinyproxy 白名单里），不直接碰目标站， +所以文档站本身不在白名单时抽取仍可通；只有回退到 +直接抓取时，目标域不在白名单才会被代理 403 挡住。 +这是 fail-closed：报错、不静默绕过、不承诺“任意 +文档站都能直接抓”。本轨**没有改** +`deploy/egress/tinyproxy-filter`（也不改主机上的文件）： +若将来主机侧要让直接抓取也通，需另行走主机变更， +filter 行须用 POSIX BASIC 正则（`^\(.*\.\)*域名$`）、 +文件 644，extended 锚定写法会把允许域也挡成 403 +（见部署教训）；该变更不在本轨授权内。 + +### 安全与隔离（不变） + +- key 只走 MCP 进程环境变量，绝不写进 config.toml、 + 日志、测试、文档；本轨测试用的 key 全是 + `*-SECRET-*` 假值。 +- worker 开关默认关；租户播种幂等、已有 section + 不重写；Docker 环境变量走 extra_env + + `[mcp_servers.websearch.env]` 路径。 +- `web_fetch` 仍只接受 http(s)、拒 localhost 与私有 + IP，且它不是沙箱边界（danger-full-access 任务本 + 来就能 curl）。 +- 老租户注意：播种助手从不覆盖内容已变的租户脚本 + （既有防覆盖规则），所以 scope 只对新播种的租户 + 立即生效；老租户拿到新脚本需另行决策，不在本轨 + 改动该规则。 + ## 诚实边界 - 本批只做代码 + 单测 + 测试租户目录的配置合并验证， @@ -115,3 +226,9 @@ Docker 租户必须传容器内路径 - `web_fetch` 不是安全边界：danger-full-access 的任务 本来就能 curl，fetch 只挡了最明显的非 http(s) 与 本地私有地址，不应把它当沙箱用。 +- 编程搜索本轨未做（见回执）：真实 key 的 live 测试、 + 主机 unit / filter / env 接线、GitHub API token + （本设计走网页搜索后端限定域名，不调 GitHub API， + 所以不需要 token；真要 GitHub 代码内容搜索 API 是 + 另一功能，不在本轨）、清单外项目官网域、老租户 + 脚本升级。