From 209584e7fd29093573e74c521e0bce9a64d66167 Mon Sep 17 00:00:00 2001 From: Jonathan Singer Date: Thu, 20 Aug 2026 01:44:01 -0400 Subject: [PATCH] add a generic MCP client and a config for connecting MCP servers --- README.md | 24 ++ pyproject.toml | 2 + strix/core/runner.py | 18 ++ strix/tools/mcp/__init__.py | 22 ++ strix/tools/mcp/client.py | 237 ++++++++++++++++++ strix/tools/mcp/config.py | 85 +++++++ strix/tools/mcp/loader.py | 73 ++++++ tests/test_mcp_client.py | 478 ++++++++++++++++++++++++++++++++++++ 8 files changed, 939 insertions(+) create mode 100644 strix/tools/mcp/__init__.py create mode 100644 strix/tools/mcp/client.py create mode 100644 strix/tools/mcp/config.py create mode 100644 strix/tools/mcp/loader.py create mode 100644 tests/test_mcp_client.py diff --git a/README.md b/README.md index 96816114..903c73ba 100644 --- a/README.md +++ b/README.md @@ -320,6 +320,30 @@ strix auth status # show the active sign-in strix auth logout # forget the sign-in ``` +#### Connect your own MCP servers + +Strix can connect to Model Context Protocol (MCP) servers you list and expose their tools to the agent during a run. Create `~/.strix/mcp-servers.json` with a JSON list of servers. Each entry is either a local `stdio` server that Strix launches as a subprocess, or a remote `http` server: + +```json +[ + { + "name": "local_fs", + "transport": "stdio", + "command": "npx", + "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"] + }, + { + "name": "vercel", + "transport": "http", + "url": "https://mcp.vercel.com", + "auth": { "kind": "bearer", "token": "your-token" }, + "allowed_tools": ["list_projects"] + } +] +``` + +Each server's tools are namespaced by `name` (for example `local_fs.read_file`). Omit `allowed_tools` to expose every tool the server offers, or set it to a list to restrict which tools the agent can call. The file is optional, and a server that fails to connect is skipped without failing the run. You can point Strix at a different file with `STRIX_MCP_CONFIG`. + **Recommended models for best results:** - [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4` diff --git a/pyproject.toml b/pyproject.toml index 77be738f..16168344 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -241,6 +241,8 @@ ignore = [ "tests/test_stream_idle_timeout.py" = ["N802", "SLF001"] "tests/test_unknown_tool_recovery.py" = ["N802"] "tests/test_report_pdf.py" = ["S105", "S106"] +# Fake MCP server matches the SDK's MCPServer signature; its args are unused. +"tests/test_mcp_client.py" = ["S105", "S106", "ARG002"] # Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a # circular dependency with strix.telemetry / strix.interface.viewer.report_pdf. "strix/interface/viewer/server.py" = ["N802", "PLC0415"] diff --git a/strix/core/runner.py b/strix/core/runner.py index b4afdfaf..c2985240 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -51,6 +51,7 @@ from strix.tools.output_store import ( if TYPE_CHECKING: + from agents.mcp import MCPServer from agents.memory import SQLiteSession from agents.result import RunResultBase @@ -253,6 +254,7 @@ async def run_strix_scan( configure_spill_writer(_spill_to_workspace) sessions_to_close: list[SQLiteSession] = [] + mcp_servers: list[MCPServer] = [] try: targets = scan_config.get("targets") or [] @@ -298,6 +300,19 @@ async def run_strix_scan( system_prompt_context=root_context, ) + # Connect any MCP servers the user listed in ~/.strix/mcp-servers.json and + # register their tools before the agent is built. These are the user's own + # servers, so no result scrub is applied. Fully fail-open: a missing file or + # a failed connection must never break a normal run (a managed run has no file). + from strix.tools.mcp import connect_mcp_servers, load_user_mcp_configs + + try: + user_mcp_configs = load_user_mcp_configs() + if user_mcp_configs: + mcp_servers = await connect_mcp_servers(user_mcp_configs) + except Exception: + logger.exception("Failed to connect user MCP servers; continuing without them") + root_agent = build_strix_agent( name="Root Agent", skills=skills, @@ -472,6 +487,9 @@ async def run_strix_scan( for s in sessions_to_close: with contextlib.suppress(Exception): s.close() + for mcp_server in mcp_servers: + with contextlib.suppress(Exception): + await mcp_server.cleanup() # type: ignore[no-untyped-call] with contextlib.suppress(Exception): await coordinator._maybe_snapshot() if cleanup_on_exit: diff --git a/strix/tools/mcp/__init__.py b/strix/tools/mcp/__init__.py new file mode 100644 index 00000000..2449b5d8 --- /dev/null +++ b/strix/tools/mcp/__init__.py @@ -0,0 +1,22 @@ +"""Generic MCP client: connect MCP servers and expose their tools.""" + +from __future__ import annotations + +from strix.tools.mcp.client import connect_mcp_servers +from strix.tools.mcp.config import ( + AwsSigV4Auth, + BearerAuth, + McpAuth, + McpConnectionConfig, +) +from strix.tools.mcp.loader import load_user_mcp_configs + + +__all__ = [ + "AwsSigV4Auth", + "BearerAuth", + "McpAuth", + "McpConnectionConfig", + "connect_mcp_servers", + "load_user_mcp_configs", +] diff --git a/strix/tools/mcp/client.py b/strix/tools/mcp/client.py new file mode 100644 index 00000000..f4419db4 --- /dev/null +++ b/strix/tools/mcp/client.py @@ -0,0 +1,237 @@ +"""Connect to MCP servers and expose their tools to the agent. + +Given one :class:`McpConnectionConfig` per server, :func:`connect_mcp_servers` +lists each server's tools, keeps the ones on the connection's allowlist (or all +of them when none is set), prefixes each with the connection name so servers do +not collide, and registers them through the agent factory. The factory applies +output bounding, per-call timeouts, and structured errors to every registered +tool, so this layer does not reimplement them. + +A server that cannot connect, or a tool set that cannot be registered, is logged +and skipped, so one bad connection never fails the run. +""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING, Any, cast + +from agents.exceptions import ModelBehaviorError +from agents.mcp import ( + MCPServer, + MCPServerStdio, + MCPServerStdioParams, + MCPServerStreamableHttp, + MCPServerStreamableHttpParams, + MCPUtil, + create_static_tool_filter, +) + +from strix.agents.factory import register_agent_tools +from strix.tools.mcp.config import BearerAuth, McpConnectionConfig + + +if TYPE_CHECKING: + from collections.abc import Callable + + from agents.tool import FunctionTool, Tool + from mcp.types import Tool as MCPTool + + # Runs on each tool's structured result before it reaches the agent. Called + # ``result_transform(namespaced_tool_name, structured_result)`` and its return + # value becomes the tool's output. ``structured_result`` is the parsed + # ``CallToolResult`` as a dict (not a serialized string), so the transform can + # project or drop individual fields. + ResultTransform = Callable[[str, Any], Any] + + +logger = logging.getLogger(__name__) + + +def _auth_headers(config: McpConnectionConfig) -> dict[str, str]: + """Build the per-server request headers from the connection's auth.""" + auth = config.auth + if auth is None: + return {} + if isinstance(auth, BearerAuth): + return {"Authorization": f"Bearer {auth.token}"} + + # The only other variant is AWS SigV4. + # TODO: AWS SigV4 transport and auth are UNVERIFIED. Confirm how the target + # AWS MCP server is reached (stdio vs streamable HTTP) and how it accepts + # SigV4-signed requests before enabling this branch. Do not fabricate request + # signing here. + raise NotImplementedError( + "AWS SigV4 MCP auth is not verified yet; confirm the server's transport " + "and request signing before connecting an aws_sigv4 connection." + ) + + +def _build_server(config: McpConnectionConfig) -> MCPServer: + """Construct (but do not connect) the SDK server for one connection. + + When ``allowed_tools`` is a list the static filter means the server will not + even list tools outside it; :func:`_register_server_tools` re-applies the + same allowlist as the authoritative gate on what gets registered. When it is + ``None`` no filter is applied and every listed tool is registered. + """ + tool_filter = ( + create_static_tool_filter(allowed_tool_names=config.allowed_tools) + if config.allowed_tools is not None + else None + ) + + if config.transport == "stdio": + stdio_params: MCPServerStdioParams = { + "command": cast("str", config.command), + "args": config.args, + "env": config.env, + } + return MCPServerStdio( + params=stdio_params, + name=config.name, + tool_filter=tool_filter, + cache_tools_list=True, + ) + + http_params: MCPServerStreamableHttpParams = { + "url": cast("str", config.url), + "headers": _auth_headers(config), + } + return MCPServerStreamableHttp( + params=http_params, + name=config.name, + tool_filter=tool_filter, + cache_tools_list=True, + ) + + +def _build_tool( + config: McpConnectionConfig, + server: MCPServer, + mcp_tool: MCPTool, + result_transform: ResultTransform | None, +) -> FunctionTool: + """Build one namespaced FunctionTool from a listed MCP tool. + + Without ``result_transform`` this is exactly the SDK's stock conversion. With + one, the SDK still builds the tool (so name override, input schema, approval + policy, error-as-result handling, and tool-origin metadata are unchanged), but + we route the underlying MCP call through :func:`_install_result_transform` so + the transform sees the structured result and decides the tool's output. + """ + namespaced_name = f"{config.name}.{mcp_tool.name}" + tool = MCPUtil.to_function_tool( + mcp_tool, + server, + convert_schemas_to_strict=False, + tool_name_override=namespaced_name, + ) + if result_transform is not None: + _install_result_transform(tool, server, mcp_tool.name, namespaced_name, result_transform) + return tool + + +def _install_result_transform( + tool: FunctionTool, + server: MCPServer, + base_tool_name: str, + namespaced_name: str, + result_transform: ResultTransform, +) -> None: + """Route a tool's MCP call through ``result_transform``, innermost. + + ``MCPUtil.to_function_tool`` serializes the result inside its own invoke, so + the structured result cannot be intercepted through it. Instead we call + ``server.call_tool`` ourselves, hand the parsed :class:`CallToolResult` to the + transform, and return the transform's output as the tool result. + + This runs INSIDE the tool's invoke. The agent factory wraps a registered + tool's ``on_invoke_tool`` with output bounding, disk spill, and tracing at + agent-build time, which is OUTSIDE this invoke, so the transform is genuinely + the innermost step: nothing sees the raw result before the transform does. + + ``to_function_tool`` wraps the real invoke in the SDK's failure-handling + invoker, which stores the inner coroutine on ``_invoke_tool_impl`` and calls + it inside its try/except. Swapping that inner impl keeps the SDK's + error-as-result handling and all tool metadata while inserting the transform. + If the SDK ever renames that attribute we fail loudly rather than silently + skip the transform. + """ + + async def _invoke(_ctx: Any, input_json: str) -> Any: + parsed: Any = json.loads(input_json) if input_json else {} + if not isinstance(parsed, dict): + raise ModelBehaviorError( + f"Invalid JSON input for tool {namespaced_name}: expected a JSON object" + ) + args = cast("dict[str, Any]", parsed) + result = await server.call_tool(base_tool_name, args) + structured_result = result.model_dump(mode="json") + return result_transform(namespaced_name, structured_result) + + # ``tool.on_invoke_tool`` is the SDK's failure-handling invoker; it is a plain + # object with the inner coroutine on ``_invoke_tool_impl``, not a function, so + # treat it as untyped to swap that attribute. + invoker = cast("Any", tool.on_invoke_tool) + if not hasattr(invoker, "_invoke_tool_impl"): + raise RuntimeError( + "agents SDK FunctionTool invoker shape changed: cannot install the " + "result transform without risking it being silently skipped." + ) + invoker._invoke_tool_impl = _invoke + + +async def _register_server_tools( + config: McpConnectionConfig, + server: MCPServer, + result_transform: ResultTransform | None = None, +) -> list[Tool]: + """List a connected server's tools, prefix + filter them, and register them. + + ``allowed_tools`` of ``None`` registers every listed tool; a list restricts + to exactly those names. + """ + allowed = config.allowed_tools + mcp_tools = await server.list_tools() + + tools: list[Tool] = [ + _build_tool(config, server, mcp_tool, result_transform) + for mcp_tool in mcp_tools + if allowed is None or mcp_tool.name in allowed + ] + + register_agent_tools(*tools) + return tools + + +async def connect_mcp_servers( + configs: list[McpConnectionConfig], + result_transform: ResultTransform | None = None, +) -> list[MCPServer]: + """Connect to each MCP server and register its tools. + + When ``result_transform`` is given, every registered tool routes its result + through it before the result reaches the agent (see + :func:`_install_result_transform`). When it is ``None`` the tools behave + exactly as the SDK builds them. + + Returns the servers that connected, so the caller can clean them up when the + run ends. Connections that fail are skipped rather than raised. + """ + connected: list[MCPServer] = [] + for config in configs: + server = _build_server(config) + try: + await server.connect() # type: ignore[no-untyped-call] + tools = await _register_server_tools(config, server, result_transform) + except Exception: + logger.exception("Skipping MCP connection %r", config.name) + await server.cleanup() # type: ignore[no-untyped-call] + continue + + logger.info("Connected MCP server %r (%d tools)", config.name, len(tools)) + connected.append(server) + + return connected diff --git a/strix/tools/mcp/config.py b/strix/tools/mcp/config.py new file mode 100644 index 00000000..2df1588b --- /dev/null +++ b/strix/tools/mcp/config.py @@ -0,0 +1,85 @@ +"""The connection-config contract for the MCP client. + +Describes one MCP server the client can connect to: its transport, endpoint or +launch command, optional auth, and an optional tool allowlist. Field names are +stable; callers build against them. +""" + +from __future__ import annotations + +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +class BearerAuth(BaseModel): + """Header-token auth, sent as ``Authorization: Bearer ``.""" + + model_config = ConfigDict(extra="forbid") + + kind: Literal["bearer"] = "bearer" + token: str = Field(min_length=1, repr=False) + + +class AwsSigV4Auth(BaseModel): + """Request-signing auth for AWS. + + AWS does not authenticate with a header token; each request is signed. The + temporary key is minted per run and passed out of band, never read from the + ambient environment. + """ + + model_config = ConfigDict(extra="forbid") + + kind: Literal["aws_sigv4"] = "aws_sigv4" + access_key_id: str = Field(min_length=1, repr=False) + secret_access_key: str = Field(min_length=1, repr=False) + session_token: str | None = Field(default=None, repr=False) + region: str = Field(min_length=1) + + +McpAuth = Annotated[BearerAuth | AwsSigV4Auth, Field(discriminator="kind")] + + +class McpConnectionConfig(BaseModel): + """One MCP server the client can connect to. + + Two transports are supported: streamable ``http`` (a remote endpoint) and + ``stdio`` (a local server launched as a subprocess). + """ + + model_config = ConfigDict(extra="forbid") + + name: str = Field(min_length=1) + """Namespaced tool prefix, unique per run (e.g. ``github``).""" + + transport: Literal["http", "stdio"] = "http" + """``http`` for a streamable HTTP endpoint, ``stdio`` for a local subprocess.""" + + url: str | None = Field(default=None, min_length=1) + """The MCP server endpoint. Required for ``http``.""" + + auth: McpAuth | None = None + """Bearer token or AWS SigV4 signing material. Optional; a local stdio + server usually needs none.""" + + command: str | None = Field(default=None, min_length=1) + """The executable to launch for ``stdio``. Required for ``stdio``.""" + + args: list[str] = Field(default_factory=list) + """Arguments passed to ``command`` (stdio only).""" + + env: dict[str, str] = Field(default_factory=dict) + """Extra environment variables for the stdio subprocess.""" + + allowed_tools: list[str] | None = None + """Tool allowlist, applied after the server lists its tools. ``None`` (the + default) exposes every tool the server lists; a list restricts to it.""" + + @model_validator(mode="after") + def _check_transport_fields(self) -> McpConnectionConfig: + if self.transport == "http" and not self.url: + raise ValueError("an http MCP connection requires 'url'") + if self.transport == "stdio" and not self.command: + raise ValueError("a stdio MCP connection requires 'command'") + return self diff --git a/strix/tools/mcp/loader.py b/strix/tools/mcp/loader.py new file mode 100644 index 00000000..1bc45c18 --- /dev/null +++ b/strix/tools/mcp/loader.py @@ -0,0 +1,73 @@ +"""Read the open-source user's MCP servers from ``~/.strix/mcp-servers.json``. + +An open-source user lists the MCP servers they want the agent to reach in a +small JSON file. Strix reads it at the start of a run, connects to each server, +and registers its tools. The file is optional: a managed/saas run simply won't +have one, which is fine. + +Parsing is fail-open. A single malformed entry is logged and skipped rather than +raising, so one bad row never blocks the servers that are valid, and a missing +or unreadable file yields an empty list. +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path +from typing import cast + +from pydantic import ValidationError + +from strix.tools.mcp.config import McpConnectionConfig + + +logger = logging.getLogger(__name__) + + +_DEFAULT_PATH: Path = Path.home() / ".strix" / "mcp-servers.json" +_PATH_ENV_VAR = "STRIX_MCP_CONFIG" + + +def _resolve_path(path: Path | None) -> Path: + if path is not None: + return path + override = os.environ.get(_PATH_ENV_VAR) + if override: + return Path(override) + return _DEFAULT_PATH + + +def load_user_mcp_configs(path: Path | None = None) -> list[McpConnectionConfig]: + """Load MCP connection configs from the user's JSON file. + + The path is ``path`` if given, else ``$STRIX_MCP_CONFIG``, else + ``~/.strix/mcp-servers.json``. The file is a JSON list of server entries. + A missing file returns ``[]``; an unreadable or non-list file is logged and + returns ``[]``; individual entries that fail validation are logged and + skipped. + """ + source = _resolve_path(path) + if not source.exists(): + return [] + + try: + raw = json.loads(source.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + logger.exception("Could not read MCP config at %s; ignoring it", source) + return [] + + if not isinstance(raw, list): + logger.warning("MCP config at %s is not a JSON list; ignoring it", source) + return [] + + entries = cast("list[object]", raw) + configs: list[McpConnectionConfig] = [] + for index, entry in enumerate(entries): + try: + configs.append(McpConnectionConfig.model_validate(entry)) + except ValidationError as exc: + logger.warning("Skipping invalid MCP server entry #%d in %s: %s", index, source, exc) + + return configs diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py new file mode 100644 index 00000000..edece027 --- /dev/null +++ b/tests/test_mcp_client.py @@ -0,0 +1,478 @@ +"""Tests for the generic MCP client: config contract, namespacing, and filtering.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +import pytest +from agents.mcp import MCPServer, MCPServerStdio, MCPServerStreamableHttp +from mcp.types import CallToolResult, TextContent +from mcp.types import Tool as MCPTool +from pydantic import ValidationError + +from strix.agents import factory +from strix.tools.mcp import ( + AwsSigV4Auth, + BearerAuth, + McpConnectionConfig, + load_user_mcp_configs, +) +from strix.tools.mcp.client import _auth_headers, _build_server, _register_server_tools + + +if TYPE_CHECKING: + from pathlib import Path + + from agents.tool import Tool + + +class FakeMCPServer(MCPServer): + """A connected MCP server stand-in, so tests never touch the network.""" + + def __init__(self, name: str, tools: list[MCPTool]) -> None: + super().__init__() + self._name = name + self._tools = tools + self.calls: list[tuple[str, dict[str, Any] | None]] = [] + + @property + def name(self) -> str: + return self._name + + async def connect(self) -> None: + return None + + async def cleanup(self) -> None: + return None + + async def list_tools( + self, + run_context: Any = None, + agent: Any = None, + ) -> list[MCPTool]: + return list(self._tools) + + async def call_tool( + self, + tool_name: str, + arguments: dict[str, Any] | None, + meta: dict[str, Any] | None = None, + ) -> CallToolResult: + self.calls.append((tool_name, arguments)) + return CallToolResult(content=[TextContent(type="text", text=f"routed:{tool_name}")]) + + async def list_prompts(self) -> Any: + raise NotImplementedError + + async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> Any: + raise NotImplementedError + + +def _mcp_tool(name: str) -> MCPTool: + return MCPTool( + name=name, + description=f"remote tool {name}", + inputSchema={"type": "object", "properties": {}}, + ) + + +def _config(name: str, allowed_tools: list[str]) -> McpConnectionConfig: + return McpConnectionConfig( + name=name, + url="https://mcp.example.com", + auth=BearerAuth(token="run-token"), + allowed_tools=allowed_tools, + ) + + +@pytest.fixture(autouse=True) +def _reset_registry() -> Any: + saved = list(factory._EXTRA_TOOLS) + factory._EXTRA_TOOLS.clear() + try: + yield + finally: + factory._EXTRA_TOOLS[:] = saved + + +# --- config contract --------------------------------------------------------- + + +def test_bearer_config_parses_from_dict() -> None: + config = McpConnectionConfig.model_validate( + { + "name": "vercel_main", + "transport": "http", + "url": "https://mcp.example.com", + "auth": {"kind": "bearer", "token": "abc"}, + "allowed_tools": ["list_projects"], + } + ) + + assert isinstance(config.auth, BearerAuth) + assert config.auth.token == "abc" + assert config.allowed_tools == ["list_projects"] + + +def test_aws_sigv4_config_parses_from_dict() -> None: + config = McpConnectionConfig.model_validate( + { + "name": "aws_production", + "url": "https://mcp.example.com", + "auth": { + "kind": "aws_sigv4", + "access_key_id": "AKIA", + "secret_access_key": "secret", + "session_token": "session", + "region": "us-east-1", + }, + } + ) + + assert isinstance(config.auth, AwsSigV4Auth) + assert config.auth.region == "us-east-1" + # transport defaults to http, allowed_tools to None ("all tools"). + assert config.transport == "http" + assert config.allowed_tools is None + + +def test_unknown_auth_kind_is_rejected() -> None: + with pytest.raises(ValidationError): + McpConnectionConfig.model_validate( + { + "name": "x", + "url": "https://mcp.example.com", + "auth": {"kind": "oauth", "token": "abc"}, + } + ) + + +def test_stdio_config_parses_from_dict() -> None: + config = McpConnectionConfig.model_validate( + { + "name": "local_fs", + "transport": "stdio", + "command": "npx", + "args": ["-y", "@modelcontextprotocol/server-filesystem", "/srv/data"], + "env": {"FOO": "bar"}, + } + ) + + assert config.transport == "stdio" + assert config.command == "npx" + assert config.args == ["-y", "@modelcontextprotocol/server-filesystem", "/srv/data"] + assert config.env == {"FOO": "bar"} + # A local stdio server needs no auth, and omitting allowed_tools means "all". + assert config.auth is None + assert config.allowed_tools is None + + +def test_http_config_without_url_is_rejected() -> None: + with pytest.raises(ValidationError): + McpConnectionConfig.model_validate( + { + "name": "x", + "transport": "http", + "auth": {"kind": "bearer", "token": "abc"}, + } + ) + + +def test_stdio_config_without_command_is_rejected() -> None: + with pytest.raises(ValidationError): + McpConnectionConfig.model_validate( + { + "name": "x", + "transport": "stdio", + } + ) + + +def test_empty_name_is_rejected() -> None: + with pytest.raises(ValidationError): + McpConnectionConfig.model_validate( + { + "name": "", + "url": "https://mcp.example.com", + "auth": {"kind": "bearer", "token": "abc"}, + } + ) + + +def test_unknown_field_is_rejected() -> None: + with pytest.raises(ValidationError): + McpConnectionConfig.model_validate( + { + "name": "x", + "url": "https://mcp.example.com", + "auth": {"kind": "bearer", "token": "abc"}, + "surprise": True, + } + ) + + +# --- auth headers ------------------------------------------------------------ + + +def test_bearer_auth_builds_authorization_header() -> None: + headers = _auth_headers(_config("vercel_main", [])) + + assert headers == {"Authorization": "Bearer run-token"} + + +def test_aws_sigv4_auth_is_not_implemented_yet() -> None: + config = McpConnectionConfig( + name="aws_production", + url="https://mcp.example.com", + auth=AwsSigV4Auth( + access_key_id="AKIA", + secret_access_key="secret", + region="us-east-1", + ), + ) + + with pytest.raises(NotImplementedError): + _auth_headers(config) + + +# --- namespacing and filtering ----------------------------------------------- + + +def _registered_names() -> list[str]: + return [tool.name for tool in factory.registered_agent_tools()] + + +@pytest.mark.asyncio +async def test_tools_are_namespaced_per_connection() -> None: + server_a = FakeMCPServer("conn_a", [_mcp_tool("describe")]) + server_b = FakeMCPServer("conn_b", [_mcp_tool("describe")]) + + await _register_server_tools(_config("conn_a", ["describe"]), server_a) + await _register_server_tools(_config("conn_b", ["describe"]), server_b) + + # Same remote tool name on two connections does not collide. + assert _registered_names() == ["conn_a.describe", "conn_b.describe"] + + +@pytest.mark.asyncio +async def test_disallowed_tool_is_not_registered() -> None: + server = FakeMCPServer( + "vercel_main", + [_mcp_tool("list_projects"), _mcp_tool("delete_project")], + ) + + await _register_server_tools(_config("vercel_main", ["list_projects"]), server) + + names = _registered_names() + assert "vercel_main.list_projects" in names + assert "vercel_main.delete_project" not in names + + +@pytest.mark.asyncio +async def test_allowed_tools_none_registers_every_listed_tool() -> None: + server = FakeMCPServer( + "local_fs", + [_mcp_tool("read_file"), _mcp_tool("write_file")], + ) + config = McpConnectionConfig(name="local_fs", url="https://mcp.example.com", allowed_tools=None) + + await _register_server_tools(config, server) + + names = _registered_names() + assert "local_fs.read_file" in names + assert "local_fs.write_file" in names + + +@pytest.mark.asyncio +async def test_allowed_tools_list_restricts_registration() -> None: + server = FakeMCPServer( + "local_fs", + [_mcp_tool("read_file"), _mcp_tool("write_file")], + ) + + await _register_server_tools(_config("local_fs", ["read_file"]), server) + + names = _registered_names() + assert names == ["local_fs.read_file"] + + +@pytest.mark.asyncio +async def test_registered_tool_routes_to_its_server_with_the_original_name() -> None: + server = FakeMCPServer("vercel_main", [_mcp_tool("list_projects")]) + + tools: list[Tool] = await _register_server_tools( + _config("vercel_main", ["list_projects"]), server + ) + tool = tools[0] + + output = await tool.on_invoke_tool(None, "{}") # type: ignore[union-attr] + + # The call reaches the right server, addressed by the unprefixed remote name. + assert server.calls == [("list_projects", {})] + assert output == {"type": "text", "text": "routed:list_projects"} + + +# --- result transform -------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_result_transform_receives_namespaced_name_and_structured_result() -> None: + server = FakeMCPServer("vercel_main", [_mcp_tool("list_projects")]) + seen: list[tuple[str, Any]] = [] + + def transform(name: str, structured: Any) -> Any: + seen.append((name, structured)) + return "scrubbed" + + tools: list[Tool] = await _register_server_tools( + _config("vercel_main", ["list_projects"]), server, result_transform=transform + ) + + output = await tools[0].on_invoke_tool(None, "{}") # type: ignore[union-attr] + + # The underlying MCP call still routes by the unprefixed remote name. + assert server.calls == [("list_projects", {})] + + # The transform is called with the namespaced name and the parsed result. + assert len(seen) == 1 + name, structured = seen[0] + assert name == "vercel_main.list_projects" + # A parsed CallToolResult (dict/list), not a pre-serialized string. + assert structured["content"][0]["text"] == "routed:list_projects" + assert structured["isError"] is False + + # The transform's return value is exactly what the tool yields. + assert output == "scrubbed" + + +@pytest.mark.asyncio +async def test_result_transform_can_rewrite_the_tool_output() -> None: + server = FakeMCPServer("vercel_main", [_mcp_tool("list_projects")]) + + def transform(_name: str, structured: Any) -> Any: + # Withhold everything but a redacted view of the text field. + return f"redacted<{structured['content'][0]['text']}>" + + tools: list[Tool] = await _register_server_tools( + _config("vercel_main", ["list_projects"]), server, result_transform=transform + ) + + output = await tools[0].on_invoke_tool(None, "{}") # type: ignore[union-attr] + + assert output == "redacted" + + +@pytest.mark.asyncio +async def test_without_result_transform_output_is_unchanged() -> None: + server = FakeMCPServer("vercel_main", [_mcp_tool("list_projects")]) + + tools: list[Tool] = await _register_server_tools( + _config("vercel_main", ["list_projects"]), server, result_transform=None + ) + + output = await tools[0].on_invoke_tool(None, "{}") # type: ignore[union-attr] + + # Same shape the SDK produces today: no transform in the path. + assert server.calls == [("list_projects", {})] + assert output == {"type": "text", "text": "routed:list_projects"} + + +# --- server build branch ----------------------------------------------------- + + +def test_build_server_stdio_branch() -> None: + config = McpConnectionConfig( + name="local_fs", + transport="stdio", + command="my-server", + args=["--flag", "value"], + env={"TOKEN": "x"}, + ) + + server = _build_server(config) + + # Built, not connected: no subprocess is launched here. + assert isinstance(server, MCPServerStdio) + assert server.name == "local_fs" + assert server.params.command == "my-server" + assert server.params.args == ["--flag", "value"] + assert server.params.env == {"TOKEN": "x"} + + +def test_build_server_http_branch() -> None: + server = _build_server(_config("vercel_main", ["list_projects"])) + + assert isinstance(server, MCPServerStreamableHttp) + assert server.name == "vercel_main" + + +# --- loader ------------------------------------------------------------------ + + +def test_loader_parses_stdio_and_http_entries(tmp_path: Path) -> None: + config_file = tmp_path / "mcp-servers.json" + config_file.write_text( + json.dumps( + [ + { + "name": "local_fs", + "transport": "stdio", + "command": "npx", + "args": ["-y", "server-filesystem"], + }, + { + "name": "vercel_main", + "transport": "http", + "url": "https://mcp.example.com", + "auth": {"kind": "bearer", "token": "abc"}, + "allowed_tools": ["list_projects"], + }, + ] + ), + encoding="utf-8", + ) + + configs = load_user_mcp_configs(config_file) + + assert [c.name for c in configs] == ["local_fs", "vercel_main"] + assert configs[0].transport == "stdio" + assert configs[1].allowed_tools == ["list_projects"] + + +def test_loader_skips_bad_entry_but_keeps_good_ones(tmp_path: Path) -> None: + config_file = tmp_path / "mcp-servers.json" + config_file.write_text( + json.dumps( + [ + {"name": "broken", "transport": "http"}, # missing url + { + "name": "local_fs", + "transport": "stdio", + "command": "npx", + }, + ] + ), + encoding="utf-8", + ) + + configs = load_user_mcp_configs(config_file) + + assert [c.name for c in configs] == ["local_fs"] + + +def test_loader_returns_empty_when_file_absent(tmp_path: Path) -> None: + assert load_user_mcp_configs(tmp_path / "does-not-exist.json") == [] + + +def test_loader_reads_env_var_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + config_file = tmp_path / "from-env.json" + config_file.write_text( + json.dumps([{"name": "local_fs", "transport": "stdio", "command": "npx"}]), + encoding="utf-8", + ) + monkeypatch.setenv("STRIX_MCP_CONFIG", str(config_file)) + + configs = load_user_mcp_configs() + + assert [c.name for c in configs] == ["local_fs"]