Merge fe3548037b into e3fade0a78

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).

integrations/parallel_search/__init__.py

@@ -0,0 +1 @@
+"""Standalone examples that use ReMe with third-party MCP services."""

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())

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()