Show errored MCP tool calls as failed in the TUI

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": "github",
+    "transport": "http",
+    "url": "https://api.githubcopilot.com/mcp/",
+    "auth": { "kind": "bearer", "token": "your-token" },
+    "allowed_tools": ["list_issues"]
+  }
+]
+```
+
+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`

docs/docs.json

@@ -47,7 +47,8 @@
             "pages": [
               "integrations/github-actions",
               "integrations/ci-cd",
-              "integrations/coding-agents"
+              "integrations/coding-agents",
+              "integrations/mcp"
             ]
           },
           {

docs/integrations/mcp.mdx

@@ -0,0 +1,113 @@
+---
+title: "MCP Servers"
+description: "Connect your own MCP servers and expose their tools to the agent"
+---
+
+Strix can connect to [Model Context Protocol (MCP)](https://modelcontextprotocol.io) servers you list and expose their tools to the agent during a run. Use this to give the agent extra capabilities — reading files, querying an issue tracker, or any other tool a server offers.
+
+## Setup
+
+Create the file `~/.strix/mcp-servers.json`. It holds a JSON list of the servers you want the agent to reach. Each entry is either a local `stdio` server that Strix launches as a subprocess, or a remote `http` server.
+
+Create the directory if it does not exist, then write the file:
+
+```bash
+mkdir -p ~/.strix
+```
+
+Paste the servers you want into `~/.strix/mcp-servers.json`. The example below shows one of each transport — a local filesystem server over `stdio` and a remote GitHub server over `http` with a bearer token:
+
+```json
+[
+  {
+    "name": "local_fs",
+    "transport": "stdio",
+    "command": "npx",
+    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"]
+  },
+  {
+    "name": "github",
+    "transport": "http",
+    "url": "https://api.githubcopilot.com/mcp/",
+    "auth": { "kind": "bearer", "token": "your-token" },
+    "allowed_tools": ["list_issues"]
+  }
+]
+```
+
+Strix reads this file at the start of each run. There is no default file, so no MCP tools are loaded until you create it. Edit `command`, `args`, `url`, and `token` to match your own servers.
+
+## Fields
+
+<ParamField path="name" type="string" required>
+  A short label for the connection. Each server's tools are namespaced by
+  `name` (for example `local_fs.read_file`), so two servers can offer the same
+  tool name without colliding.
+</ParamField>
+
+<ParamField path="transport" type="string">
+  `stdio` for a local subprocess server, or `http` for a remote server.
+</ParamField>
+
+<ParamField path="command" type="string">
+  For `stdio` servers: the executable Strix launches (for example `npx`).
+</ParamField>
+
+<ParamField path="args" type="array">
+  For `stdio` servers: the arguments passed to `command`.
+</ParamField>
+
+<ParamField path="url" type="string">
+  For `http` servers: the server endpoint URL.
+</ParamField>
+
+<ParamField path="auth" type="object">
+  For `http` servers that need a bearer token:
+  `{ "kind": "bearer", "token": "your-token" }`.
+</ParamField>
+
+<ParamField path="allowed_tools" type="array">
+  Restrict which tools the agent can call. Omit it to expose every tool the
+  server offers, or set it to a list of tool names to allow only those.
+</ParamField>
+
+<ParamField path="notes" type="string">
+  Free-text notes for the agent about what this connection is and how you want
+  it used, for example "Staging analytics database, read-only, prefer aggregate
+  queries." When set, the notes are given to the agent at the start of the run
+  as a description of the connection.
+</ParamField>
+
+## Choosing connections per run
+
+By default every connection in the file is used on each run. To narrow it for a
+single run without editing the file, use either flag (both repeatable):
+
+```bash
+strix --mcp-server github -t ...        # use only the named connection(s)
+strix --mcp-exclude staging-db -t ...   # use everything except the named one(s)
+```
+
+`--mcp-server` keeps only the connections you name; `--mcp-exclude` drops the
+ones you name. Connection names must be unique in the file; if two entries share
+a name, the first is kept and the rest are ignored.
+
+## Pointing at a different file
+
+To read the config from another path instead of `~/.strix/mcp-servers.json`, either pass `--mcp-config <path>` on the command line:
+
+```bash
+strix --mcp-config ./mcp-servers.json -t ...
+```
+
+or set the `STRIX_MCP_CONFIG` environment variable to that path. The flag takes precedence when both are given.
+
+## Startup confirmation
+
+When servers are configured, Strix prints a one-line summary at scan startup, for example `MCP: connected 1 server (14 tools): local_fs`, so you can confirm your servers connected.
+
+## Behavior
+
+- The config file is optional. Without it, a run simply gets no MCP tools.
+- A server that fails to connect is skipped and logged, and the run continues without it.
+- A single malformed entry is skipped without blocking the valid ones.

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"]

strix/config/loader.py

@@ -54,16 +54,12 @@ def apply_config_override(path: Path) -> None:
 
 
 def persist_current() -> None:
-    """Merge currently-set env vars into the active config file (0o600).
-
-    Keys already on disk are preserved (including keys unknown to this
-    version's schema); env vars win on conflicts.
-    """
+    """Write currently-set env vars to the active config file (0o600)."""
     s = load_settings()
     target = _override or _DEFAULT_PATH
     target.parent.mkdir(parents=True, exist_ok=True)
 
-    env_block: dict[str, Any] = _read_persisted_env(target)
+    env_block: dict[str, str] = {}
     for sub_name in s.model_fields:
         sub_model = getattr(s, sub_name)
         if not isinstance(sub_model, BaseModel):
@@ -78,24 +74,6 @@ def persist_current() -> None:
     write_secret_text(target, json.dumps({"env": env_block}, indent=2))
 
 
-def _read_persisted_env(path: Path) -> dict[str, Any]:
-    """Read the ``{"env": {...}}`` block already stored at ``path``."""
-    if not path.exists():
-        return {}
-    try:
-        data = json.loads(path.read_text(encoding="utf-8"))
-    except (json.JSONDecodeError, OSError):
-        return {}
-    env_block = data.get("env", {}) if isinstance(data, dict) else {}
-    if not isinstance(env_block, dict):
-        return {}
-    return {
-        str(key).upper(): value
-        for key, value in env_block.items()
-        if value is not None and value != ""
-    }
-
-
 def _aliases_for(finfo: FieldInfo) -> list[str]:
     """Collect every env-var name that should populate ``finfo``."""
     aliases: list[str] = []

strix/core/runner.py

@@ -51,10 +51,12 @@ from strix.tools.output_store import (
 
 
 if TYPE_CHECKING:
+    from agents.mcp import MCPServer
     from agents.memory import SQLiteSession
     from agents.result import RunResultBase
 
     from strix.runtime.status import StatusSink
+    from strix.tools.mcp import ConnectedMcpServer
 
 
 logger = logging.getLogger(__name__)
@@ -62,6 +64,33 @@ logger = logging.getLogger(__name__)
 StreamEventSink = Callable[[str, Any], None]
 
 
+def _mcp_startup_summary(connections: list[ConnectedMcpServer]) -> str:
+    """One user-facing line summarizing the MCP servers that connected."""
+    server_count = len(connections)
+    tool_count = sum(c.tool_count for c in connections)
+    servers_word = "server" if server_count == 1 else "servers"
+    tools_word = "tool" if tool_count == 1 else "tools"
+    names = ", ".join(c.name for c in connections)
+    return f"MCP: connected {server_count} {servers_word} ({tool_count} {tools_word}): {names}"
+
+
+def _mcp_connection_notes(connections: list[ConnectedMcpServer]) -> str | None:
+    """A block describing the connections the user left notes on, for the agent.
+
+    Only connections with notes are listed, so the note describes the connection
+    once rather than being repeated onto every tool. Returns ``None`` when no
+    connection has notes.
+    """
+    noted = [(c.name, c.notes) for c in connections if c.notes]
+    if not noted:
+        return None
+    lines = "\n".join(f"- `{name}.*` tools: {notes}" for name, notes in noted)
+    return (
+        "The user connected these MCP servers for this run and left notes on how "
+        f"to use each:\n{lines}"
+    )
+
+
 def _merge_root_prompt_context(
     scope_context: dict[str, Any],
     extra_system_prompt_context: dict[str, Any] | None,
@@ -253,6 +282,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 +328,24 @@ 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. Fail-open: a missing
+        # config, or a server that will not connect, must never break a run.
+        from strix.tools.mcp import connect_mcp_servers, load_user_mcp_configs
+
+        try:
+            user_mcp_configs = load_user_mcp_configs()
+            if user_mcp_configs:
+                connections = await connect_mcp_servers(user_mcp_configs)
+                mcp_servers = [c.server for c in connections]
+                if connections:
+                    report(_mcp_startup_summary(connections))
+                    notes_block = _mcp_connection_notes(connections)
+                    if notes_block:
+                        root_task = f"{root_task}\n\n{notes_block}"
+        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 +520,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:

strix/interface/cli_args.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import argparse
+import os
 import sys
 from pathlib import Path
 
@@ -219,6 +220,30 @@ Examples:
         help="Path to a custom config file (JSON) to use instead of ~/.strix/cli-config.json",
     )
 
+    parser.add_argument(
+        "--mcp-config",
+        type=str,
+        metavar="PATH",
+        help="Path to an MCP servers JSON file to use instead of ~/.strix/mcp-servers.json.",
+    )
+
+    parser.add_argument(
+        "--mcp-server",
+        dest="mcp_server",
+        action="append",
+        metavar="NAME",
+        help="Use only this MCP connection for the run, by its config name "
+        "(repeatable). Every other configured connection is skipped.",
+    )
+
+    parser.add_argument(
+        "--mcp-exclude",
+        dest="mcp_exclude",
+        action="append",
+        metavar="NAME",
+        help="Skip this MCP connection for the run, by its config name (repeatable).",
+    )
+
     parser.add_argument(
         "--max-budget",
         "--max-budget-usd",
@@ -267,6 +292,20 @@ Examples:
     if args.config:
         apply_config_override(validate_config_file(args.config))
 
+    if args.mcp_config:
+        mcp_config_path = Path(args.mcp_config).expanduser()
+        if not mcp_config_path.is_file():
+            parser.error(f"--mcp-config file not found: {args.mcp_config}")
+        # The MCP loader reads this env var as its config-path override, so
+        # setting it here makes the flag win over the default location.
+        os.environ["STRIX_MCP_CONFIG"] = str(mcp_config_path)
+
+    # The MCP loader reads these as its per-run include/exclude selection.
+    if args.mcp_server:
+        os.environ["STRIX_MCP_ONLY"] = ",".join(args.mcp_server)
+    if args.mcp_exclude:
+        os.environ["STRIX_MCP_EXCLUDE"] = ",".join(args.mcp_exclude)
+
     if args.update:
         sys.exit(0 if self_update() else 1)
 

strix/interface/tui/internal/render/registry.go

@@ -22,19 +22,20 @@ func statusIcon(status string) (string, lipgloss.Style) {
 	return "○ Unknown", Dim()
 }
 
-// renderGenericTool ports registry._render_default_tool_widget.
-func renderGenericTool(name string, args map[string]any, result any, status string) string {
+// renderGenericTool ports registry._render_default_tool_widget. It shows the
+// tool name, its arguments, and a status line only. The raw result is
+// deliberately not rendered: a generic/MCP result (e.g. a multi-kilobyte JSON
+// payload from a database query tool) is noise on screen, and the agent narrates
+// what it got in its next message. The full result still lives in the event
+// data, the run log, and the `strix view` viewer.
+func renderGenericTool(name string, args map[string]any, status string) string {
 	var b strings.Builder
 	b.WriteString(Dim().Render("→ Using tool ") + Bold(Blue).Render(name) + "\n")
 	for _, k := range SortedKeys(args) {
 		b.WriteString("  " + Dim().Render(k) + ": " + StringValue(args[k]) + "\n")
 	}
-	if (status == "completed" || status == "failed" || status == "error") && result != nil {
-		b.WriteString(lipgloss.NewStyle().Bold(true).Render("Result: ") + StringValue(result))
-	} else {
-		icon, style := statusIcon(status)
-		b.WriteString(style.Render(icon))
-	}
+	icon, style := statusIcon(status)
+	b.WriteString(style.Render(icon))
 	return b.String()
 }
 
@@ -87,7 +88,7 @@ func Tool(data map[string]any) string {
 	case "list_requests", "view_request", "repeat_request", "list_sitemap", "view_sitemap_entry", "scope_rules":
 		return renderProxyTool(name, args, result, status)
 	}
-	return renderGenericTool(name, args, result, status)
+	return renderGenericTool(name, args, status)
 }
 
 // ---------------------------------------------------------------------------

strix/interface/tui/internal/render/render_test.go

@@ -203,7 +203,7 @@ func TestToolDispatchCoversKnownTools(t *testing.T) {
 		{
 			"unknown tool falls back to generic",
 			tool("brand_new_tool", map[string]any{"alpha": "1"}, "done", "completed"),
-			[]string{"brand_new_tool", "alpha", "Result:", "done"},
+			[]string{"brand_new_tool", "alpha", "Done"},
 		},
 	}
 
@@ -214,6 +214,18 @@ func TestToolDispatchCoversKnownTools(t *testing.T) {
 	}
 }
 
+func TestGenericToolOmitsRawResult(t *testing.T) {
+	// The generic/MCP renderer shows tool name, args, and a status line only —
+	// never the raw result payload.
+	long := strings.Repeat("x", 5000)
+	out := ansi.Strip(Tool(tool("db_query", map[string]any{"query": "select 1"}, long, "completed")))
+
+	requireContains(t, out, "db_query", "query", "Done")
+	if strings.Contains(out, "Result:") || strings.Contains(out, strings.Repeat("x", 20)) {
+		t.Fatalf("generic result body must not be rendered:\n%s", out)
+	}
+}
+
 func TestCollapseToolShellPreviewAndExpand(t *testing.T) {
 	lines := make([]string, 16)
 	for i := range lines {

strix/tools/mcp/__init__.py

@@ -0,0 +1,21 @@
+"""Generic MCP client: connect MCP servers and expose their tools."""
+
+from __future__ import annotations
+
+from strix.tools.mcp.client import ConnectedMcpServer, connect_mcp_servers
+from strix.tools.mcp.config import (
+    BearerAuth,
+    McpAuth,
+    McpConnectionConfig,
+)
+from strix.tools.mcp.loader import load_user_mcp_configs
+
+
+__all__ = [
+    "BearerAuth",
+    "ConnectedMcpServer",
+    "McpAuth",
+    "McpConnectionConfig",
+    "connect_mcp_servers",
+    "load_user_mcp_configs",
+]

strix/tools/mcp/client.py

@@ -0,0 +1,348 @@
+"""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 contextlib
+import json
+import logging
+from typing import TYPE_CHECKING, Any, NamedTuple, 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
+
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from agents.tool import FunctionTool, Tool
+    from mcp.types import Tool as MCPTool
+
+    from strix.tools.mcp.config import McpConnectionConfig
+
+    # 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__)
+
+
+class ConnectedMcpServer(NamedTuple):
+    """One successfully connected MCP server and how many tools it registered.
+
+    ``server`` is kept so the caller can clean it up when the run ends;
+    ``name`` and ``tool_count`` let the caller show the user a startup summary;
+    ``notes`` carries the connection's optional free-text description so the
+    caller can surface it to the agent as context about the connection.
+    """
+
+    server: MCPServer
+    name: str
+    tool_count: int
+    notes: str | None = None
+
+
+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 {}
+    return {"Authorization": f"Bearer {auth.token}"}
+
+
+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.
+
+    The SDK builds the tool (so name override, input schema, approval policy,
+    error-as-result handling, and tool-origin metadata are unchanged). With a
+    ``result_transform`` we route the underlying MCP call through
+    :func:`_install_result_transform` so the transform sees the structured result
+    and decides the tool's output. Without one (the stock path), we still route
+    the call, through :func:`_install_error_status_capture`, so an errored result
+    reads as failed in the TUI while the agent's content is unchanged.
+    """
+    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)
+    else:
+        _install_error_status_capture(tool, server, mcp_tool.name, namespaced_name)
+    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)
+
+    _replace_tool_invoke(tool, _invoke)
+
+
+def _replace_tool_invoke(tool: FunctionTool, invoke: Callable[[Any, str], Any]) -> None:
+    """Swap a FunctionTool's inner invoke, failing loudly if the SDK shape changed.
+
+    ``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 own try/except. Swapping that inner impl keeps the SDK's
+    error-as-result handling and every piece of tool metadata intact. It is a
+    plain object with the coroutine as an attribute, not a function, so we treat
+    it as untyped to swap it. If the SDK ever renames that attribute we raise
+    rather than silently leave the swap un-applied.
+    """
+    invoker = cast("Any", tool.on_invoke_tool)
+    if not hasattr(invoker, "_invoke_tool_impl"):
+        raise RuntimeError(
+            "agents SDK FunctionTool invoker shape changed: cannot swap the tool "
+            "invoke without risking it being silently skipped."
+        )
+    invoker._invoke_tool_impl = invoke
+
+
+def _mcp_result_to_tool_output(server: MCPServer, result: Any) -> Any:
+    """Serialize a ``CallToolResult`` to a tool output, mirroring the agents SDK.
+
+    This reproduces the serialization in ``agents.mcp.util.MCPUtil.invoke_mcp_tool``
+    (structured-content JSON when the server asks for it, otherwise text/image
+    content blocks, unwrapping a single block). Because the stock path now routes
+    its own call, this is what makes the agent see byte-identical content to what
+    the SDK would have produced on its own.
+    """
+    if getattr(server, "use_structured_content", False) and result.structuredContent:
+        return json.dumps(result.structuredContent)
+
+    outputs: list[dict[str, Any]] = []
+    for item in result.content:
+        if item.type == "text":
+            outputs.append({"type": "text", "text": item.text})
+        elif item.type == "image":
+            outputs.append(
+                {"type": "image", "image_url": f"data:{item.mimeType};base64,{item.data}"}
+            )
+        else:
+            outputs.append({"type": "text", "text": str(item.model_dump(mode="json"))})
+    if len(outputs) == 1:
+        return outputs[0]
+    return outputs
+
+
+def _install_error_status_capture(
+    tool: FunctionTool,
+    server: MCPServer,
+    base_tool_name: str,
+    namespaced_name: str,
+) -> None:
+    """Make an errored MCP result read as failed in the TUI, agent content unchanged.
+
+    The stock SDK invoke returns only the text/image tool output and drops the
+    ``CallToolResult.isError`` flag, so the TUI cannot tell an errored MCP call
+    (which it renders as a green "done") from a successful one. We route the call
+    the same way :func:`_install_result_transform` does, read ``isError`` off the
+    full result, and on an error tag the returned output dict with
+    ``success: False``.
+
+    That tag reaches the human-facing status but not the agent. The SDK stores the
+    raw return value on the run item's ``output`` (which the TUI reads to derive a
+    tool's status), but hands the agent the value re-projected through its
+    ToolOutput schema, which keeps only the known ``type``/``text`` fields and
+    drops the extra ``success`` key. So the status flips to failed while the agent
+    still receives exactly the same error content it does today. Non-error calls
+    return the stock output unchanged and keep rendering as done.
+    """
+
+    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)
+        tool_output = _mcp_result_to_tool_output(server, result)
+        if getattr(result, "isError", False) and isinstance(tool_output, dict):
+            return {**tool_output, "success": False}
+        return tool_output
+
+    _replace_tool_invoke(tool, _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[ConnectedMcpServer]:
+    """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 one :class:`ConnectedMcpServer` per server that connected, carrying
+    the SDK server (so the caller can clean it up when the run ends) plus the
+    server name and how many tools it registered (so the caller can show the
+    user a startup summary). Connections that fail are skipped rather than
+    raised.
+    """
+    connected: list[ConnectedMcpServer] = []
+    for config in configs:
+        server: MCPServer | None = None
+        try:
+            server = _build_server(config)
+            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)
+            if server is not None:
+                with contextlib.suppress(Exception):
+                    await server.cleanup()  # type: ignore[no-untyped-call]
+            continue
+        except BaseException:
+            # A cancellation (or other non-Exception failure) mid-connect must not
+            # orphan MCP subprocesses or HTTP sessions. Clean up the server being
+            # connected and every server already connected, then re-raise so the
+            # caller still stops. The runner only receives the list on a clean
+            # return, so on an abnormal exit this function owns the cleanup.
+            if server is not None:
+                with contextlib.suppress(Exception):
+                    await server.cleanup()  # type: ignore[no-untyped-call]
+            for established in connected:
+                with contextlib.suppress(Exception):
+                    await established.server.cleanup()  # type: ignore[no-untyped-call]
+            raise
+
+        logger.info("Connected MCP server %r (%d tools)", config.name, len(tools))
+        connected.append(
+            ConnectedMcpServer(
+                server=server, name=config.name, tool_count=len(tools), notes=config.notes
+            )
+        )
+
+    return connected

strix/tools/mcp/config.py

@@ -0,0 +1,73 @@
+"""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)
+
+
+McpAuth = Annotated[BearerAuth, 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 for the server. 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."""
+
+    notes: str | None = None
+    """Free-text notes for the agent describing what this connection is and how
+    to use it. When set, they are prefixed onto each of the connection's tool
+    descriptions so the agent sees them."""
+
+    @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

strix/tools/mcp/loader.py

@@ -0,0 +1,132 @@
+"""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; without it the run simply gets
+no MCP tools.
+
+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"
+# Per-run selection, set by the --mcp-server / --mcp-exclude CLI flags. Each is a
+# comma-separated list of connection names.
+_ONLY_ENV_VAR = "STRIX_MCP_ONLY"
+_EXCLUDE_ENV_VAR = "STRIX_MCP_EXCLUDE"
+
+
+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 _dedupe_by_name(configs: list[McpConnectionConfig]) -> list[McpConnectionConfig]:
+    """Keep the first connection of each name, dropping later duplicates.
+
+    Names namespace a server's tools (``<name>.<tool>``), so two connections
+    sharing a name would collide and the second's tools would be silently
+    rejected at registration. Drop the duplicate here, with a warning, instead.
+    """
+    seen: set[str] = set()
+    unique: list[McpConnectionConfig] = []
+    for config in configs:
+        if config.name in seen:
+            logger.warning(
+                "Ignoring MCP server %r: another connection already uses that name "
+                "(names must be unique because they namespace the server's tools).",
+                config.name,
+            )
+            continue
+        seen.add(config.name)
+        unique.append(config)
+    return unique
+
+
+def _parse_names(env_var: str) -> set[str]:
+    return {name.strip() for name in os.environ.get(env_var, "").split(",") if name.strip()}
+
+
+def _apply_run_selection(configs: list[McpConnectionConfig]) -> list[McpConnectionConfig]:
+    """Restrict this run's connections to an optional include/exclude selection.
+
+    ``STRIX_MCP_ONLY`` (if set) keeps only the named connections; then
+    ``STRIX_MCP_EXCLUDE`` drops any named connection. With neither set, every
+    connection is kept.
+    """
+    only = _parse_names(_ONLY_ENV_VAR)
+    exclude = _parse_names(_EXCLUDE_ENV_VAR)
+    if not only and not exclude:
+        return configs
+
+    available = {config.name for config in configs}
+    for name in sorted((only | exclude) - available):
+        logger.warning(
+            "MCP connection selection named %r, which is not configured; ignoring it", name
+        )
+
+    selected: list[McpConnectionConfig] = []
+    for config in configs:
+        if only and config.name not in only:
+            continue
+        if config.name in exclude:
+            continue
+        selected.append(config)
+    return selected
+
+
+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. Connections sharing a name are de-duplicated (first wins), and an
+    optional per-run include/exclude selection is applied last.
+    """
+    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 _apply_run_selection(_dedupe_by_name(configs))

tests/test_cli_mcp_config.py

@@ -0,0 +1,88 @@
+"""Tests for the --mcp-config CLI flag."""
+
+from __future__ import annotations
+
+import importlib
+import os
+import sys
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+cli_main: Any = importlib.import_module("strix.interface.main")
+
+
+def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        cli_main,
+        "load_settings",
+        lambda: SimpleNamespace(runtime=SimpleNamespace(max_local_copy_mb=1024)),
+    )
+
+
+def test_mcp_config_flag_sets_loader_override(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    config = tmp_path / "servers.json"
+    config.write_text("[]", encoding="utf-8")
+    _stub_settings(monkeypatch)
+    # delenv records "originally absent" so monkeypatch removes whatever the
+    # parser sets, keeping the override from leaking into other tests.
+    monkeypatch.delenv("STRIX_MCP_CONFIG", raising=False)
+    monkeypatch.setattr(
+        sys, "argv", ["strix", "-t", "https://test.com/", "-n", "--mcp-config", str(config)]
+    )
+
+    args = cli_main.parse_arguments()
+
+    assert args.mcp_config == str(config)
+    assert os.environ["STRIX_MCP_CONFIG"] == str(config)
+
+
+def test_mcp_config_flag_rejects_missing_file(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    _stub_settings(monkeypatch)
+    monkeypatch.delenv("STRIX_MCP_CONFIG", raising=False)
+    missing = tmp_path / "nope.json"
+    monkeypatch.setattr(
+        sys, "argv", ["strix", "-t", "https://test.com/", "-n", "--mcp-config", str(missing)]
+    )
+
+    with pytest.raises(SystemExit):
+        cli_main.parse_arguments()
+
+    assert "--mcp-config file not found" in capsys.readouterr().err
+
+
+def test_mcp_server_flags_set_selection_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    _stub_settings(monkeypatch)
+    monkeypatch.delenv("STRIX_MCP_ONLY", raising=False)
+    monkeypatch.delenv("STRIX_MCP_EXCLUDE", raising=False)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "strix",
+            "-t",
+            "https://test.com/",
+            "-n",
+            "--mcp-server",
+            "a",
+            "--mcp-server",
+            "b",
+            "--mcp-exclude",
+            "c",
+        ],
+    )
+
+    cli_main.parse_arguments()
+
+    assert os.environ["STRIX_MCP_ONLY"] == "a,b"
+    assert os.environ["STRIX_MCP_EXCLUDE"] == "c"

tests/test_config_loader.py

@@ -208,69 +208,6 @@ def test_persist_current_writes_env_block(tmp_path: Path, monkeypatch: pytest.Mo
     }
 
 
-def test_persist_current_preserves_keys_missing_from_env(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.delenv("STRIX_LLM", raising=False)
-    monkeypatch.delenv("LLM_API_KEY", raising=False)
-    monkeypatch.setenv("OPENAI_API_KEY", "env-key")
-    target = tmp_path / "cli-config.json"
-    target.write_text(
-        json.dumps({"env": {"STRIX_LLM": "file-model", "STRIX_FUTURE_KEY": "keep-me"}}),
-        encoding="utf-8",
-    )
-    loader.apply_config_override(target)
-
-    loader.persist_current()
-
-    assert json.loads(target.read_text(encoding="utf-8")) == {
-        "env": {
-            "STRIX_LLM": "file-model",
-            "STRIX_FUTURE_KEY": "keep-me",
-            "OPENAI_API_KEY": "env-key",
-        }
-    }
-
-
-def test_persist_current_env_wins_over_persisted_value(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setenv("STRIX_LLM", "env-model")
-    target = tmp_path / "cli-config.json"
-    target.write_text(json.dumps({"env": {"STRIX_LLM": "file-model"}}), encoding="utf-8")
-    loader.apply_config_override(target)
-
-    loader.persist_current()
-
-    assert json.loads(target.read_text(encoding="utf-8"))["env"]["STRIX_LLM"] == "env-model"
-
-
-def test_persist_current_preserves_dict_valued_key(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.delenv("STRIX_LLM", raising=False)
-    monkeypatch.delenv("LLM_EXTRA_HEADERS", raising=False)
-    target = tmp_path / "cli-config.json"
-    target.write_text(
-        json.dumps(
-            {
-                "env": {
-                    "STRIX_LLM": "file-model",
-                    "LLM_EXTRA_HEADERS": {"X-Foo": "bar"},
-                }
-            }
-        ),
-        encoding="utf-8",
-    )
-    loader.apply_config_override(target)
-
-    loader.persist_current()
-
-    persisted_env = json.loads(target.read_text(encoding="utf-8"))["env"]
-    assert persisted_env["STRIX_LLM"] == "file-model"
-    assert persisted_env["LLM_EXTRA_HEADERS"] == {"X-Foo": "bar"}
-
-
 def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("STRIX_LLM", "persisted-model")
     target = tmp_path / "cli-config.json"

tests/test_mcp_client.py

@@ -0,0 +1,631 @@
+"""Tests for the generic MCP client: config contract, namespacing, and filtering."""
+
+from __future__ import annotations
+
+import asyncio
+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.core.runner import _mcp_connection_notes
+from strix.interface.tui.live_view import _tool_status_from_result
+from strix.tools.mcp import (
+    BearerAuth,
+    ConnectedMcpServer,
+    McpConnectionConfig,
+    load_user_mcp_configs,
+)
+from strix.tools.mcp import client as mcp_client
+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": "files_main",
+            "transport": "http",
+            "url": "https://mcp.example.com",
+            "auth": {"kind": "bearer", "token": "abc"},
+            "allowed_tools": ["list_files"],
+        }
+    )
+
+    assert isinstance(config.auth, BearerAuth)
+    assert config.auth.token == "abc"
+    assert config.allowed_tools == ["list_files"]
+
+
+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("files_main", []))
+
+    assert headers == {"Authorization": "Bearer run-token"}
+
+
+# --- 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(
+        "files_main",
+        [_mcp_tool("list_files"), _mcp_tool("search")],
+    )
+
+    await _register_server_tools(_config("files_main", ["list_files"]), server)
+
+    names = _registered_names()
+    assert "files_main.list_files" in names
+    assert "files_main.search" 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("files_main", [_mcp_tool("list_files")])
+
+    tools: list[Tool] = await _register_server_tools(
+        _config("files_main", ["list_files"]), 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_files", {})]
+    assert output == {"type": "text", "text": "routed:list_files"}
+
+
+# --- result transform --------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_result_transform_receives_namespaced_name_and_structured_result() -> None:
+    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
+    seen: list[tuple[str, Any]] = []
+
+    def transform(name: str, structured: Any) -> Any:
+        seen.append((name, structured))
+        return {"kept": structured["content"][0]["text"]}
+
+    tools: list[Tool] = await _register_server_tools(
+        _config("files_main", ["list_files"]), 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_files", {})]
+
+    # The transform is called with the namespaced name and the parsed result.
+    assert len(seen) == 1
+    name, structured = seen[0]
+    assert name == "files_main.list_files"
+    # A parsed CallToolResult (dict/list), not a pre-serialized string.
+    assert structured["content"][0]["text"] == "routed:list_files"
+    assert structured["isError"] is False
+
+    # The transform's return value is exactly what the tool yields.
+    assert output == {"kept": "routed:list_files"}
+
+
+@pytest.mark.asyncio
+async def test_result_transform_can_rewrite_the_tool_output() -> None:
+    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
+
+    def transform(_name: str, structured: Any) -> Any:
+        # Keep only a truncated view of the text field.
+        return structured["content"][0]["text"][:6]
+
+    tools: list[Tool] = await _register_server_tools(
+        _config("files_main", ["list_files"]), server, result_transform=transform
+    )
+
+    output = await tools[0].on_invoke_tool(None, "{}")  # type: ignore[union-attr]
+
+    assert output == "routed"
+
+
+@pytest.mark.asyncio
+async def test_without_result_transform_output_is_unchanged() -> None:
+    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
+
+    tools: list[Tool] = await _register_server_tools(
+        _config("files_main", ["list_files"]), 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_files", {})]
+    assert output == {"type": "text", "text": "routed:list_files"}
+
+
+# --- error status capture ----------------------------------------------------
+
+
+class ErroringMCPServer(FakeMCPServer):
+    """A connected server whose calls come back as MCP errors (isError=True)."""
+
+    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"boom:{tool_name}")],
+            isError=True,
+        )
+
+
+@pytest.mark.asyncio
+async def test_errored_mcp_result_is_flagged_failed_for_the_tui() -> None:
+    server = ErroringMCPServer("files_main", [_mcp_tool("list_files")])
+
+    tools: list[Tool] = await _register_server_tools(
+        _config("files_main", ["list_files"]), server
+    )
+    output = await tools[0].on_invoke_tool(None, "{}")  # type: ignore[union-attr]
+
+    # The error text stays exactly what the agent gets today; a success:False tag
+    # rides alongside it purely so the TUI can tell the call apart from a success.
+    assert output == {"type": "text", "text": "boom:list_files", "success": False}
+    assert _tool_status_from_result(output) == "failed"
+
+
+@pytest.mark.asyncio
+async def test_successful_mcp_result_stays_completed_for_the_tui() -> None:
+    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
+
+    tools: list[Tool] = await _register_server_tools(
+        _config("files_main", ["list_files"]), server
+    )
+    output = await tools[0].on_invoke_tool(None, "{}")  # type: ignore[union-attr]
+
+    # A non-error result is untouched and keeps rendering as done.
+    assert output == {"type": "text", "text": "routed:list_files"}
+    assert _tool_status_from_result(output) == "completed"
+
+
+# --- 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("files_main", ["list_files"]))
+
+    assert isinstance(server, MCPServerStreamableHttp)
+    assert server.name == "files_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": "files_main",
+                    "transport": "http",
+                    "url": "https://mcp.example.com",
+                    "auth": {"kind": "bearer", "token": "abc"},
+                    "allowed_tools": ["list_files"],
+                },
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    configs = load_user_mcp_configs(config_file)
+
+    assert [c.name for c in configs] == ["local_fs", "files_main"]
+    assert configs[0].transport == "stdio"
+    assert configs[1].allowed_tools == ["list_files"]
+
+
+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"]
+
+
+# --- connection notes --------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_connection_notes_are_carried_on_the_connection(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    server = FakeMCPServer("db", [_mcp_tool("query")])
+    monkeypatch.setattr(mcp_client, "_build_server", lambda _config: server)
+    config = McpConnectionConfig(
+        name="db",
+        url="https://mcp.example.com",
+        notes="Staging analytics DB; read-only.",
+        allowed_tools=["query"],
+    )
+
+    connections = await mcp_client.connect_mcp_servers([config])
+
+    # Notes ride on the connection (surfaced once), not stapled onto each tool.
+    assert connections[0].notes == "Staging analytics DB; read-only."
+
+
+def test_connection_notes_block_lists_only_noted_connections() -> None:
+    connections = [
+        ConnectedMcpServer(
+            server=FakeMCPServer("db", []), name="db", tool_count=2, notes="staging, read-only"
+        ),
+        ConnectedMcpServer(server=FakeMCPServer("fs", []), name="fs", tool_count=1, notes=None),
+    ]
+
+    block = _mcp_connection_notes(connections)
+
+    assert block is not None
+    assert "db" in block
+    assert "staging, read-only" in block
+    # A connection without notes is not listed.
+    assert "fs" not in block
+
+
+def test_connection_notes_block_is_none_without_notes() -> None:
+    connections = [
+        ConnectedMcpServer(server=FakeMCPServer("db", []), name="db", tool_count=1, notes=None)
+    ]
+
+    assert _mcp_connection_notes(connections) is None
+
+
+# --- cancellation cleanup ----------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_connect_cleans_up_when_cancelled_mid_connect(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    cleaned: list[str] = []
+
+    class _Tracking(FakeMCPServer):
+        def __init__(self, name: str, *, fail_connect: bool = False) -> None:
+            super().__init__(name, [_mcp_tool("t")])
+            self._fail_connect = fail_connect
+
+        async def connect(self) -> None:
+            if self._fail_connect:
+                raise asyncio.CancelledError
+
+        async def cleanup(self) -> None:
+            cleaned.append(self._name)
+
+    servers = {"good": _Tracking("good"), "bad": _Tracking("bad", fail_connect=True)}
+    monkeypatch.setattr(mcp_client, "_build_server", lambda config: servers[config.name])
+
+    configs = [
+        McpConnectionConfig(name="good", url="https://mcp.example.com", allowed_tools=["t"]),
+        McpConnectionConfig(name="bad", url="https://mcp.example.com", allowed_tools=["t"]),
+    ]
+
+    with pytest.raises(asyncio.CancelledError):
+        await mcp_client.connect_mcp_servers(configs)
+
+    # The server being connected when cancelled, and the one already connected,
+    # are both cleaned up rather than orphaned.
+    assert cleaned == ["bad", "good"]
+
+
+# --- duplicate names and run selection ---------------------------------------
+
+
+def _names_file(tmp_path: Path, *names: str) -> Path:
+    config_file = tmp_path / "mcp-servers.json"
+    config_file.write_text(
+        json.dumps([{"name": n, "transport": "stdio", "command": "npx"} for n in names]),
+        encoding="utf-8",
+    )
+    return config_file
+
+
+def test_loader_drops_duplicate_named_connections(tmp_path: Path) -> None:
+    config_file = tmp_path / "mcp-servers.json"
+    config_file.write_text(
+        json.dumps(
+            [
+                {"name": "dup", "transport": "stdio", "command": "first"},
+                {"name": "dup", "transport": "stdio", "command": "second"},
+                {"name": "other", "transport": "stdio", "command": "npx"},
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    configs = load_user_mcp_configs(config_file)
+
+    # Duplicate name is dropped; the first entry wins.
+    assert [c.name for c in configs] == ["dup", "other"]
+    assert configs[0].command == "first"
+
+
+def test_loader_include_selection_keeps_only_named(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    config_file = _names_file(tmp_path, "a", "b", "c")
+    monkeypatch.setenv("STRIX_MCP_ONLY", "a,c")
+
+    configs = load_user_mcp_configs(config_file)
+
+    assert [c.name for c in configs] == ["a", "c"]
+
+
+def test_loader_exclude_selection_drops_named(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    config_file = _names_file(tmp_path, "a", "b", "c")
+    monkeypatch.setenv("STRIX_MCP_EXCLUDE", "b")
+
+    configs = load_user_mcp_configs(config_file)
+
+    assert [c.name for c in configs] == ["a", "c"]