mirror of
https://github.com/usestrix/strix.git
synced 2026-08-21 18:52:47 +02:00
add a generic MCP client and a config for connecting MCP servers
This commit is contained in:
Jonathan Singer
@@ -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`
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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
|
||||
@@ -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 <token>``."""
|
||||
|
||||
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
|
||||
@@ -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
|
||||
@@ -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<routed:list_projects>"
|
||||
|
||||
|
||||
@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"]
|
||||
Reference in New Issue
Block a user