From 0b9f3f60d61aed0a4e7753e6ba8f62f8c8a741c2 Mon Sep 17 00:00:00 2001 From: George Pickett <297992784+georgeatparallel@users.noreply.github.com> Date: Sat, 17 Oct 2026 21:32:01 +0000 Subject: [PATCH] Add optional anonymous Parallel Search MCP tools --- CHANGELOG.md | 6 + README.md | 21 +++ core/iteration_budget.py | 1 + core/mcp.py | 37 +++++- tests/test_parallel_search.py | 241 ++++++++++++++++++++++++++++++++++ tools/parallel_search.py | 94 +++++++++++++ tools/registry.py | 24 +++- 7 files changed, 422 insertions(+), 2 deletions(-) create mode 100644 tests/test_parallel_search.py create mode 100644 tools/parallel_search.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d37b94a..85ce3b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] — 3.1.x — Competitiveness pass +### Added — Optional Parallel web research +- `parallel_search` and `parallel_fetch` provide anonymous web search and source + page excerpts through Parallel Search MCP, using the existing HTTP client. + DuckDuckGo remains available. Both tools follow web-toolset disable controls + and normal tool approvals, with no additional dependency or API key. + ### Added — Turn-completion notifications (harvested from Hermes) - **`core/notify.py`**: rings the terminal bell (`\a`, propagates over SSH) and optionally raises a native desktop notification (macOS `osascript`, Linux diff --git a/README.md b/README.md index 0e521cf..31c1d00 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,27 @@ python -m core.bootstrap --browser # just the Chromium binary python -m core.bootstrap --check # status only ``` +## Optional Parallel web research + +Ask Operon to use `parallel_search` for focused web excerpts with source URLs, +and `parallel_fetch` to read a specific page when those excerpts are insufficient. +[Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp) +provides these tools without an account, API key, or additional package. +DuckDuckGo and `web_scrape` remain available; Parallel is an explicit alternative. + +For example: `Use parallel_search to find Python 3.13 release highlights, then +summarize the excerpts with source links.` Search accepts an `objective` and +`search_queries` (a list of related keyword queries). Fetch accepts `urls` (up to +20 HTTP(S) URLs), an optional `objective` of up to 200 characters, and +`full_content=true` when the entire page markdown is needed. + +Queries and fetched URLs are sent to `https://search.parallel.ai/mcp`. Anonymous +usage has service limits; errors are returned without switching providers. +Requests time out after 15 seconds for setup/discovery and 30 seconds for the +tool call. Ctrl-C interrupts the call. `/toolsets disable web` disables both +Parallel tools alongside the other web tools; `/toolsets enable web` restores +them. Normal `/approve on` prompts and delegated-agent tool restrictions apply. + ## Updating ```bash diff --git a/core/iteration_budget.py b/core/iteration_budget.py index 0c5fe63..60d561e 100644 --- a/core/iteration_budget.py +++ b/core/iteration_budget.py @@ -19,6 +19,7 @@ REFUNDABLE_TOOLS: FrozenSet[str] = frozenset({ "file_read", "file_exists", "file_info", "dir_list", "file_search", "duckduckgo_search", "web_scrape", "web_fetch", + "parallel_search", "parallel_fetch", "browser_get_url", "browser_snapshot", "db_query", "knowledge_get", "knowledge_list", "git_status", "git_diff", "git_log", diff --git a/core/mcp.py b/core/mcp.py index e910f34..8d0f9a5 100644 --- a/core/mcp.py +++ b/core/mcp.py @@ -142,14 +142,49 @@ class _HttpTransport: Falls back to classic JSON-RPC POST if the server doesn't respond to /mcp. """ - def __init__(self, url: str, headers: dict = None): + def __init__(self, url: str, headers: dict = None, *, exact_url: bool = False, + auth=None): if not _HTTP_OK: raise ImportError("requests library required for HTTP MCP transport.") self._base = url.rstrip("/") self._headers = {"Content-Type": "application/json", **(headers or {})} + self._exact_url = exact_url + self._auth = auth + + def _post_exact(self, message: dict, timeout: float) -> dict: + """Use a complete Streamable HTTP endpoint without endpoint retries.""" + try: + with _requests.post(self._base, json=message, headers=self._headers, + timeout=timeout, stream=True, auth=self._auth) as resp: + resp.raise_for_status() + if "id" not in message: + return {} + if "text/event-stream" in resp.headers.get("Content-Type", ""): + resp.encoding = "utf-8" + data = [] + for line in resp.iter_lines(decode_unicode=True): + if line.startswith("data:"): + data.append(line[5:].lstrip(" ")) + elif not line and data: + event = json.loads("\n".join(data)) + data = [] + if isinstance(event, dict) and event.get("id") == message["id"]: + return event + raise ValueError("MCP stream ended without a matching response") + result = resp.json() + if not isinstance(result, dict) or result.get("id") != message["id"]: + raise ValueError("MCP response ID does not match request") + return result + except Exception as e: + return {"error": {"code": -1, "message": str(e)}} + + def send_notification(self, method: str, timeout: float = 15.0) -> dict: + return self._post_exact({"jsonrpc": "2.0", "method": method}, timeout) def send_request(self, method: str, params: Any = None, timeout: float = 15.0) -> dict: req = _make_request(method, params) + if self._exact_url: + return self._post_exact(req, timeout) # Try /mcp endpoint first (Streamable HTTP), fall back to base URL for endpoint in (f"{self._base}/mcp", self._base): try: diff --git a/tests/test_parallel_search.py b/tests/test_parallel_search.py new file mode 100644 index 0000000..1b75232 --- /dev/null +++ b/tests/test_parallel_search.py @@ -0,0 +1,241 @@ +"""Real registry, HTTP MCP transport and scripted agent-loop coverage (no live LLM).""" + +import json +from unittest.mock import MagicMock + +import pytest +import requests + +from core import mcp +from core.iteration_budget import IterationBudget +from core.version import __version__ +from tools import registry +from tools.parallel_search import _ENDPOINT + + +SOURCE = "https://www.python.org/downloads/release/python-3130/" +PAYLOAD = {"results": [{"url": SOURCE, "title": "Python 3.13.0", + "excerpts": ["Python 3.13 introduces a new interactive interpreter."]}]} +SEARCH = {"objective": "Find Python 3.13 highlights", "search_queries": ["Python 3.13 release"]} + + +def response(body, content_type="application/json", status=200): + resp = requests.Response() + resp.status_code = status + resp.headers["Content-Type"] = content_type + resp._content = body.encode("utf-8") + resp._content_consumed = True + return resp + + +@pytest.fixture +def wire(monkeypatch): + calls = [] + monkeypatch.delenv("PARALLEL_API_KEY", raising=False) + monkeypatch.delenv("PARALLEL_SEARCH_API_KEY", raising=False) + + def post(url, **kwargs): + calls.append((url, kwargs)) + message = kwargs["json"] + if "id" not in message: + return response("", status=202) + method = message["method"] + if method == "initialize": + result = {"protocolVersion": "2024-11-05", "capabilities": {}} + elif method == "tools/list": + result = {"tools": [{"name": "web_search"}, {"name": "web_fetch"}]} + else: + assert method == "tools/call" + result = {"content": [{"type": "text", "text": json.dumps(PAYLOAD)}], + "structuredContent": PAYLOAD} + return response(json.dumps({"jsonrpc": "2.0", "id": message["id"], "result": result})) + + monkeypatch.setattr(mcp._requests, "post", post) + return calls + + +def test_registry_search_fetch_no_key_and_preserved_choices(wire, monkeypatch, tmp_path): + # The optional tools never load, overwrite or auto-connect saved MCP servers. + saved = tmp_path / "mcp_servers.json" + saved.write_text('{"chosen": {"transport": "http", "url": "https://example.org/mcp"}}') + before = saved.read_bytes() + monkeypatch.setattr(mcp, "MCP_CONFIG_PATH", saved) + monkeypatch.setenv("PARALLEL_API_KEY", "unrelated-configured-key") + ddg = registry._DISPATCH["duckduckgo_search"] + reg = registry.ToolRegistry() + assert reg.execute("parallel_search", SEARCH) == {"success": True, "output": PAYLOAD, "error": ""} + fetched = reg.execute("parallel_fetch", {"urls": [SOURCE], "objective": "Release highlights"}) + assert fetched["success"] and fetched["output"]["results"][0]["url"] == SOURCE + assert saved.read_bytes() == before + assert registry._DISPATCH["duckduckgo_search"] is ddg + assert "parallel_search" in reg.get_descriptions() + assert "parallel_fetch" in reg.get_compact_descriptions() + assert [c[1]["json"]["method"] for c in wire] == [ + "initialize", "notifications/initialized", "tools/list", "tools/call"] * 2 + for url, kwargs in wire: + assert url == _ENDPOINT + assert kwargs["headers"]["User-Agent"] == f"Operon/{__version__} (+https://github.com/OperonAgent/Operon)" + assert kwargs["headers"]["Accept"] == "application/json, text/event-stream" + assert "Authorization" not in kwargs["headers"] + assert kwargs["timeout"] == (30.0 if kwargs["json"]["method"] == "tools/call" else 15.0) + prepared = requests.Request("POST", url, auth=kwargs["auth"], + headers={"Authorization": "unrelated"}).prepare() + assert "Authorization" not in prepared.headers + arguments = [kw["json"]["params"]["arguments"] for _, kw in wire + if kw["json"]["method"] == "tools/call"] + assert arguments[0]["search_queries"] == SEARCH["search_queries"] + assert arguments[1]["urls"] == [SOURCE] + assert arguments[0]["session_id"] == arguments[1]["session_id"] + assert len(arguments[0]["session_id"]) == 32 + + +def test_web_toolset_disable_enable_and_read_only_budget(wire): + reg = registry.ToolRegistry() + try: + registry.disable_toolset("web") + for name in ("parallel_search", "parallel_fetch"): + assert not reg.execute(name, SEARCH)["success"] + assert not wire + finally: + registry.enable_toolset("web") + assert reg.execute("parallel_search", SEARCH)["success"] + budget = IterationBudget(1) + assert budget.consume() + assert budget.refund_if_read_only("parallel_search") + assert budget.refund_if_read_only("parallel_fetch") + + +@pytest.mark.parametrize("name,params", [ + ("parallel_search", {"objective": "", "search_queries": ["query"]}), + ("parallel_search", {"objective": "find", "search_queries": "query"}), + ("parallel_search", {"objective": "find", "search_queries": []}), + ("parallel_fetch", {"urls": []}), + ("parallel_fetch", {"urls": ["file:///etc/passwd"]}), + ("parallel_fetch", {"urls": ["https://user:secret@example.org"]}), + ("parallel_fetch", {"urls": [SOURCE] * 21}), + ("parallel_fetch", {"urls": [SOURCE], "objective": "x" * 201}), + ("parallel_fetch", {"urls": [SOURCE], "full_content": "true"}), +]) +def test_invalid_input_never_requests(wire, name, params): + assert not registry.ToolRegistry().execute(name, params)["success"] + assert not wire + + +@pytest.mark.parametrize("failure", ["timeout", "http", "rpc", "tool", "malformed", "id", "discovery"]) +def test_failure_returns_error_without_retry_or_provider_fallback(wire, monkeypatch, failure): + good_post = mcp._requests.post + + def post(url, **kwargs): + method = kwargs["json"]["method"] + target = "tools/list" if failure == "discovery" else "tools/call" + if method != target: + return good_post(url, **kwargs) + wire.append((url, kwargs)) + if failure == "timeout": + raise requests.Timeout("request timed out") + if failure == "http": + return response("limited", status=429) + if failure == "malformed": + return response("not json") + reply = {"jsonrpc": "2.0", "id": kwargs["json"]["id"]} + if failure == "rpc": + reply["error"] = {"code": -32602, "message": "invalid arguments"} + else: + reply["result"] = {"isError": True, "content": [{"type": "text", "text": "service error"}]} + if failure == "id": + reply["id"] = "wrong" + return response(json.dumps(reply)) + + monkeypatch.setattr(mcp._requests, "post", post) + result = registry.ToolRegistry().execute("parallel_search", SEARCH) + assert result["success"] is False and result["error"] + assert len(wire) == (3 if failure == "discovery" else 4) + assert all(url == _ENDPOINT for url, _ in wire) + + +def test_streamable_sse_utf8_and_text_only_result(wire, monkeypatch): + good_post = mcp._requests.post + + def post(url, **kwargs): + if kwargs["json"]["method"] != "tools/call": + return good_post(url, **kwargs) + wire.append((url, kwargs)) + result = {"content": [{"type": "text", "text": "Résumé " + SOURCE}]} + event = {"jsonrpc": "2.0", "id": kwargs["json"]["id"], "result": result} + return response(': ping\n\nevent: message\ndata: ' + json.dumps(event, ensure_ascii=False) + '\n\n', + "text/event-stream") + + monkeypatch.setattr(mcp._requests, "post", post) + result = registry.ToolRegistry().execute("parallel_fetch", {"urls": [SOURCE], "full_content": True}) + assert result["success"] and result["output"] == "Résumé " + SOURCE + assert wire[-1][1]["json"]["params"]["arguments"]["full_content"] is True + + +@pytest.mark.parametrize("stage", ["initialize", "tools/list", "tools/call"]) +def test_keyboard_interrupt_propagates_without_retry(wire, monkeypatch, stage): + good_post = mcp._requests.post + def interrupted(url, **kwargs): + if kwargs["json"]["method"] == stage: + raise KeyboardInterrupt() + return good_post(url, **kwargs) + monkeypatch.setattr(mcp._requests, "post", interrupted) + with pytest.raises(KeyboardInterrupt): + registry.ToolRegistry().execute("parallel_search", SEARCH) + + +def test_explicit_duckduckgo_selection_does_not_use_parallel(wire, monkeypatch): + selected = MagicMock(return_value={"success": True, "results": [], "error": ""}) + monkeypatch.setitem(registry._DISPATCH, "duckduckgo_search", selected) + assert registry.ToolRegistry().execute("duckduckgo_search", {"query": "Python"})["success"] + selected.assert_called_once_with(query="Python") + assert not wire + + +def run_scripted_loop(monkeypatch, approval=False, denied=False): + import main + from core.session import SessionManager + + session = SessionManager() + session.add_message("user", "Use Parallel to find Python 3.13 highlights and read the release page.") + config = MagicMock() + values = {"max_tool_iters": 4, "streaming_enabled": False, "memory_enabled": False} + config.get.side_effect = lambda key, default=None: values.get(key, default) + config.resolve_model.return_value = {"provider": "openai"} + router = MagicMock() + actions = [ + {"action": {"type": "tool", "tool_name": "parallel_search", "params": SEARCH}}, + {"action": {"type": "tool", "tool_name": "parallel_fetch", "params": {"urls": [SOURCE]}}}, + {"action": {"type": "response", "content": "Python 3.13 introduces a new interactive interpreter. Source: " + SOURCE}}, + ] + router.complete.side_effect = [json.dumps(a) for a in actions] + # Use the production parser; only model inference and unrelated background work are stubbed. + router.parse_response.side_effect = main.ModelRouter.parse_response + monkeypatch.setattr(main, "ThinkingSpinner", MagicMock()) + monkeypatch.setattr(main, "_APPROVAL_MODE", approval) + monkeypatch.setattr(main, "_VECTOR_MEMORY_AVAILABLE", False) + monkeypatch.setattr(main, "_OBSIDIAN_AVAILABLE", False) + monkeypatch.setattr(main, "_SKILL_SYNTH_AVAILABLE", False) + monkeypatch.setattr(main, "_retry_mgr", None) + monkeypatch.setattr(main, "log_tool_call", lambda *args: None) + monkeypatch.setattr("builtins.input", lambda: "n" if denied else "y") + monkeypatch.setattr("core.background_review.BackgroundReviewer.maybe_spawn", lambda *args, **kwargs: None) + main.run_agent_loop(session, router, MagicMock(), registry.ToolRegistry(), + MagicMock(), config, MagicMock()) + return session, router + + +def test_scripted_agent_loop_consumes_search_and_fetch(wire, monkeypatch): + session, router = run_scripted_loop(monkeypatch) + transcript = session.get_messages_for_api() + for name in ("parallel_search", "parallel_fetch"): + assert any(f"[TOOL_RESULT: {name}]" in m["content"] and SOURCE in m["content"] + and "interactive interpreter" in m["content"] for m in transcript) + assert any(SOURCE in m["content"] for m in router.complete.call_args.kwargs["messages"]) + assert len(wire) == 8 + + +def test_scripted_agent_loop_honors_denied_approval(wire, monkeypatch): + session, _ = run_scripted_loop(monkeypatch, approval=True, denied=True) + assert not wire + assert any("[TOOL_BLOCKED: parallel_search]" in m["content"] + for m in session.get_messages_for_api()) diff --git a/tools/parallel_search.py b/tools/parallel_search.py new file mode 100644 index 0000000..1fecb96 --- /dev/null +++ b/tools/parallel_search.py @@ -0,0 +1,94 @@ +"""Optional anonymous web search and page extraction via Parallel Search MCP.""" + +import uuid +from urllib.parse import urlsplit + +from core.mcp import _HttpTransport +from core.version import __version__ + + +_ENDPOINT = "https://search.parallel.ai/mcp" +_HEADERS = { + "Accept": "application/json, text/event-stream", + "User-Agent": f"Operon/{__version__} (+https://github.com/OperonAgent/Operon)", +} +# Reuse one anonymous identity throughout this Operon process, including fetches. +_SESSION_ID = uuid.uuid4().hex + + +def _anonymous_auth(request): + """Prevent requests from picking up unrelated credentials from .netrc.""" + request.headers.pop("Authorization", None) + return request + + +def _result(response: dict) -> dict: + if "error" in response: + raise ValueError(str(response["error"])) + result = response.get("result") + if not isinstance(result, dict): + raise ValueError("Invalid MCP result") + return result + + +def _call(tool_name: str, arguments: dict) -> dict: + transport = _HttpTransport(_ENDPOINT, headers=_HEADERS, exact_url=True, + auth=_anonymous_auth) + try: + _result(transport.send_request("initialize", { + "protocolVersion": "2024-11-05", "capabilities": {}, + "clientInfo": {"name": "operon", "version": __version__}, + })) + notification = transport.send_notification("notifications/initialized") + if "error" in notification: + raise ValueError(str(notification["error"])) + tools = _result(transport.send_request("tools/list")).get("tools", []) + if not any(tool.get("name") == tool_name for tool in tools): + raise ValueError(f"Parallel MCP does not advertise {tool_name}") + result = _result(transport.send_request("tools/call", { + "name": tool_name, + "arguments": {**arguments, "session_id": _SESSION_ID}, + }, timeout=30.0)) + text = "\n".join(c.get("text", "") for c in result.get("content", []) + if c.get("type") == "text") + if result.get("isError"): + return {"success": False, "output": None, + "error": text or "Parallel MCP tool failed"} + output = result.get("structuredContent", text) + if not output: + raise ValueError("Parallel MCP returned no content") + return {"success": True, "output": output, "error": ""} + except Exception as e: + return {"success": False, "output": None, "error": str(e)} + finally: + transport.close() + + +def parallel_search(objective: str, search_queries: list) -> dict: + """Return focused web excerpts, titles and source URLs without an API key.""" + if (not isinstance(objective, str) or not objective.strip() + or not isinstance(search_queries, list) or not search_queries + or any(not isinstance(q, str) or not q.strip() for q in search_queries)): + return {"success": False, "output": None, + "error": "Provide an objective and a nonempty list of search_queries."} + return _call("web_search", {"objective": objective, "search_queries": search_queries}) + + +def parallel_fetch(urls: list, objective: str = "", full_content: bool = False) -> dict: + """Extract excerpts from up to 20 HTTP(S) source pages.""" + try: + if not isinstance(urls, list) or not 1 <= len(urls) <= 20: + raise ValueError("Provide between 1 and 20 HTTP(S) URLs.") + for url in urls: + parsed = urlsplit(url) if isinstance(url, str) else None + if (parsed is None or parsed.scheme not in ("http", "https") + or not parsed.hostname or parsed.username or parsed.password): + raise ValueError("Each URL must be HTTP(S) without embedded credentials.") + if not isinstance(objective, str) or len(objective) > 200: + raise ValueError("Fetch objective must be a string of at most 200 characters.") + if not isinstance(full_content, bool): + raise ValueError("full_content must be a boolean.") + except ValueError as e: + return {"success": False, "output": None, "error": str(e)} + return _call("web_fetch", {"urls": urls, "objective": objective or None, + "full_content": full_content}) diff --git a/tools/registry.py b/tools/registry.py index 32f5dd4..fecd605 100644 --- a/tools/registry.py +++ b/tools/registry.py @@ -14,6 +14,7 @@ ) from tools.shell_exec import shell_exec from tools.web_search import duckduckgo_search, web_scrape, x_search +from tools.parallel_search import parallel_search, parallel_fetch from tools.code_exec import python_exec from tools.http_client import http_request from tools.file_search import file_search @@ -242,6 +243,23 @@ def _td_required(td: dict) -> set: "max_chars": "integer — max characters to return, default 8000 (optional)", }, }, + { + "name": "parallel_search", + "description": "Optional Parallel web search with focused excerpts and source URLs. No account or API key. Use when Parallel is requested; DuckDuckGo remains available.", + "params": { + "objective": "string — focused information to find (required)", + "search_queries": "array of strings — related keyword queries (required)", + }, + }, + { + "name": "parallel_fetch", + "description": "Extract source-page excerpts with Parallel. Use for a specific URL or when search excerpts are insufficient. No account or API key.", + "params": { + "urls": "array of strings — 1 to 20 HTTP(S) URLs (required)", + "objective": "string — information to find, up to 200 characters (optional)", + "full_content": "boolean — return full markdown, default false (optional)", + }, + }, # ── Code execution ──────────────────────────────────────────────────────── { "name": "python_exec", @@ -2088,6 +2106,8 @@ def _td_required(td: dict) -> set: # Web "duckduckgo_search": duckduckgo_search, "web_scrape": web_scrape, + "parallel_search": parallel_search, + "parallel_fetch": parallel_fetch, # Code "python_exec": python_exec, # HTTP @@ -2399,7 +2419,8 @@ def spawn_agent(persona: str = "generalist", objective: str = "", "filesystem": ["file_read", "file_write", "file_append", "file_patch", "file_delete", "dir_list", "file_exists", "file_info", "file_search"], "shell": ["shell_exec", "python_exec"], - "web": ["duckduckgo_search", "web_scrape", "x_search", "http_request"], + "web": ["duckduckgo_search", "web_scrape", "x_search", "http_request", + "parallel_search", "parallel_fetch"], "email": ["email_draft"], "browser": ["browser_navigate", "browser_snapshot", "browser_screenshot", "browser_click", "browser_type", "browser_scroll", "browser_hover", @@ -2472,6 +2493,7 @@ def _get_original_fn(tool_name: str): "file_patch": file_patch, "file_delete": file_delete, "dir_list": dir_list, "file_exists": file_exists, "file_info": file_info, "shell_exec": shell_exec, "duckduckgo_search": duckduckgo_search, "web_scrape": web_scrape, + "parallel_search": parallel_search, "parallel_fetch": parallel_fetch, "x_search": x_search, "python_exec": python_exec, "http_request": http_request, "file_search": file_search, "email_draft": email_draft, "browser_navigate": browser_navigate, "browser_snapshot": browser_snapshot,