diff --git a/integrations/parallel_search/README.md b/integrations/parallel_search/README.md new file mode 100644 index 00000000..f623b2a9 --- /dev/null +++ b/integrations/parallel_search/README.md @@ -0,0 +1,33 @@ +# Parallel Search MCP example + +This opt-in script demonstrates ReMe's `MCPClient` with Parallel's current +streamable-HTTP MCP endpoint. It discovers `web_search` and `web_fetch`, searches +for a topic, then fetches a URL you select. Search and fetch are available +without an API key at lower rate limits. The example prints results only; it +does not call ReMe memory jobs, save files, or ingest web content. +Running it sends the topic and selected URL to Parallel for processing. + +## Install and run + +From the ReMe repository root, install ReMe and its declared FastMCP dependency: + +```bash +python -m pip install -e . +python integrations/parallel_search/example.py "recent ReMe memory updates" https://github.com/agentscope-ai/ReMe +``` + +Python 3.11 or later is required. This example connects anonymously and does +not read or persist credentials. See Parallel's documentation for authenticated +setups. The script uses a 60-second client timeout and reports provider +rate-limit responses separately. + +## Tests + +The unit tests mock the MCP client and transport, so they do not contact Parallel: + +```bash +python -m pytest tests/unit/test_parallel_search_example.py -q +``` + +For provider details and current tool arguments, see Parallel's +[MCP quickstart](https://docs.parallel.ai/integrations/mcp/quickstart). diff --git a/integrations/parallel_search/__init__.py b/integrations/parallel_search/__init__.py new file mode 100644 index 00000000..881fd08d --- /dev/null +++ b/integrations/parallel_search/__init__.py @@ -0,0 +1 @@ +"""Standalone examples that use ReMe with third-party MCP services.""" diff --git a/integrations/parallel_search/example.py b/integrations/parallel_search/example.py new file mode 100644 index 00000000..68fe5383 --- /dev/null +++ b/integrations/parallel_search/example.py @@ -0,0 +1,79 @@ +"""Call Parallel's public web-search MCP tools through ReMe's MCP client.""" + +import argparse +import asyncio +import sys +from urllib.parse import urlparse + +from fastmcp.client import StreamableHttpTransport + +from reme.components.client import MCPClient + +SERVER_URL = "https://mcp.parallel.ai/mcp" +REQUEST_TIMEOUT_SECONDS = 60 + + +def create_client() -> MCPClient: + """Create a ReMe client for Parallel's anonymous streamable-HTTP endpoint.""" + transport = StreamableHttpTransport(url=SERVER_URL) + return MCPClient(transport=transport, timeout=REQUEST_TIMEOUT_SECONDS) + + +def validate_url(value: str) -> str: + """Accept absolute HTTP(S) URLs without credentials for the remote fetch tool.""" + parsed = urlparse(value) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise argparse.ArgumentTypeError("fetch URL must be an absolute http:// or https:// URL") + if parsed.username is not None or parsed.password is not None: + raise argparse.ArgumentTypeError("fetch URL must not include embedded credentials") + return value + + +async def run_example(topic: str, fetch_url: str, client: MCPClient, emit=print) -> None: + """Discover and run the two public search tools without writing to ReMe memory.""" + async with client: + tools = await client.list_actions() + available = {tool["name"] for tool in tools} + required = {"web_search", "web_fetch"} + missing = sorted(required - available) + if missing: + raise RuntimeError(f"Parallel MCP endpoint is missing required tools: {', '.join(missing)}") + + emit(f"## Search: {topic}") + async for result in client( + action="web_search", + objective=topic, + search_queries=[topic], + ): + emit(result) + + emit(f"## Fetch: {fetch_url}") + async for result in client(action="web_fetch", urls=[fetch_url], objective=topic): + emit(result) + + +def main(argv: list[str] | None = None) -> int: + """Run the opt-in example from the command line.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("topic", help="what to search for on the public web") + parser.add_argument("url", type=validate_url, help="HTTP(S) URL to fetch after searching") + args = parser.parse_args(argv) + + try: + asyncio.run(run_example(args.topic, args.url, create_client())) + except Exception as exc: # Report provider and transport failures to the caller. + status_code = getattr(getattr(exc, "response", None), "status_code", None) + if status_code == 429: + print( + "Parallel Search rate limit reached. Wait before retrying or see provider docs " + "for authenticated access.", + file=sys.stderr, + ) + else: + print(f"Parallel MCP request failed ({type(exc).__name__}): {exc}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/unit/test_parallel_search_example.py b/tests/unit/test_parallel_search_example.py new file mode 100644 index 00000000..b8c519b5 --- /dev/null +++ b/tests/unit/test_parallel_search_example.py @@ -0,0 +1,133 @@ +"""Offline tests for the opt-in Parallel MCP example.""" + +import argparse +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from integrations.parallel_search import example + + +class FakeClient: + """Mimic the small public MCPClient surface used by the example.""" + + def __init__(self, actions=None, responses=None): + self.list_actions = AsyncMock(return_value=actions or []) + self.calls = [] + self.responses = responses or {} + + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc_info): + return None + + def __call__(self, action, **kwargs): + self.calls.append((action, kwargs)) + + async def stream(): + for response in self.responses.get(action, []): + yield response + + return stream() + + +def test_create_client_uses_current_parallel_endpoint(monkeypatch): + """The custom transport targets Parallel's current public MCP endpoint.""" + transport = object() + client = object() + transport_calls = [] + client_calls = [] + + def transport_factory(**kwargs): + transport_calls.append(kwargs) + return transport + + def client_factory(**kwargs): + client_calls.append(kwargs) + return client + + monkeypatch.setattr(example, "StreamableHttpTransport", transport_factory) + monkeypatch.setattr(example, "MCPClient", client_factory) + + result = example.create_client() + + assert transport_calls == [{"url": "https://mcp.parallel.ai/mcp"}] + assert client_calls == [{"transport": transport, "timeout": 60}] + assert result is client + + +def test_validate_url_requires_public_http_url(): + """Only absolute HTTP(S) fetch targets without embedded credentials are accepted.""" + assert example.validate_url("https://example.com/article") == "https://example.com/article" + assert example.validate_url("http://example.com/article") == "http://example.com/article" + with pytest.raises(argparse.ArgumentTypeError, match="absolute http"): + example.validate_url("file:///etc/passwd") + with pytest.raises(argparse.ArgumentTypeError, match="absolute http"): + example.validate_url("example.com/article") + with pytest.raises(argparse.ArgumentTypeError, match="embedded credentials"): + example.validate_url("https://user:password@example.com/article") + + +def test_run_example_discovers_and_calls_both_tools_in_order(): + """The example uses the documented tool arguments and prints both results.""" + client = FakeClient( + actions=[{"name": "web_search"}, {"name": "web_fetch"}], + responses={"web_search": ["search result"], "web_fetch": ["page content"]}, + ) + output = [] + + topic = "latest memory system updates" + asyncio.run(example.run_example(topic, "https://example.com", client, output.append)) + + assert client.list_actions.await_count == 1 + assert client.calls == [ + ( + "web_search", + {"objective": topic, "search_queries": [topic]}, + ), + ( + "web_fetch", + {"urls": ["https://example.com"], "objective": topic}, + ), + ] + assert output == [ + f"## Search: {topic}", + "search result", + "## Fetch: https://example.com", + "page content", + ] + + +def test_run_example_fails_before_calling_when_tools_are_missing(): + """An incompatible endpoint fails before any tool is dispatched.""" + client = FakeClient(actions=[{"name": "web_search"}]) + + with pytest.raises(RuntimeError, match="web_fetch"): + asyncio.run(example.run_example("topic", "https://example.com", client)) + + assert not client.calls + + +class RateLimitError(Exception): + """Fake HTTP error carrying the provider's 429 status.""" + + def __init__(self): + super().__init__("too many requests") + self.response = SimpleNamespace(status_code=429) + + +def test_main_reports_provider_rate_limit(monkeypatch, capsys): + """A provider 429 gives the user actionable rate-limit guidance.""" + client = FakeClient(actions=[{"name": "web_search"}, {"name": "web_fetch"}]) + + async def fail(*_args, **_kwargs): + raise RateLimitError() + + monkeypatch.setattr(example, "create_client", lambda: client) + monkeypatch.setattr(example, "run_example", fail) + + assert example.main(["topic", "https://example.com"]) == 1 + assert "rate limit" in capsys.readouterr().err.lower()