Show errored MCP tool calls as failed in the TUI

README.md

@@ -167,10 +167,15 @@ strix view
 
 # ...or open a specific run by name
 strix view my-run-name
+
+# Expose the viewer on all IPv4 interfaces at a fixed port
+strix view --host 0.0.0.0 --port 8080 --no-open
 ```
 
 `strix view` starts a lightweight local server (bound to `127.0.0.1` on a random port) and opens your browser to a private, tokened link. Nothing leaves your machine: the dashboard reads the run's files straight off disk, with no cloud account or upload required. The UI ships prebuilt with Strix, so there is no extra install and no JS build step.
 
+Use `--host 0.0.0.0` to make the viewer reachable from other machines. Replace `0.0.0.0` in the printed URL with the server's reachable IP or hostname. The token in that URL grants access to the selected run's scan data, history, and steering, so only share it with trusted users and restrict the port with your firewall. Requests without the token-derived session cannot read run data.
+
 ### What's in the dashboard
 
 - **Overview**: run status, target, and a severity breakdown of everything found so far.
@@ -315,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/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.
+// renderGenericTool ports registry._render_default_tool_widget. It shows the
-func renderGenericTool(name string, args map[string]any, result any, status string) string {
+// 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 {
+	icon, style := statusIcon(status)
-		b.WriteString(lipgloss.NewStyle().Bold(true).Render("Result: ") + StringValue(result))
+	b.WriteString(style.Render(icon))
-	} else {
-		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/interface/viewer/cli.py

@@ -45,7 +45,11 @@ def run_view(argv: list[str]) -> None:
         default=0,
         help="Port to serve on (default: an available ephemeral port).",
     )
-    parser.add_argument("--host", default="127.0.0.1", help=argparse.SUPPRESS)
+    parser.add_argument(
+        "--host",
+        default="127.0.0.1",
+        help="Host to bind to (default: 127.0.0.1; use 0.0.0.0 for all IPv4 interfaces).",
+    )
     parser.add_argument(
         "--no-open",
         action="store_true",

strix/interface/viewer/server.py

@@ -135,8 +135,9 @@ class _ViewerState:
         # exchanged for a session cookie only when presented on the initial page
         # load. It is the request-level authorization the review asked for:
         # reachability of the port (e.g. when bound with ``--host``) is not
-        # enough to steer a live scan, trigger a report, or browse history --
+        # enough to read run data, steer a live scan, trigger a report, or
-        # the token is never handed to a caller who merely reaches ``/``.
+        # browse history -- the token is never handed to a caller who merely
+        # reaches ``/``.
         self.session_token = secrets.token_urlsafe(32)
         # Finalized in ``serve()`` once the port is known (the server binds
         # after this state is constructed); see SESSION_COOKIE_PREFIX.
@@ -234,11 +235,11 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             self.end_headers()
 
         def _handle_api(self, path: str, query: dict[str, list[str]]) -> None:
-            # The launched run is always viewable with no verification. The
+            # The cross-run history list (/api/runs) unlocks its entries only for
-            # cross-run history list (/api/runs) unlocks its entries only for a
+            # a caller that holds this process's session capability *and* is
-            # caller that holds this process's session capability *and* is email
+            # email verified, so merely reaching an exposed --host port never
-            # verified, so merely reaching an exposed --host port never leaks the
+            # leaks the run list (the payload still advertises the count as a
-            # run list (the payload still advertises the count as a teaser).
+            # teaser).
             if path == "/api/runs":
                 unlocked = self._has_session() and auth.is_verified()
                 payload = build_runs_payload(state.base_dir, verified=unlocked)
@@ -253,6 +254,13 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
                 self._handle_auth_status()
                 return
 
+            # All remaining GET endpoints expose run metadata or scan output.
+            # Require the capability even for the run used to launch the viewer;
+            # reachability of an exposed --host port must not grant data access.
+            if not self._has_session():
+                self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
+                return
+
             run_values = query.get("run")
             run_param = run_values[0] if run_values else None
             run_dir = resolve_run_dir(state.base_dir, run_param, state.run_dir)
@@ -260,18 +268,12 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
                 self._send_json(HTTPStatus.NOT_FOUND, {"error": "unknown run"})
                 return
 
-            # The launched run is always viewable. Any *other* run's data is part
+            # Any run other than the one used to launch the viewer is part of the
-            # of the gated history: it needs this process's session capability
+            # email-gated history. The session check above applies to both paths;
-            # (so merely reaching an exposed --host port is not enough) *and*
+            # verification adds a second gate for historical run data.
-            # email verification -- otherwise knowing a run name would leak its
+            if run_dir.resolve() != state.run_dir.resolve() and not auth.is_verified():
-            # metadata, vulnerabilities, report, and transcript.
+                self._send_json(HTTPStatus.UNAUTHORIZED, {"error": "unverified"})
-            if run_dir.resolve() != state.run_dir.resolve():
+                return
-                if not self._has_session():
-                    self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
-                    return
-                if not auth.is_verified():
-                    self._send_json(HTTPStatus.UNAUTHORIZED, {"error": "unverified"})
-                    return
 
             if path == "/api/run":
                 self._send_json(HTTPStatus.OK, read_run_summary(run_dir))
@@ -385,7 +387,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             except auth.RelayError as exc:
                 self._send_relay_error(exc)
                 return
-            # The password is returned only to the local (127.0.0.1) browser.
+            # The password is returned only to a session-authorized browser.
             self._send_json(
                 HTTPStatus.OK,
                 {"ok": True, "password": password, "filename": filename},

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

tests/test_viewer.py

@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING
 from urllib.parse import urlsplit
 
 from strix.core.paths import latest_run_dir, runs_base_dir
+from strix.interface.viewer.cli import run_view
 from strix.interface.viewer.server import serve
 from strix.interface.viewer.transcript import (
     build_run_state,
@@ -48,6 +49,31 @@ def test_latest_run_dir_none_when_no_runs(tmp_path: Path, monkeypatch: pytest.Mo
     assert runs_base_dir() == tmp_path / "strix_runs"
 
 
+def test_view_cli_help_includes_host(capsys: pytest.CaptureFixture[str]) -> None:
+    try:
+        run_view(["--help"])
+    except SystemExit as exc:
+        assert exc.code == 0
+    else:
+        raise AssertionError("--help should exit")
+
+    help_text = capsys.readouterr().out
+    assert "--host HOST" in help_text
+    assert "0.0.0.0" in help_text
+
+
+def test_server_can_bind_all_ipv4_interfaces(tmp_path: Path) -> None:
+    run_dir = _make_run(tmp_path, "remote", status="running", end_time=None)
+
+    httpd, url, _ = serve(run_dir, host="0.0.0.0", open_browser=False)
+    try:
+        assert httpd.server_address[0] == "0.0.0.0"
+        assert url == f"http://0.0.0.0:{httpd.server_address[1]}"
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
 def test_latest_run_dir_picks_newest_by_record_mtime(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
@@ -173,14 +199,15 @@ def test_server_serves_api_and_static(tmp_path: Path, monkeypatch: pytest.Monkey
     (assets / "assets" / "app.js").write_text("console.log(1)", encoding="utf-8")
     monkeypatch.setattr("strix.interface.viewer.server.bundle_dir", lambda: assets)
 
-    httpd, url, _ = serve(run_dir, open_browser=False)
+    httpd, url, token = serve(run_dir, open_browser=False)
     try:
-        status, ctype, body = _get(f"{url}/api/run")
+        cookie = _session_cookie(url, token)
+        status, ctype, body = _get(f"{url}/api/run", cookie=cookie)
         assert status == 200
         assert "application/json" in ctype
         assert json.loads(body)["finished"] is True
 
-        status, _, body = _get(f"{url}/api/transcript")
+        status, _, body = _get(f"{url}/api/transcript", cookie=cookie)
         assert {a["id"] for a in json.loads(body)["agents"]} == {"root", "child"}
 
         # Real asset is served.
@@ -429,6 +456,22 @@ def test_unauthorized_client_cannot_acquire_capability(
         httpd.server_close()
 
 
+def test_run_data_requires_session(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    run_dir = _make_run(tmp_path, "private", status="completed", end_time="2026-01-01T00:00:00Z")
+    _bundle(tmp_path, monkeypatch)
+
+    httpd, url, token = serve(run_dir, open_browser=False)
+    try:
+        cookie = _session_cookie(url, token)
+        for path in ("/api/run", "/api/vulnerabilities", "/api/report", "/api/transcript"):
+            assert _get_status(url + path) == 403, path
+            assert _get_status(url + path, cookie=f"{_cookie_name(url)}=wrong") == 403, path
+            assert _get_status(url + path, cookie=cookie) == 200, path
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
 def test_auth_status_reflects_expiry(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     run_dir = _make_run(tmp_path, "status", status="running", end_time=None)
     _bundle(tmp_path, monkeypatch)
@@ -561,11 +604,10 @@ def test_historical_run_data_requires_verification(
 
     httpd, url, token = serve(launched, open_browser=False)
     try:
-        # The launched run is always viewable, no verification and no cookie.
+        # The launched run needs the session capability, but not email verification.
-        status, _, _ = _get(f"{url}/api/run")
+        assert _get_status(f"{url}/api/run") == 403
-        assert status == 200
-
         cookie = _session_cookie(url, token)
+        assert _get_status(f"{url}/api/run", cookie=cookie) == 200
 
         # A different run needs the session capability first: a cookie-less
         # caller is forbidden even once the machine is verified.