name the run's MCP connections in the prompt again, alongside list_mcps

pyproject.toml

@@ -243,6 +243,8 @@ ignore = [
 "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"]
+# MCP connection request in a test carries a dummy bearer token.
+"tests/test_runner_root_prompt.py" = ["S106"]
 # 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/agents/factory.py

@@ -29,6 +29,7 @@ from strix.tools.agents_graph.tools import (
 from strix.tools.coverage.tools import list_coverage, record_coverage, update_coverage
 from strix.tools.finish.tool import finish_scan
 from strix.tools.load_skill.tool import load_skill
+from strix.tools.mcp import call_mcp, describe_mcp, list_mcps
 from strix.tools.notes.tools import (
     create_note,
     delete_note,
@@ -587,6 +588,9 @@ _BASE_TOOLS: tuple[Tool, ...] = (
     list_sitemap,
     view_sitemap_entry,
     scope_rules,
+    list_mcps,
+    describe_mcp,
+    call_mcp,
     view_agent_graph,
     send_message_to_agent,
     wait_for_agents,

strix/agents/prompts/system_prompt.jinja

@@ -75,6 +75,22 @@ AUTHORIZED TARGETS:
 {% endfor %}
 {% endif %}
 
+{% if system_prompt_context and system_prompt_context.mcp_available %}
+MCP CONNECTIONS (available this run):
+- The user connected one or more MCP (Model Context Protocol) servers — external tool providers you can reach on demand. Their individual tools do NOT appear in your tool list; three dispatch tools are the only way in.
+{% if system_prompt_context.mcp_connections %}
+- Connected this run (call describe_mcp on one to see its tools):
+{% for connection in system_prompt_context.mcp_connections %}
+  - {{ connection.name }} ({{ connection.tool_count }} tools){% if connection.purpose %}: {{ connection.purpose }}{% endif %}
+{% endfor %}
+{% endif %}
+- Reach for a connection whenever the target itself cannot give you information a connection could: its database schema and access policies, real deployment or infrastructure configuration, known issues or prior findings, or server logs. In those cases call list_mcps early to see what is available, and prefer a connection's authoritative data over inferring from the target's responses. Do not wait to be told a connection exists.
+  1. Call list_mcps() to discover the available connections.
+  2. Call describe_mcp(connection="<name>") to inspect one connection's tools, each with its name, description, and JSON input schema.
+  3. Call call_mcp(connection="<name>", tool="<tool>", arguments={...}) to run one, passing an arguments object that matches the schema (omit arguments for a tool that takes none).
+- Do not assume a connection or tool exists; discover it with list_mcps and describe it with describe_mcp before calling.
+{% endif %}
+
 AUTHORIZATION STATUS:
 - You have FULL AUTHORIZATION for authorized security validation on in-scope targets to help secure the target systems/app
 - All permission checks have been COMPLETED and APPROVED - never question your authority

strix/core/runner.py

@@ -58,7 +58,7 @@ if TYPE_CHECKING:
     from agents.result import RunResultBase
 
     from strix.runtime.status import StatusSink
-    from strix.tools.mcp import ConnectedMcpServer
+    from strix.tools.mcp import ConnectedMcpServer, McpConnectionRequest
 
 
 logger = logging.getLogger(__name__)
@@ -91,23 +91,6 @@ def _record_mcp_connections(connections: list[ConnectedMcpServer]) -> None:
     report_state.record_mcp_connections([connection.name for connection in connections])
 
 
-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,
@@ -173,6 +156,7 @@ async def run_strix_scan(
     root_instructions_override: str | None = None,
     extra_system_prompt_context: dict[str, Any] | None = None,
     status_sink: StatusSink | None = None,
+    mcp_connection_requests: list[McpConnectionRequest] | None = None,
 ) -> RunResultBase | None:
     """Run or resume one Strix scan against a sandbox.
 
@@ -184,6 +168,11 @@ async def run_strix_scan(
     ``extra_system_prompt_context`` is merged into the root agent's scan
     context before prompt rendering. Child agents keep the standard scan prompt
     and context.
+    ``mcp_connection_requests`` supplies the run's MCP connections from any
+    source: when given, the engine connects those requests; when ``None`` (the
+    command-line default) it reads ``~/.strix/mcp-servers.json`` itself. Either
+    way the engine does the connecting, so the caller passes inert configs plus
+    metadata and never live sessions.
     """
 
     def report(phase: str) -> None:
@@ -344,6 +333,61 @@ async def run_strix_scan(
             coordinator.set_budget_extender(hooks.extend_budget)
 
         scope_context = build_scope_context(scan_config)
+
+        # Attach the run's MCP connections and hold their live sessions in a
+        # per-run registry. The connections are source-agnostic: a caller
+        # (the SaaS/pro product) can supply them as mcp_connection_requests, and
+        # when it does not the command-line path reads them from
+        # ~/.strix/mcp-servers.json here. Either way one shared engine routine
+        # does the connecting and populating. Nothing is registered as an agent
+        # tool: every agent reaches these connections on demand through the
+        # list_mcps / describe_mcp / call_mcp tools, guided by brief static prompt
+        # guidance when any connection exists. Fail-open: a missing config, or a
+        # server that will not connect, must never break a run.
+        from strix.tools.mcp import (
+            McpConnectionRequest,
+            McpRegistry,
+            attach_mcp_requests,
+            load_user_mcp_configs,
+        )
+
+        mcp_registry = McpRegistry()
+        try:
+            if mcp_connection_requests is None:
+                # Command-line default: read the user's file and wrap each config
+                # in a bare request (no provider or transform), so this path is
+                # exactly the old behavior.
+                mcp_requests = [
+                    McpConnectionRequest(config=config) for config in load_user_mcp_configs()
+                ]
+            else:
+                mcp_requests = mcp_connection_requests
+            if mcp_requests:
+                connections = await attach_mcp_requests(mcp_requests, mcp_registry)
+                mcp_servers = [c.server for c in connections]
+                # Recorded even when nothing connected, so a resumed run does not
+                # keep attributing tool calls to servers it no longer has.
+                _record_mcp_connections(connections)
+                if connections:
+                    report(_mcp_startup_summary(connections))
+                    # Name the connected servers in the prompt so every agent
+                    # (root and children, both deriving from scope_context) sees
+                    # what is available at the start; they can still re-list or
+                    # inspect them at run time via list_mcps / describe_mcp. Set
+                    # only when a connection exists, so a run with no MCP leaves
+                    # the prompt context unchanged.
+                    scope_context["mcp_available"] = bool(mcp_registry)
+                    scope_context["mcp_connections"] = [
+                        {
+                            "name": summary.name,
+                            "purpose": summary.purpose,
+                            "tool_count": summary.tool_count,
+                        }
+                        for summary in mcp_registry.summaries()
+                    ]
+        except Exception:
+            logger.exception("Failed to connect user MCP servers; continuing without them")
+
         root_context = _merge_root_prompt_context(scope_context, extra_system_prompt_context)
         root_instructions = _compose_root_instructions_override(
             root_instructions_override,
@@ -355,27 +399,6 @@ 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]
-                # Recorded even when nothing connected, so a resumed run does not
-                # keep attributing tool calls to servers it no longer has.
-                _record_mcp_connections(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,
@@ -427,6 +450,7 @@ async def run_strix_scan(
             "coordinator": coordinator,
             "sandbox_session": bundle["session"],
             "caido_client": bundle["caido_client"],
+            "mcp_registry": mcp_registry,
             "agent_id": root_id,
             "parent_id": None,
             "interactive": interactive,

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

@@ -33,3 +33,15 @@ func renderMcpTool(connection, toolName string, args map[string]any, status stri
 	b.WriteString(style.Render(icon))
 	return b.String()
 }
+
+// renderMcpInspect renders describe_mcp: a request to inspect one connection's
+// catalog rather than a call to a tool on it. There is no underlying tool, so
+// the connection is the whole subject and leads. Same icon and colors as a tool
+// call so the two read as one family while scrolling a transcript.
+func renderMcpInspect(connection, status string) string {
+	var b strings.Builder
+	b.WriteString(mcpIcon + Dim().Render("Inspecting MCP server ") + Bold(Mint).Render(connection) + "\n")
+	icon, style := statusIcon(status)
+	b.WriteString(style.Render(icon))
+	return b.String()
+}

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

@@ -57,6 +57,11 @@ func Tool(data map[string]any) string {
 	// nothing here. The tag is only ever set from the connections the run made,
 	// so it is the one thing that can tell such a call apart from a built-in.
 	if connection := StringValue(data["mcp_connection"]); connection != "" {
+		// describe_mcp inspects a connection's catalog rather than calling a tool
+		// on it, so there is no underlying tool and the connection is the subject.
+		if name == "describe_mcp" {
+			return renderMcpInspect(connection, status)
+		}
 		toolName := StringValue(data["mcp_tool"])
 		if toolName == "" {
 			toolName = name

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

@@ -227,7 +227,9 @@ func TestGenericToolOmitsRawResult(t *testing.T) {
 }
 
 func TestMcpToolLeadsWithActionAndNamesTheServer(t *testing.T) {
-	data := tool("local_fs_read_file", map[string]any{"path": "/etc/hosts"}, "file body", "completed")
+	// call_mcp is the dispatch tool; the connection and the server's own tool
+	// name are tagged onto the event from its arguments.
+	data := tool("call_mcp", map[string]any{"path": "/etc/hosts"}, "file body", "completed")
 	data["mcp_connection"] = "local_fs"
 	data["mcp_tool"] = "read_file"
 
@@ -244,11 +246,26 @@ func TestMcpToolLeadsWithActionAndNamesTheServer(t *testing.T) {
 	}
 }
 
-func TestMcpToolWithoutTaggedNameFallsBackToFullName(t *testing.T) {
+func TestMcpToolWithoutTaggedToolFallsBackToDispatchName(t *testing.T) {
-	data := tool("local_fs_read_file", nil, nil, "running")
+	// A call_mcp whose underlying tool could not be read still renders as an MCP
+	// row, falling back to the dispatch tool name.
+	data := tool("call_mcp", nil, nil, "running")
 	data["mcp_connection"] = "local_fs"
 
-	requireContains(t, ansi.Strip(Tool(data)), "local_fs_read_file", "In progress")
+	requireContains(t, ansi.Strip(Tool(data)), mcpIcon+"call_mcp", "local_fs", "In progress")
+}
+
+func TestMcpDescribeInspectsConnection(t *testing.T) {
+	// describe_mcp inspects a connection; the connection is the subject and the
+	// dispatch tool name is not shown as if it were a server tool.
+	data := tool("describe_mcp", nil, nil, "completed")
+	data["mcp_connection"] = "local_fs"
+
+	out := ansi.Strip(Tool(data))
+	requireContains(t, out, mcpIcon, "Inspecting MCP server", "local_fs", "Done")
+	if strings.Contains(out, "describe_mcp") {
+		t.Fatalf("describe_mcp must read as inspecting the connection, not name the dispatch tool:\n%s", out)
+	}
 }
 
 func TestCollapseToolShellPreviewAndExpand(t *testing.T) {

strix/interface/tui/live_view.py

@@ -8,17 +8,13 @@ from typing import TYPE_CHECKING, Any
 
 
 if TYPE_CHECKING:
-    from collections.abc import Iterable
     from pathlib import Path
 
 from agents.tool import ToolOutputImage
 
 from strix.core.paths import runtime_state_dir
 from strix.interface.tui.history import load_session_history
-
+from strix.tools.mcp import resolve_mcp_call
-# Imported from the naming module rather than the mcp package so a projection
-# never pulls in the MCP client and the agents SDK behind it.
-from strix.tools.mcp.naming import resolve_mcp_tool
 
 
 class TuiLiveView:
@@ -31,27 +27,23 @@ class TuiLiveView:
         self._user_instruction: str | None = None
         self._user_instruction_at: str | None = None
         self._user_instruction_shown = False
-        self._mcp_connections: tuple[str, ...] = ()
 
-    def set_mcp_connections(self, names: Iterable[str]) -> None:
+    def _mcp_tool_fields(self, tool_name: str, args: dict[str, Any]) -> dict[str, str]:
-        """The MCP servers this run connected, so its tool calls can name theirs.
-
-        A server's tools are offered to the model under a name built from the
-        connection name and the tool's own name. That name cannot be split back
-        apart on its own, so tool calls are matched against these names instead.
-        """
-        self._mcp_connections = tuple(str(name) for name in names)
-
-    def _mcp_tool_fields(self, tool_name: str) -> dict[str, str]:
         """Event fields naming the MCP server a tool call went out to, if any.
 
-        Empty for every built-in tool, which is what tells an interface to render
+        Delegates to the shared engine resolver :func:`resolve_mcp_call` so a
-        the call as one of its own rather than as a call to a user's server.
+        dispatch call is attributed the same way here and in strix-pro's tracer.
+        The projection has no live registry, so it passes none: it reports the
+        connection and tool read from the call's arguments and leaves the provider
+        out. Empty for every other tool, which is what tells an interface to
+        render the call as one of its own rather than as a call to a user's
+        server. ``describe_mcp`` resolves with an empty tool, which tells both
+        renderers to present the row as inspecting the connection itself.
         """
-        origin = resolve_mcp_tool(tool_name, self._mcp_connections)
+        info = resolve_mcp_call(tool_name, args)
-        if origin is None:
+        if info is None:
             return {}
-        return {"mcp_connection": origin.connection, "mcp_tool": origin.tool}
+        return {"mcp_connection": info.connection, "mcp_tool": info.tool}
 
     def set_user_instruction(self, text: str | None, *, timestamp: str | None = None) -> None:
         """Open the transcript with what the user asked for.
@@ -98,8 +90,7 @@ class TuiLiveView:
 
     def hydrate_from_run_dir(self, run_dir: Path) -> None:
         # Armed before the agents are added so the root agent's arrival puts the
-        # user's opening message ahead of the replayed history, and before the
+        # user's opening message ahead of the replayed history.
-        # history is replayed so its MCP tool calls are attributed too.
         self._load_run_record(run_dir)
         state_dir = runtime_state_dir(run_dir)
         agents_path = state_dir / "agents.json"
@@ -128,16 +119,13 @@ class TuiLiveView:
         self._hydrate_sdk_session_history(run_dir, statuses.keys())
 
     def _load_run_record(self, run_dir: Path) -> None:
-        """Take the user's opening message and the run's MCP servers off the record."""
+        """Take the user's opening message off the record."""
         try:
             record = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
         except (OSError, json.JSONDecodeError):
             return
         if not isinstance(record, dict):
             return
-        connections = record.get("mcp_connections")
-        if isinstance(connections, list):
-            self.set_mcp_connections(name for name in connections if isinstance(name, str))
         instruction = record.get("user_instruction")
         if not isinstance(instruction, str):
             return
@@ -348,7 +336,7 @@ class TuiLiveView:
             "status": "running",
             "agent_id": agent_id,
             "call_id": call_id,
-            **self._mcp_tool_fields(call["tool_name"]),
+            **self._mcp_tool_fields(call["tool_name"], call["args"]),
         }
         if existing is None:
             event = self._append_event(agent_id, "tool", tool_data, timestamp=timestamp)
@@ -371,6 +359,10 @@ class TuiLiveView:
         event_key = (agent_id, call_id)
         event = self._tool_event_by_agent_and_call_id.get(event_key)
         if event is None:
+            # No prior call event to update, so its arguments are gone and the
+            # connection an MCP call went out to cannot be recovered. The matching
+            # call event, when there is one, already carries the MCP fields; this
+            # arrives only when the call was never projected, so it stays generic.
             event = self._append_event(
                 agent_id,
                 "tool",
@@ -380,7 +372,6 @@ class TuiLiveView:
                     "status": "completed",
                     "agent_id": agent_id,
                     "call_id": call_id,
-                    **self._mcp_tool_fields(output["tool_name"]),
                 },
                 timestamp=timestamp,
             )

strix/interface/tui/runtime.py

@@ -207,23 +207,9 @@ class GoTuiRuntime:
             self.controller.notify_changed()
 
     def capture_event(self, agent_id: str, event: Any) -> None:
-        self._refresh_mcp_connections()
         self.live_view.ingest_sdk_event(agent_id, event)
         self.controller.notify_changed()
 
-    def _refresh_mcp_connections(self) -> None:
-        """Hand the projection the MCP servers the scan connected.
-
-        The scan records them as it connects, which is before the agent can call
-        anything, and the projection needs them to say which server a tool call
-        went out to. Read on the way in rather than pushed, so no tool call can
-        be projected before they arrive.
-        """
-        if self.report_state is None:
-            return
-        connections = self.report_state.run_record.get("mcp_connections") or []
-        self.live_view.set_mcp_connections(connections)
-
     async def _sync_agent_state(self) -> bool:
         parent_of, statuses, names, errors = await self.coordinator.graph_snapshot()
         changed = False

strix/interface/viewer/frontend/src/components/live/tool-renderers/McpRenderer.tsx

@@ -47,13 +47,29 @@ export default function McpRenderer({
   const lines = argLines(args);
   const failed = status === "failed" || status === "error";
   const error = failed ? errorText(result) : null;
+  // describe_mcp inspects a connection's catalog rather than calling a tool on
+  // it, so the connection is the subject and there is no underlying tool.
+  const inspecting = toolName === "describe_mcp";
 
   return (
     <div>
       <div className="flex items-center gap-2 flex-wrap">
-        <span className="font-mono text-teal-300 font-semibold text-sm">{mcpTool || toolName}</span>
+        {inspecting ? (
-        <span className="text-[13px] text-[#555]">via MCP server</span>
+          <>
-        {mcpConnection && <span className="text-[13px] text-teal-400/80">{mcpConnection}</span>}
+            <span className="text-[13px] text-[#555]">Inspecting MCP server</span>
+            {mcpConnection && (
+              <span className="font-mono text-teal-300 font-semibold text-sm">{mcpConnection}</span>
+            )}
+          </>
+        ) : (
+          <>
+            <span className="font-mono text-teal-300 font-semibold text-sm">
+              {mcpTool || toolName}
+            </span>
+            <span className="text-[13px] text-[#555]">via MCP server</span>
+            {mcpConnection && <span className="text-[13px] text-teal-400/80">{mcpConnection}</span>}
+          </>
+        )}
       </div>
 
       {lines.length > 0 && (

strix/interface/viewer/frontend/src/components/live/tool-renderers/index.ts

@@ -181,8 +181,9 @@ function resolveCategory(toolName: string): ToolCategory | null {
 
 /**
  * A call to a tool from one of the user's MCP servers is placed by the
- * connection it was tagged with, ahead of every name-keyed lookup below: its
+ * connection it was tagged with, ahead of every name-keyed lookup below. Every
- * name belongs to that server and matches nothing in this table.
+ * MCP call goes through the `call_mcp` / `describe_mcp` dispatch tools, so the
+ * connection tag, not the tool name, is what routes it to the MCP renderer.
  */
 export function getToolRenderer(
   toolName: string,

strix/interface/viewer/frontend/src/types/events.ts

@@ -101,9 +101,10 @@ export interface ToolRendererProps {
   status: "running" | "completed" | "failed" | "error";
   /**
    * Set only on a call to a tool from an MCP server the user connected: the name
-   * they gave that connection, and the server's own name for the tool. The
+   * they gave that connection, and the server's own name for the tool. Every MCP
-   * engine resolves both, because `toolName` is the two glued together and
+   * call goes through the `call_mcp` / `describe_mcp` dispatch tools, so the
-   * cannot be split back apart here.
+   * engine reads both out of the call's arguments; `describe_mcp` inspects a
+   * connection and leaves `mcpTool` empty.
    */
   mcpConnection?: string | null;
   mcpTool?: string | null;

strix/interface/viewer/static/index.html

@@ -6,7 +6,7 @@
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <meta name="color-scheme" content="dark" />
     <title>Strix Results</title>
-    <script type="module" crossorigin src="./assets/index-C9c1WbvP.js"></script>
+    <script type="module" crossorigin src="./assets/index-CYf9nnT3.js"></script>
     <link rel="stylesheet" crossorigin href="./assets/index-D0453ODW.css">
   </head>
   <body>

strix/tools/mcp/__init__.py

@@ -1,25 +1,54 @@
-"""Generic MCP client: connect MCP servers and expose their tools."""
+"""Generic MCP client: connect MCP servers and reach their tools on demand."""
 
 from __future__ import annotations
 
-from strix.tools.mcp.client import ConnectedMcpServer, connect_mcp_servers
+from strix.tools.mcp.agent_tools import call_mcp, describe_mcp, list_mcps
+from strix.tools.mcp.client import (
+    ConnectedMcpServer,
+    attach_mcp_requests,
+    connect_mcp_servers,
+)
 from strix.tools.mcp.config import (
     BearerAuth,
     McpAuth,
     McpConnectionConfig,
 )
 from strix.tools.mcp.loader import load_user_mcp_configs
-from strix.tools.mcp.naming import McpToolOrigin, namespaced_tool_name, resolve_mcp_tool
+from strix.tools.mcp.naming import namespaced_tool_name
+from strix.tools.mcp.registry import (
+    CALL_MCP_TOOL,
+    DESCRIBE_MCP_TOOL,
+    MCP_DISPATCH_TOOLS,
+    MCP_REGISTRY_CONTEXT_KEY,
+    McpCallInfo,
+    McpConnectionEntry,
+    McpConnectionRequest,
+    McpConnectionSummary,
+    McpRegistry,
+    resolve_mcp_call,
+)
 
 
 __all__ = [
+    "CALL_MCP_TOOL",
+    "DESCRIBE_MCP_TOOL",
+    "MCP_DISPATCH_TOOLS",
+    "MCP_REGISTRY_CONTEXT_KEY",
     "BearerAuth",
     "ConnectedMcpServer",
     "McpAuth",
+    "McpCallInfo",
     "McpConnectionConfig",
-    "McpToolOrigin",
+    "McpConnectionEntry",
+    "McpConnectionRequest",
+    "McpConnectionSummary",
+    "McpRegistry",
+    "attach_mcp_requests",
+    "call_mcp",
     "connect_mcp_servers",
+    "describe_mcp",
+    "list_mcps",
     "load_user_mcp_configs",
     "namespaced_tool_name",
-    "resolve_mcp_tool",
+    "resolve_mcp_call",
 ]

strix/tools/mcp/agent_tools.py

@@ -0,0 +1,173 @@
+"""The three generic MCP dispatch tools every agent carries.
+
+Under the generic-dispatch model an agent does not get one tool per MCP tool.
+It gets exactly these three and discovers connections on demand:
+
+- ``list_mcps()`` returns the connections available this run — each connection's
+  id, name, description, and tool count, with no tool schemas — so the model can
+  discover what it can reach without any inventory in the system prompt.
+- ``describe_mcp(connection)`` returns, as text, one connection's tools with
+  their names, descriptions, and JSON input schemas — the schemas the model
+  needs, fetched on demand instead of loaded onto every request up front.
+- ``call_mcp(connection, tool, arguments)`` dispatches one call to a
+  connection's tool and returns its result.
+
+All three read the per-run :class:`~strix.tools.mcp.registry.McpRegistry` from the
+run context under :data:`~strix.tools.mcp.registry.MCP_REGISTRY_CONTEXT_KEY`. They
+are ordinary ``FunctionTool`` objects placed in the agent factory's base tool set,
+so the factory's output-bounding and disk-spill wrapping apply to their results
+automatically.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+
+from agents import RunContextWrapper, function_tool
+
+from strix.tools.mcp.client import dispatch_mcp_call
+from strix.tools.mcp.naming import namespaced_tool_name
+from strix.tools.mcp.registry import MCP_REGISTRY_CONTEXT_KEY, McpRegistry
+
+
+if TYPE_CHECKING:
+    from mcp.types import Tool as MCPTool
+
+
+def _registry_from_ctx(ctx: RunContextWrapper) -> McpRegistry | None:
+    context = ctx.context if isinstance(ctx.context, dict) else {}
+    registry = context.get(MCP_REGISTRY_CONTEXT_KEY)
+    return registry if isinstance(registry, McpRegistry) else None
+
+
+_NO_CONNECTIONS = "No MCP connections are configured for this run."
+
+
+def _unknown_connection(connection: str, registry: McpRegistry) -> str:
+    available = ", ".join(registry.names()) or "(none)"
+    return f"Unknown MCP connection {connection!r}. Available connections: {available}."
+
+
+def _format_tool(tool: MCPTool) -> str:
+    schema = json.dumps(tool.inputSchema or {"type": "object"}, indent=2, ensure_ascii=False)
+    description = (tool.description or "").strip() or "(no description)"
+    return f"- {tool.name}: {description}\n  input schema:\n{schema}"
+
+
+@function_tool(timeout=60)
+async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
+    """List the MCP connections available this run, so you can discover them.
+
+    Read-only. Returns one entry per connection with its ``id`` (the exact name
+    you pass to ``describe_mcp`` and ``call_mcp``), ``name``, ``description``, and
+    ``tool_count`` — no tool schemas. The three MCP tools work in order: call
+    ``list_mcps`` to discover the available connections, then ``describe_mcp`` on
+    one connection to inspect its tools and their input schemas, then ``call_mcp``
+    to run one of its tools. Returns an empty ``connections`` list when the run has
+    no MCP connections.
+    """
+    registry = _registry_from_ctx(ctx)
+    if registry is None or not registry:
+        return {"connections": []}
+    return {
+        "connections": [
+            {
+                "id": summary.name,
+                "name": summary.name,
+                "description": summary.purpose,
+                "tool_count": summary.tool_count,
+            }
+            for summary in registry.summaries()
+        ]
+    }
+
+
+@function_tool(timeout=60)
+async def describe_mcp(ctx: RunContextWrapper, connection: str) -> str:
+    """List the tools one MCP connection offers, with their input schemas.
+
+    Read-only. Look up a connection by the id ``list_mcps`` reported for it; this
+    returns each of its tools with the tool's name, description, and JSON input
+    schema — the argument shape you pass to ``call_mcp``. Call this before
+    ``call_mcp`` on any connection you have not used yet. Nothing is fetched from
+    or run against the connection's data.
+
+    Args:
+        connection: The connection name exactly as reported by ``list_mcps``.
+    """
+    registry = _registry_from_ctx(ctx)
+    if registry is None or not registry:
+        return _NO_CONNECTIONS
+    entry = registry.get(connection)
+    if entry is None:
+        return _unknown_connection(connection, registry)
+    tools = await entry.server.list_tools()
+    if not tools:
+        return f"MCP connection {connection!r} offers no tools."
+    header = f"MCP connection {connection!r} offers {len(tools)} tool(s):"
+    body = "\n".join(_format_tool(tool) for tool in tools)
+    return f"{header}\n{body}"
+
+
+@function_tool(timeout=120, strict_mode=False)
+async def call_mcp(
+    ctx: RunContextWrapper,
+    connection: str,
+    tool: str,
+    arguments: Any = None,
+) -> Any:
+    """Call one tool on one MCP connection and return its result.
+
+    Address the tool by the connection id from ``list_mcps`` and the tool name
+    from ``describe_mcp`` on that connection. Pass the tool's arguments as an
+    object matching the input schema ``describe_mcp`` showed for it (omit it, or
+    pass an empty object, for a tool that takes no arguments).
+
+    Args:
+        connection: The connection name exactly as reported by ``list_mcps``.
+        tool: The tool name, exactly as reported by ``describe_mcp``.
+        arguments: The tool's arguments as a JSON object of names to values (for
+            example ``{"path": "app.py"}``), or omitted/empty for a tool that
+            takes none. Pass an object, not a stringified one. Its shape is
+            whatever ``describe_mcp`` showed for the tool rather than a shape this
+            tool fixes in advance.
+    """
+    registry = _registry_from_ctx(ctx)
+    if registry is None or not registry:
+        return _NO_CONNECTIONS
+    entry = registry.get(connection)
+    if entry is None:
+        return _unknown_connection(connection, registry)
+    invalid_arguments = (
+        f"Invalid arguments for {connection!r}.{tool}: expected a JSON object of "
+        "argument names to values, or none. Call describe_mcp for the input schema."
+    )
+    if isinstance(arguments, str):
+        # The ``arguments`` parameter is schema-less (an open object is not
+        # expressible as a strict tool schema), so some models serialize it as a
+        # JSON string instead of a bare object. Accept a string that decodes to an
+        # object so a correct call is not rejected over its encoding.
+        stripped = arguments.strip()
+        try:
+            arguments = json.loads(stripped) if stripped else {}
+        except json.JSONDecodeError:
+            return invalid_arguments
+    if arguments is not None and not isinstance(arguments, dict):
+        return invalid_arguments
+    available = await entry.server.list_tools()
+    valid_names = {mcp_tool.name for mcp_tool in available}
+    if tool not in valid_names:
+        offered = ", ".join(sorted(valid_names)) or "(none)"
+        return (
+            f"Unknown tool {tool!r} on MCP connection {connection!r}. "
+            f"Tools this connection offers: {offered}. "
+            "Call describe_mcp for their input schemas."
+        )
+    return await dispatch_mcp_call(
+        entry.server,
+        tool,
+        arguments or {},
+        label=namespaced_tool_name(connection, tool),
+        result_transform=entry.result_transform,
+    )

strix/tools/mcp/client.py

@@ -1,14 +1,15 @@
-"""Connect to MCP servers and expose their tools to the agent.
+"""Connect to MCP servers so a run can reach their tools on demand.
 
 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
+connects each server, counts the tools it offers (honoring the connection's
-of them when none is set), prefixes each with the connection name so servers do
+allowlist), and returns the live sessions. It does NOT register anything as an
-not collide, and registers them through the agent factory. The factory applies
+agent tool: under the generic-dispatch model the run holds these sessions in a
-output bounding, per-call timeouts, and structured errors to every registered
+per-run :class:`~strix.tools.mcp.registry.McpRegistry`, and the agent reaches
-tool, so this layer does not reimplement them.
+them through the two dispatch tools (``describe_mcp`` / ``call_mcp``), which call
+:func:`dispatch_mcp_call` here to run one tool and serialize its result.
 
-A server that cannot connect, or a tool set that cannot be registered, is logged
+A server that cannot connect is logged and skipped, so one bad connection never
-and skipped, so one bad connection never fails the run.
+fails the run.
 """
 
 from __future__ import annotations
@@ -16,36 +17,34 @@ from __future__ import annotations
 import contextlib
 import json
 import logging
+import os
+from pathlib import Path
 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 mcp.client.stdio import stdio_client
-from strix.agents.factory import register_agent_tools
-from strix.tools.mcp.naming import namespaced_tool_name
 
 
 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
+    from strix.tools.mcp.registry import McpConnectionRequest, McpRegistry
 
-    # Runs on each tool's structured result before it reaches the agent. Called
+    # Runs on one tool call's structured result before it reaches the agent.
-    # ``result_transform(namespaced_tool_name, structured_result)`` and its return
+    # Called ``result_transform(label, structured_result)`` and its return value
-    # value becomes the tool's output. ``structured_result`` is the parsed
+    # becomes the tool's output. ``label`` is the model-facing
-    # ``CallToolResult`` as a dict (not a serialized string), so the transform can
+    # ``<connection>_<tool>`` name so a transform keyed on names still resolves
-    # project or drop individual fields.
+    # the same way it did under per-tool registration; ``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]
 
 
@@ -53,12 +52,14 @@ logger = logging.getLogger(__name__)
 
 
 class ConnectedMcpServer(NamedTuple):
-    """One successfully connected MCP server and how many tools it registered.
+    """One successfully connected MCP server and how many tools it offers.
 
-    ``server`` is kept so the caller can clean it up when the run ends;
+    ``server`` is kept so the caller can clean it up when the run ends, and so
-    ``name`` and ``tool_count`` let the caller show the user a startup summary;
+    the caller can hand the live session to the run's
+    :class:`~strix.tools.mcp.registry.McpRegistry`; ``name`` and ``tool_count``
+    let the caller show the user a startup summary and fill the prompt inventory;
     ``notes`` carries the connection's optional free-text description so the
-    caller can surface it to the agent as context about the connection.
+    caller can surface it as the connection's purpose in the inventory.
     """
 
     server: MCPServer
@@ -75,13 +76,42 @@ def _auth_headers(config: McpConnectionConfig) -> dict[str, str]:
     return {"Authorization": f"Bearer {auth.token}"}
 
 
+@contextlib.asynccontextmanager
+async def _quiet_stdio_streams(params: Any) -> Any:
+    """Run a stdio MCP server with its stderr sent to the void.
+
+    A stdio MCP server chats on stderr as it boots (the filesystem server, for
+    one, prints ``Allowed directories: [ ... ]``). The mcp library forwards that
+    stderr to the parent's ``sys.stderr`` by default, which is the terminal the
+    TUI is drawing on, so the banner corrupts the display. Pointing ``errlog`` at
+    ``os.devnull`` drops that chatter. Connection failures are unaffected: they
+    still raise from ``connect`` and are logged by :func:`connect_mcp_servers`.
+    """
+    with Path(os.devnull).open("w", encoding="utf-8") as errlog:
+        async with stdio_client(params, errlog=errlog) as streams:
+            yield streams
+
+
+class _QuietMCPServerStdio(MCPServerStdio):
+    """``MCPServerStdio`` whose subprocess stderr is kept off the terminal.
+
+    The SDK's ``create_streams`` calls ``stdio_client(self.params)`` with no
+    ``errlog``, so the subprocess stderr defaults to ``sys.stderr`` and paints
+    server banners over the running TUI. Overriding it lets us redirect that
+    stream; everything else about the stdio transport is unchanged.
+    """
+
+    def create_streams(self) -> Any:
+        return _quiet_stdio_streams(self.params)
+
+
 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
+    even list tools outside it, so it is the authoritative gate on what
-    same allowlist as the authoritative gate on what gets registered. When it is
+    ``describe_mcp`` and ``call_mcp`` can see. When it is ``None`` no filter is
-    ``None`` no filter is applied and every listed tool is registered.
+    applied and every listed tool is reachable.
     """
     tool_filter = (
         create_static_tool_filter(allowed_tool_names=config.allowed_tools)
@@ -95,7 +125,7 @@ def _build_server(config: McpConnectionConfig) -> MCPServer:
             "args": config.args,
             "env": config.env,
         }
-        return MCPServerStdio(
+        return _QuietMCPServerStdio(
             params=stdio_params,
             name=config.name,
             tool_filter=tool_filter,
@@ -114,105 +144,14 @@ def _build_server(config: McpConnectionConfig) -> MCPServer:
     )
 
 
-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 = namespaced_tool_name(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
+    content blocks, unwrapping a single block). Because the dispatch tool routes
     its own call, this is what makes the agent see byte-identical content to what
-    the SDK would have produced on its own.
+    the SDK would have produced building the tool itself.
     """
     if getattr(server, "use_structured_content", False) and result.structuredContent:
         return json.dumps(result.structuredContent)
@@ -232,85 +171,88 @@ def _mcp_result_to_tool_output(server: MCPServer, result: Any) -> Any:
     return outputs
 
 
-def _install_error_status_capture(
+async def dispatch_mcp_call(
-    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,
+    tool_name: str,
+    arguments: dict[str, Any],
+    *,
+    label: str,
     result_transform: ResultTransform | None = None,
-) -> list[Tool]:
+) -> Any:
-    """List a connected server's tools, prefix + filter them, and register them.
+    """Run one MCP tool call and convert its result to a tool output.
 
-    ``allowed_tools`` of ``None`` registers every listed tool; a list restricts
+    Shared single dispatch point for the generic ``call_mcp`` tool. Calls
-    to exactly those names.
+    ``server.call_tool`` with the tool's unprefixed name, then:
+
+    - with a ``result_transform`` (strix-pro's sanitizer), hands the parsed
+      :class:`CallToolResult` to it as ``result_transform(label, structured)`` and
+      returns whatever the transform returns; or
+    - without one, serializes the result the way the agents SDK does (see
+      :func:`_mcp_result_to_tool_output`) and, when the result is an MCP error,
+      normalizes it through :func:`_errored_tool_output` so the failure reaches
+      the interfaces (see that function for the representation and why it does not
+      corrupt the content the agent receives).
+    """
+    result = await server.call_tool(tool_name, arguments)
+    if result_transform is not None:
+        return result_transform(label, result.model_dump(mode="json"))
+    tool_output = _mcp_result_to_tool_output(server, result)
+    if getattr(result, "isError", False):
+        return _errored_tool_output(tool_output)
+    return tool_output
+
+
+def _errored_tool_output(tool_output: Any) -> dict[str, Any]:
+    """Tag a serialized MCP error so the interfaces render it as failed.
+
+    Both the TUI and the run viewer decide a tool call failed by reading a
+    ``success`` key off a top-level dict in the result (``success is False`` means
+    failed). :func:`_mcp_result_to_tool_output` returns a dict only for a single
+    content block; a structured-content result comes back as a string and a
+    multi-block result as a list, and on those the failure flag had nowhere to
+    ride, so the interfaces showed a failed call as done. This normalizes every
+    errored result to a top-level dict carrying ``success: False``:
+
+    - a single content block (already a dict) keeps its ``type``/``text`` and gains
+      ``success: False`` alongside. The SDK's ToolOutput projection keeps the known
+      ``type``/``text`` fields and drops ``success`` before the agent sees it, so
+      the agent still receives exactly the error content;
+    - a list (multiple blocks) or a string (structured content) is placed under a
+      stable ``content`` key so the flag has a top-level dict to ride on. The agent
+      still receives the full error content, under ``content``, rather than losing
+      it.
+    """
+    if isinstance(tool_output, dict):
+        return {**tool_output, "success": False}
+    return {"success": False, "content": tool_output}
+
+
+async def _count_server_tools(config: McpConnectionConfig, server: MCPServer) -> int:
+    """Count a connected server's reachable tools for the startup summary.
+
+    ``allowed_tools`` of ``None`` counts every listed tool; a list counts only
+    those names. The count matches what ``describe_mcp`` will show, because the
+    static tool filter built in :func:`_build_server` restricts the server's own
+    ``list_tools`` to the same allowlist.
     """
     allowed = config.allowed_tools
     mcp_tools = await server.list_tools()
-
+    return sum(1 for mcp_tool in mcp_tools if allowed is None or mcp_tool.name in allowed)
-    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.
+    """Connect to each MCP server and return its live session.
-
-    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
+    the SDK server (so the caller can clean it up when the run ends and hand it to
-    server name and how many tools it registered (so the caller can show the
+    the run's registry) plus the server name, how many tools it offers, and the
-    user a startup summary). Connections that fail are skipped rather than
+    connection's notes. Connections that fail are skipped rather than raised.
-    raised.
+
+    Nothing is registered as an agent tool: the caller builds a per-run
+    :class:`~strix.tools.mcp.registry.McpRegistry` from these sessions, and the
+    agent reaches each tool on demand through ``describe_mcp`` / ``call_mcp``.
     """
     connected: list[ConnectedMcpServer] = []
     for config in configs:
@@ -318,7 +260,7 @@ async def connect_mcp_servers(
         try:
             server = _build_server(config)
             await server.connect()  # type: ignore[no-untyped-call]
-            tools = await _register_server_tools(config, server, result_transform)
+            tool_count = await _count_server_tools(config, server)
         except Exception:
             logger.exception("Skipping MCP connection %r", config.name)
             if server is not None:
@@ -339,11 +281,47 @@ async def connect_mcp_servers(
                     await established.server.cleanup()  # type: ignore[no-untyped-call]
             raise
 
-        logger.info("Connected MCP server %r (%d tools)", config.name, len(tools))
+        logger.info("Connected MCP server %r (%d tools)", config.name, tool_count)
         connected.append(
             ConnectedMcpServer(
-                server=server, name=config.name, tool_count=len(tools), notes=config.notes
+                server=server, name=config.name, tool_count=tool_count, notes=config.notes
             )
         )
 
     return connected
+
+
+async def attach_mcp_requests(
+    requests: list[McpConnectionRequest],
+    registry: McpRegistry,
+) -> list[ConnectedMcpServer]:
+    """Connect a caller's MCP requests and populate the run's registry.
+
+    The one shared attach-and-populate path both the command-line and the
+    SaaS/pro product go through, so all connecting and cleanup lives in one owner.
+    The caller supplies inert :class:`McpConnectionRequest` objects (a config plus
+    a provider label, an optional per-connection ``result_transform``, and an
+    optional ``purpose``) and never a live session: the engine connects each
+    config here, reusing :func:`connect_mcp_servers` so the fail-open behavior (a
+    connection that will not connect is logged and skipped) and the cancellation
+    cleanup are preserved unchanged.
+
+    For each connection that came up, this registers it under its config name with
+    its tool count, its ``provider`` label, its ``result_transform``, and a purpose
+    of ``request.purpose`` when set else the connection's notes. Returns the
+    connected servers (the runner records them and cleans them up when the run
+    ends).
+    """
+    request_by_name = {request.config.name: request for request in requests}
+    connections = await connect_mcp_servers([request.config for request in requests])
+    for connection in connections:
+        request = request_by_name[connection.name]
+        registry.add(
+            name=connection.name,
+            server=connection.server,
+            tool_count=connection.tool_count,
+            purpose=request.purpose or connection.notes,
+            provider=request.provider,
+            result_transform=request.result_transform,
+        )
+    return connections

strix/tools/mcp/config.py

@@ -61,9 +61,9 @@ class McpConnectionConfig(BaseModel):
 
     notes: str | None = None
     """Free-text notes for the agent describing what this connection is and how
-    to use it. When set, the runner collects the notes of every connection into
+    to use it. When set, the note becomes the connection's purpose line in the
-    a single block on the root task, so a note describes its connection once
+    MCP inventory every agent renders in its prompt, so it describes the
-    rather than being repeated onto each of its tools."""
+    connection once rather than being repeated onto each of its tools."""
 
     @model_validator(mode="after")
     def _check_transport_fields(self) -> McpConnectionConfig:

strix/tools/mcp/loader.py

@@ -1,9 +1,10 @@
 """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,
+small JSON file. Strix reads it at the start of a run and connects to each
-and registers its tools. The file is optional; without it the run simply gets
+server, holding the live sessions in the run's registry for the agent to reach
-no MCP tools.
+on demand. The file is optional; without it the run simply gets no MCP
+connections.
 
 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
@@ -46,9 +47,9 @@ def _resolve_path(path: Path | None) -> 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
+    A connection's name is its key in the run's registry, so two connections
-    sharing a name would collide and the second's tools would be silently
+    sharing a name would collide and the second would overwrite the first. Drop
-    rejected at registration. Drop the duplicate here, with a warning, instead.
+    the duplicate here, with a warning, instead.
     """
     seen: set[str] = set()
     unique: list[McpConnectionConfig] = []

strix/tools/mcp/naming.py

@@ -1,18 +1,13 @@
-"""How an MCP server's tools are named for the model, and how to read that back.
+"""How an MCP server's tools are named for the model.
 
-Kept apart from the client, and stdlib-only, so the interfaces can resolve which
+Kept apart from the client, and stdlib-only, so a caller can build the
-connection a tool call went to without importing the MCP client (and through it
+model-facing name for a connection's tool without importing the MCP client (and
-the agents SDK and every registered tool).
+through it the agents SDK and every registered tool).
 """
 
 from __future__ import annotations
 
 import re
-from typing import TYPE_CHECKING, NamedTuple
-
-
-if TYPE_CHECKING:
-    from collections.abc import Iterable
 
 
 # A tool name offered to a model has to be letters, digits, underscores or
@@ -32,49 +27,3 @@ def namespaced_tool_name(connection: str, tool: str) -> str:
     which tool is invoked.
     """
     return _INVALID_TOOL_NAME_CHARS.sub("_", f"{connection}_{tool}")
-
-
-class McpToolOrigin(NamedTuple):
-    """Where a model-facing tool name came from, for showing the user.
-
-    ``connection`` is the name the user gave the connection in their config, so
-    it reads the way they wrote it. ``tool`` is what is left of the model-facing
-    name once the connection prefix is removed, which is the server's own name
-    for the tool and the part a reader cares about.
-    """
-
-    connection: str
-    tool: str
-
-
-def resolve_mcp_tool(tool_name: str, connections: Iterable[str]) -> McpToolOrigin | None:
-    """Split a model-facing tool name against the run's connections, or ``None``.
-
-    Matched against the connections the run actually made rather than by
-    splitting the name on the separator: the connection name and the server's own
-    tool name can both contain underscores, so a split is ambiguous and would
-    attribute calls to a connection that does not exist. Each connection name is
-    sanitized the same way :func:`namespaced_tool_name` sanitizes it before
-    comparing, so a connection whose name has characters a model-facing name
-    cannot carry still matches.
-
-    The longest match wins, so one connection whose name is a prefix of another's
-    still resolves to the right one. The character after the prefix has to be a
-    separator rather than more of a name, which any non-alphanumeric satisfies,
-    so this holds whichever separator :func:`namespaced_tool_name` uses.
-    """
-    best: McpToolOrigin | None = None
-    best_length = 0
-    for connection in connections:
-        prefix = _INVALID_TOOL_NAME_CHARS.sub("_", connection)
-        if not prefix or len(tool_name) <= len(prefix) or not tool_name.startswith(prefix):
-            continue
-        if tool_name[len(prefix)].isalnum():
-            continue
-        if len(prefix) > best_length:
-            # Past the prefix and its single separator character is the tool's
-            # own name; if a server named a tool nothing but separators, fall
-            # back to the whole name so the row still says something.
-            tool = tool_name[len(prefix) + 1 :] or tool_name
-            best, best_length = McpToolOrigin(connection, tool), len(prefix)
-    return best

strix/tools/mcp/registry.py

@@ -0,0 +1,216 @@
+"""Per-run registry of the MCP connections a scan may reach.
+
+Replaces per-tool registration. The old model turned every tool of every
+connected MCP server into its own agent tool, so a run with a handful of
+connections put dozens of provider tool schemas on the root agent's first LLM
+request. Instead, a run holds its live connections here, keyed by the name the
+user gave each connection, and every agent reaches them through three generic
+dispatch tools: ``list_mcps`` to discover the available connections, ``describe_mcp``
+to learn one connection's tool schemas on demand, and ``call_mcp`` to run one of
+its tools.
+
+One :class:`McpRegistry` is built per run in :mod:`strix.core.runner`, stored in
+the run context under :data:`MCP_REGISTRY_CONTEXT_KEY`, and shared by the root
+agent and every child (the child context is a copy of the parent's, so it
+carries the same registry object).
+
+strix-pro imports :class:`McpRegistry` to add its cloud connections into the
+same registry and to attach a per-connection ``result_transform`` (its
+sanitizer), which :func:`strix.tools.mcp.client.dispatch_mcp_call` applies at the
+single dispatch point.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from typing import TYPE_CHECKING, Any, NamedTuple
+
+
+if TYPE_CHECKING:
+    from agents.mcp import MCPServer
+
+    from strix.tools.mcp.client import ResultTransform
+    from strix.tools.mcp.config import McpConnectionConfig
+
+
+# The run-context key under which the runner stores the per-run registry, and
+# the two dispatch tools read it back. Kept here so the tools, the runner, and
+# strix-pro all agree on one name.
+MCP_REGISTRY_CONTEXT_KEY = "mcp_registry"
+
+
+# The two connection-scoped dispatch tools an interface attributes to a specific
+# MCP connection. ``call_mcp`` runs one tool on a connection; ``describe_mcp``
+# lists a connection's tool schemas. (``list_mcps`` is deliberately not here: it
+# names no single connection, so it renders as an ordinary tool call.) Kept here
+# (not in the interface layer) so the engine, the OSS viewer, and strix-pro's
+# tracer all recognise a connection-scoped dispatch call by the same names.
+CALL_MCP_TOOL = "call_mcp"
+DESCRIBE_MCP_TOOL = "describe_mcp"
+MCP_DISPATCH_TOOLS = frozenset({CALL_MCP_TOOL, DESCRIBE_MCP_TOOL})
+
+
+@dataclasses.dataclass(frozen=True)
+class McpConnectionEntry:
+    """One live MCP connection a scan may reach, keyed by ``name``.
+
+    ``server`` is the connected SDK session the dispatch tools list tools on and
+    call tools through. ``purpose`` is the human label ``list_mcps`` reports as the
+    connection's description (the user's connection notes, or whatever the caller
+    supplies). ``tool_count`` is how many tools the connection offers, also
+    reported by ``list_mcps``. ``result_transform``, when set, runs on each call's structured result
+    at the single dispatch point (strix-pro's sanitizer uses it). ``provider`` is
+    an optional source label (e.g. ``"supabase"``) the caller tags the connection
+    with; the command-line path leaves it ``None``, and event tagging surfaces it
+    when set.
+    """
+
+    server: MCPServer
+    name: str
+    purpose: str | None = None
+    tool_count: int = 0
+    result_transform: ResultTransform | None = None
+    provider: str | None = None
+
+
+@dataclasses.dataclass(frozen=True)
+class McpConnectionSummary:
+    """One connection summary ``list_mcps`` returns: what an agent needs to decide
+    whether to ``describe_mcp`` a connection, with no tool schemas."""
+
+    name: str
+    purpose: str | None
+    tool_count: int
+    provider: str | None = None
+
+
+@dataclasses.dataclass(frozen=True)
+class McpConnectionRequest:
+    """A source-agnostic request to attach one MCP connection to a run.
+
+    The caller hands the engine an inert ``config`` (how to reach the server, its
+    name, and any auth token) plus metadata, and never a live session: the engine
+    owns connecting and cleaning up. ``provider`` is an optional source label
+    (e.g. ``"supabase"``; empty for the command-line path). ``result_transform``
+    is an optional per-connection transform run on each call's structured result
+    at the single dispatch point (strix-pro's sanitizer; empty for the
+    command-line path). ``purpose`` is the human label ``list_mcps`` reports as the
+    connection's description; when unset it falls back to ``config.notes``.
+    """
+
+    config: McpConnectionConfig
+    provider: str | None = None
+    result_transform: ResultTransform | None = None
+    purpose: str | None = None
+
+
+class McpCallInfo(NamedTuple):
+    """What one MCP dispatch call resolved to: the connection name, the
+    underlying tool (empty for ``describe_mcp``), and the connection's provider
+    label (``None`` when unknown or untagged)."""
+
+    connection: str
+    tool: str
+    provider: str | None
+
+
+class McpRegistry:
+    """Connection name -> live MCP connection, built per run and shared by every
+    agent in the run.
+
+    Public API (strix-pro builds against it): the constructor, :meth:`add`,
+    :meth:`get`, and :meth:`summaries`.
+    """
+
+    def __init__(self) -> None:
+        self._entries: dict[str, McpConnectionEntry] = {}
+
+    def add(
+        self,
+        *,
+        name: str,
+        server: MCPServer,
+        purpose: str | None = None,
+        tool_count: int = 0,
+        result_transform: ResultTransform | None = None,
+        provider: str | None = None,
+    ) -> McpConnectionEntry:
+        """Register one connection under ``name`` (last write wins)."""
+        entry = McpConnectionEntry(
+            server=server,
+            name=name,
+            purpose=purpose,
+            tool_count=tool_count,
+            result_transform=result_transform,
+            provider=provider,
+        )
+        self._entries[name] = entry
+        return entry
+
+    def get(self, name: str) -> McpConnectionEntry | None:
+        """The connection registered under ``name``, or ``None``."""
+        return self._entries.get(name)
+
+    def names(self) -> list[str]:
+        """The registered connection names, in insertion order."""
+        return list(self._entries)
+
+    def summaries(self) -> list[McpConnectionSummary]:
+        """One inventory summary per connection, in insertion order."""
+        return [
+            McpConnectionSummary(
+                name=entry.name,
+                purpose=entry.purpose,
+                tool_count=entry.tool_count,
+                provider=entry.provider,
+            )
+            for entry in self._entries.values()
+        ]
+
+    def clear(self) -> None:
+        """Drop every connection (the sessions themselves are closed by the
+        runner)."""
+        self._entries.clear()
+
+    def __len__(self) -> int:
+        return len(self._entries)
+
+    def __bool__(self) -> bool:
+        return bool(self._entries)
+
+
+def resolve_mcp_call(
+    tool_name: str,
+    args: dict[str, Any],
+    registry: McpRegistry | None = None,
+) -> McpCallInfo | None:
+    """Resolve one tool call to the MCP connection/tool/provider it went out to.
+
+    The single resolver both the OSS viewer and strix-pro's tracer read a
+    dispatch call through, so a call is attributed the same way everywhere. Every
+    MCP call an agent makes goes through ``call_mcp`` or ``describe_mcp``, and the
+    connection (and, for ``call_mcp``, the server's own tool name) ride in the
+    call's ``args`` rather than the tool name, so they are read from there.
+
+    Returns ``None`` when ``tool_name`` is not one of the two dispatch tools, when
+    the call carries no connection name, or when a ``registry`` is supplied and
+    has no connection under that name. ``tool`` is the underlying tool for
+    ``call_mcp`` and empty for ``describe_mcp`` (which inspects the connection
+    itself). ``provider`` comes from the registry entry; it is ``None`` when no
+    ``registry`` is supplied (the viewer projects calls without one) or when the
+    connection carries no provider label.
+    """
+    if tool_name not in MCP_DISPATCH_TOOLS:
+        return None
+    connection = args.get("connection")
+    if not isinstance(connection, str) or not connection:
+        return None
+    provider: str | None = None
+    if registry is not None:
+        entry = registry.get(connection)
+        if entry is None:
+            return None
+        provider = entry.provider
+    raw_tool = args.get("tool") if tool_name == CALL_MCP_TOOL else ""
+    tool = raw_tool if isinstance(raw_tool, str) else ""
+    return McpCallInfo(connection=connection, tool=tool, provider=provider)

tests/conftest.py

@@ -0,0 +1,26 @@
+"""Shared test fixtures."""
+
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _isolate_mcp_config(
+    monkeypatch: pytest.MonkeyPatch, tmp_path_factory: pytest.TempPathFactory
+) -> None:
+    """Keep the whole suite from reading the developer's real MCP config.
+
+    ``run_strix_scan`` connects the MCP servers listed in
+    ``~/.strix/mcp-servers.json`` and threads an inventory of them into the
+    prompt context. Without isolation, any test that drives the runner on a
+    machine that has a real config would do real network I/O and see MCP
+    connections it never asked for. Point the loader at a path that does not
+    exist so it resolves to "no connections", and clear the per-run selection
+    env vars. Tests that exercise the loader itself set their own
+    ``STRIX_MCP_CONFIG`` after this runs and so override it.
+    """
+    missing = tmp_path_factory.mktemp("mcp-isolation") / "no-servers.json"
+    monkeypatch.setenv("STRIX_MCP_CONFIG", str(missing))
+    monkeypatch.delenv("STRIX_MCP_ONLY", raising=False)
+    monkeypatch.delenv("STRIX_MCP_EXCLUDE", raising=False)

tests/test_runner_mcp.py

@@ -0,0 +1,142 @@
+"""The runner attaches MCP connections source-agnostically.
+
+When a caller supplies ``mcp_connection_requests`` the runner attaches those;
+when it does not, the runner reads ``~/.strix/mcp-servers.json`` itself and wraps
+each config in a bare request. Either way the one shared ``attach_mcp_requests``
+routine does the connecting.
+"""
+
+from __future__ import annotations
+
+import types
+from typing import Any
+
+import pytest
+from agents import ModelSettings
+
+import strix.tools.mcp as mcp_pkg
+import strix.tools.notes.tools as notes_tools
+import strix.tools.todo.tools as todo_tools
+from strix.core import runner
+from strix.core.agents import AgentCoordinator
+from strix.runtime import session_manager
+from strix.tools.mcp import McpConnectionConfig, McpConnectionRequest
+
+
+def _settings() -> Any:
+    return types.SimpleNamespace(
+        llm=types.SimpleNamespace(
+            model="openai/gpt-4o",
+            reasoning_effort="high",
+            force_required_tool_choice=False,
+            timeout=300,
+            prompt_cache=True,
+            extra_headers=None,
+        ),
+        runtime=types.SimpleNamespace(max_context_images=3),
+    )
+
+
+def _wire_runner(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
+    monkeypatch.setattr(runner, "run_dir_for", lambda _scan_id: tmp_path)
+    monkeypatch.setattr(runner, "runtime_state_dir", lambda _run_dir: tmp_path)
+    monkeypatch.setattr(runner, "setup_scan_logging", lambda _run_dir: lambda: None)
+    monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
+    monkeypatch.setattr(runner, "load_settings", _settings)
+    monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _s: None)
+    monkeypatch.setattr(runner, "uses_chat_completions_tool_schema", lambda _m, _s: False)
+    monkeypatch.setattr(todo_tools, "hydrate_todos_from_disk", lambda _d: None)
+    monkeypatch.setattr(notes_tools, "hydrate_notes_from_disk", lambda _d: None)
+
+    async def _create_or_reuse(*_a: Any, **_k: Any) -> dict[str, Any]:
+        return {"client": object(), "session": object(), "caido_client": None}
+
+    async def _cleanup(*_a: Any, **_k: Any) -> None:
+        return None
+
+    monkeypatch.setattr(session_manager, "create_or_reuse", _create_or_reuse)
+    monkeypatch.setattr(session_manager, "cleanup", _cleanup)
+    monkeypatch.setattr(runner, "build_root_task", lambda _c: "task")
+    monkeypatch.setattr(runner, "build_scope_context", lambda _c: {})
+    monkeypatch.setattr(runner, "make_model_settings", lambda *_a, **_k: ModelSettings())
+    monkeypatch.setattr(runner, "build_strix_agent", lambda **_k: object())
+    monkeypatch.setattr(runner, "make_child_factory", lambda **_k: lambda **_kk: object())
+    monkeypatch.setattr(runner, "open_agent_session", lambda _root_id, _db: object())
+
+    async def _run_agent_loop(**_kwargs: Any) -> None:
+        return None
+
+    monkeypatch.setattr(runner, "run_agent_loop", _run_agent_loop)
+
+
+@pytest.mark.asyncio
+async def test_none_default_attaches_from_the_user_config_file(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    _wire_runner(monkeypatch, tmp_path)
+
+    file_config = McpConnectionConfig(
+        name="local_fs", transport="stdio", command="npx", notes="local files"
+    )
+    monkeypatch.setattr(mcp_pkg, "load_user_mcp_configs", lambda: [file_config])
+
+    captured: list[list[McpConnectionRequest]] = []
+
+    async def _capture(requests: list[McpConnectionRequest], _registry: Any) -> list[Any]:
+        captured.append(requests)
+        return []
+
+    monkeypatch.setattr(mcp_pkg, "attach_mcp_requests", _capture)
+
+    await runner.run_strix_scan(
+        scan_config={"targets": [], "scan_mode": "deep"},
+        scan_id="scan-none",
+        image="img",
+        coordinator=AgentCoordinator(),
+    )
+
+    # Each config from the file is wrapped in a bare request: no provider, no
+    # transform, no explicit purpose (purpose falls back to notes at attach time).
+    (requests,) = captured
+    assert len(requests) == 1
+    assert requests[0].config is file_config
+    assert requests[0].provider is None
+    assert requests[0].result_transform is None
+    assert requests[0].purpose is None
+
+
+@pytest.mark.asyncio
+async def test_supplied_requests_are_attached_and_the_user_file_is_not_read(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    _wire_runner(monkeypatch, tmp_path)
+
+    def _fail_if_read() -> list[Any]:
+        raise AssertionError("load_user_mcp_configs must not be read when requests are supplied")
+
+    monkeypatch.setattr(mcp_pkg, "load_user_mcp_configs", _fail_if_read)
+
+    captured: list[list[McpConnectionRequest]] = []
+
+    async def _capture(requests: list[McpConnectionRequest], _registry: Any) -> list[Any]:
+        captured.append(requests)
+        return []
+
+    monkeypatch.setattr(mcp_pkg, "attach_mcp_requests", _capture)
+
+    supplied = [
+        McpConnectionRequest(
+            config=McpConnectionConfig(name="db", url="https://mcp.example.com"),
+            provider="supabase",
+        )
+    ]
+
+    await runner.run_strix_scan(
+        scan_config={"targets": [], "scan_mode": "deep"},
+        scan_id="scan-supplied",
+        image="img",
+        coordinator=AgentCoordinator(),
+        mcp_connection_requests=supplied,
+    )
+
+    assert captured == [supplied]

tests/test_runner_root_prompt.py

@@ -14,11 +14,13 @@ import pytest
 from agents import ModelSettings
 from openai import RateLimitError
 
+import strix.tools.mcp as mcp_pkg
 import strix.tools.notes.tools as notes_tools
 import strix.tools.todo.tools as todo_tools
 from strix.core import runner
 from strix.core.agents import AgentCoordinator
 from strix.runtime import session_manager
+from strix.tools.mcp import BearerAuth, McpConnectionConfig, McpConnectionRequest
 
 
 def _make_rate_limit_error() -> RateLimitError:
@@ -180,6 +182,72 @@ async def test_root_prompt_options_default_to_none(
     assert kwargs["system_prompt_context"] == {"scope": "built-in"}
 
 
+@pytest.mark.asyncio
+async def test_mcp_available_flag_set_when_a_connection_attaches(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Any,
+) -> None:
+    """When at least one MCP connection attaches, the runner sets ``mcp_available``
+    plus a named ``mcp_connections`` inventory into the scan context that reaches
+    every agent, so each agent sees which connections exist at the start while
+    still being able to re-list them at run time via list_mcps."""
+    scope_context: dict[str, Any] = {"scope": "built-in"}
+    captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
+
+    async def _attach(_requests: Any, registry: Any) -> list[Any]:
+        registry.add(name="fs", server=object(), purpose="local files", tool_count=2)
+        return [types.SimpleNamespace(name="fs", tool_count=2, server=object())]
+
+    monkeypatch.setattr(mcp_pkg, "attach_mcp_requests", _attach)
+
+    request = McpConnectionRequest(
+        config=McpConnectionConfig(
+            name="fs",
+            url="https://mcp.example.com",
+            auth=BearerAuth(token="run-token"),
+        )
+    )
+
+    await runner.run_strix_scan(
+        scan_config={"targets": [], "scan_mode": "deep"},
+        scan_id="scan-mcp-available",
+        image="img",
+        coordinator=AgentCoordinator(),
+        mcp_connection_requests=[request],
+    )
+
+    kwargs = captured["kwargs"]
+    assert kwargs["system_prompt_context"]["mcp_available"] is True
+    # The named inventory names each connected server for the prompt.
+    assert kwargs["system_prompt_context"]["mcp_connections"] == [
+        {"name": "fs", "purpose": "local files", "tool_count": 2}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_mcp_available_flag_absent_without_a_connection(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Any,
+) -> None:
+    """With no MCP connection, the scan context carries no MCP key at all, so the
+    prompt's MCP section stays off."""
+    scope_context: dict[str, Any] = {"scope": "built-in"}
+    captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
+
+    monkeypatch.setattr(mcp_pkg, "load_user_mcp_configs", list)
+
+    await runner.run_strix_scan(
+        scan_config={"targets": [], "scan_mode": "deep"},
+        scan_id="scan-mcp-absent",
+        image="img",
+        coordinator=AgentCoordinator(),
+    )
+
+    kwargs = captured["kwargs"]
+    assert "mcp_available" not in kwargs["system_prompt_context"]
+    assert "mcp_connections" not in kwargs["system_prompt_context"]
+
+
 @pytest.mark.asyncio
 async def test_unknown_tool_calls_are_returned_to_the_model(
     monkeypatch: pytest.MonkeyPatch,