Merge bd58656bdf into cb581bf93d

README.md

@@ -352,6 +352,37 @@ Strix can connect to Model Context Protocol (MCP) servers you list and expose th
 
 Each server's tools are namespaced by `name`, for example `github_list_issues`. See the [MCP documentation](https://docs.strix.ai/integrations/mcp) for the full schema, tool filtering, and `stdio` servers.
 
+#### Run Strix as an MCP server: `strix mcp-server`
+
+The reverse of the above: expose Strix's own tools to an MCP client such as Claude Code, Cursor, or Zed. The client supplies the model and the orchestration loop, so this mode needs **no `STRIX_LLM` and no `LLM_API_KEY`** — if you already use Claude Code or another coding agent, Strix rides on that subscription. Strix contributes its sandbox container, the HTTP proxy, the skill packs, and the CVSS-scored reporting tools.
+
+**One-line install for Claude Code** (needs Docker and the `claude` CLI already installed):
+
+```bash
+curl -LsSf https://raw.githubusercontent.com/IshanKulkarni02/strix-ClaudeCode/feat/mcp-server/install-claude-code.sh | bash
+```
+
+That installs the `strix` CLI (installing `uv` first if needed) and registers it as a user-scoped MCP server. Then open Claude Code and ask it to test a target.
+
+To wire it up manually instead, register it with Claude Code:
+
+```bash
+claude mcp add strix -- strix mcp-server
+```
+
+Or add it to any client's MCP config as a `stdio` server whose command is `strix mcp-server`. Then ask the client to test a target — it drives Strix's tools directly. Docker must be running; the sandbox container starts on the first tool call that needs it.
+
+To review a local repository's code, mount it with `--target-path`; it appears **read-only** in the sandbox at `/workspace/<name>` for the shell tools to read:
+
+```bash
+cd /path/to/your-repo
+claude mcp add strix -- strix mcp-server --target-path "$PWD"
+```
+
+Without `--target-path`, the sandbox has no source mounted — good for URL, live-web, and network targets. `--target-path` is repeatable to mount several repos.
+
+This path uses the client's agent loop rather than Strix's tuned agent graph. To run the native graph, use `strix --target ...` with a configured LLM as usual. Only test targets you are authorized to test.
+
 **Recommended models for best results:**
 
 - [Z.ai GLM-5.3 on Vercel AI Gateway](https://vercel.com/ai-gateway/models/glm-5.3) - `vercel_ai_gateway/zai/glm-5.3` (recommended)

install-claude-code.sh

@@ -0,0 +1,73 @@
+#!/usr/bin/env bash
+#
+# One-line installer: run Strix as an MCP server inside Claude Code.
+#
+#   curl -LsSf https://raw.githubusercontent.com/IshanKulkarni02/strix-ClaudeCode/feat/mcp-server/install-claude-code.sh | bash
+#
+# It installs the `strix` CLI (via uv, installing uv first if needed) and
+# registers it with Claude Code so the agent can drive Strix's tools. No
+# STRIX_LLM or LLM_API_KEY is needed: Claude Code supplies the model.
+#
+# Requirements the script checks for: Claude Code (`claude`) and Docker.
+
+set -euo pipefail
+
+REPO="${STRIX_REPO:-github.com/IshanKulkarni02/strix-ClaudeCode}"
+BRANCH="${STRIX_BRANCH:-feat/mcp-server}"
+SPEC="git+https://${REPO}@${BRANCH}"
+
+bold() { printf '\033[1m%s\033[0m\n' "$1"; }
+info() { printf '  %s\n' "$1"; }
+warn() { printf '\033[33m!\033[0m %s\n' "$1"; }
+die()  { printf '\033[31m✗\033[0m %s\n' "$1" >&2; exit 1; }
+
+bold "Strix → Claude Code installer"
+
+# 1. Claude Code must be present (it is the model + orchestration loop).
+if ! command -v claude >/dev/null 2>&1; then
+  die "Claude Code not found. Install it first: https://claude.com/claude-code"
+fi
+info "Claude Code: $(command -v claude)"
+
+# 2. uv: install if missing, then make sure its bin dir is on PATH now.
+export PATH="$HOME/.local/bin:$PATH"
+if ! command -v uv >/dev/null 2>&1; then
+  info "Installing uv (Python package manager)…"
+  curl -LsSf https://astral.sh/uv/install.sh | sh
+  # shellcheck disable=SC1090
+  [ -f "$HOME/.local/bin/env" ] && . "$HOME/.local/bin/env"
+fi
+command -v uv >/dev/null 2>&1 || die "uv install failed; see https://docs.astral.sh/uv/"
+info "uv: $(command -v uv)"
+
+# 3. Install the strix CLI from the fork branch as a global tool.
+info "Installing strix from ${REPO}@${BRANCH} (first run downloads deps)…"
+uv tool install --force "$SPEC" >/dev/null
+# Resolve the binary uv just created explicitly — do not trust PATH order, in
+# case an official strix (e.g. ~/.strix/bin/strix) shadows this one.
+UV_BIN_DIR="$(uv tool dir --bin 2>/dev/null || echo "$HOME/.local/bin")"
+STRIX_BIN="$UV_BIN_DIR/strix"
+[ -x "$STRIX_BIN" ] || die "strix was not installed at $STRIX_BIN"
+info "strix: $STRIX_BIN"
+
+# 4. Register the MCP server with Claude Code (user scope = all projects).
+#    Use the absolute path so it resolves regardless of the shell's PATH.
+claude mcp remove strix -s user >/dev/null 2>&1 || true
+claude mcp add strix -s user -- "$STRIX_BIN" mcp-server
+info "Registered 'strix' as an MCP server (user scope)."
+
+# 5. Docker is needed for the sandbox; warn but don't fail if it is down.
+if command -v docker >/dev/null 2>&1; then
+  if docker info >/dev/null 2>&1; then
+    info "Docker: running."
+  else
+    warn "Docker is installed but not running — start it before the first scan."
+  fi
+else
+  warn "Docker not found — install and start Docker Desktop before the first scan."
+fi
+
+printf '\n'
+bold "Done."
+info "Open Claude Code and ask, e.g.:  \"use strix to test https://example.com\""
+info "The sandbox container is pulled automatically on the first tool call."

strix/interface/main.py

@@ -65,6 +65,7 @@ Additional commands:
   strix auth ...           Manage model-subscription sign-in
   strix view [RUN]         View a completed or running scan
   strix completions SHELL  Generate zsh, bash, or fish tab completion
+  strix mcp-server         Serve Strix's tools to an MCP client over stdio
 """
 
 
@@ -469,6 +470,14 @@ def main() -> None:
 
         sys.exit(run_cloud(sys.argv[2:]))
 
+    # `strix mcp-server` serves Strix's tools to an MCP client over stdio. The
+    # client supplies the model, so this path needs no LLM configuration and
+    # skips the scan argument parser entirely.
+    if len(sys.argv) > 1 and sys.argv[1] == "mcp-server":
+        from strix.interface.mcp_server import run_mcp_server
+
+        sys.exit(run_mcp_server(sys.argv[2:]))
+
     start_import_warmup()
 
     args = parse_arguments()

strix/interface/mcp_server.py

@@ -0,0 +1,586 @@
+"""``strix mcp-server`` — expose Strix's tools over the MCP stdio transport.
+
+Lets an MCP client (Claude Code, Cursor, Zed, ...) call Strix's existing tools
+directly. The client supplies the model and the orchestration loop, so this
+mode needs neither ``STRIX_LLM`` nor ``LLM_API_KEY``. Strix contributes the
+sandbox container, the proxy, the skill packs, and the reporting tools.
+
+This is additive: ``strix --target ...`` still runs the native agent graph
+against a configured LLM, unchanged.
+
+Strix's orchestration and bookkeeping tools (agent graph, ``finish_scan``,
+``wait_for_user``, notes, todos, ``think``) are intentionally not exposed — the
+MCP client brings its own planning, memory, and control flow.
+
+stdio transport owns stdout, so all logging here goes to stderr.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import contextlib
+import json
+import logging
+import sys
+import uuid
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+# ExecCommandTool / WriteStdinTool are not re-exported from the capabilities
+# package, so they are imported from their defining module. Strix pins the
+# agents SDK to >=0.19,<0.20 (see pyproject), so this internal path is stable
+# for the supported SDK range.
+from agents.sandbox.capabilities.shell import (  # type: ignore[attr-defined]
+    ExecCommandTool,
+    WriteStdinTool,
+)
+from agents.tool_context import ToolContext
+from mcp.server import Server
+from mcp.server.stdio import stdio_server
+from mcp.types import TextContent
+from mcp.types import Tool as MCPTool
+
+from strix.config import load_settings
+from strix.core.paths import run_dir_for, runtime_state_dir
+from strix.interface.cli_args import get_version
+from strix.interface.utils import derive_local_base_name
+from strix.report.state import (
+    ReportState,
+    get_global_report_state,
+    set_global_report_state,
+)
+from strix.runtime import session_manager
+from strix.tools.coverage.tools import (
+    hydrate_coverage_from_disk,
+    list_coverage,
+    record_coverage,
+    update_coverage,
+)
+from strix.tools.load_skill.tool import load_skill
+from strix.tools.proxy.tools import (
+    list_requests,
+    list_sitemap,
+    repeat_request,
+    scope_rules,
+    view_request,
+    view_sitemap_entry,
+)
+from strix.tools.reporting.tool import (
+    create_dependency_report,
+    create_vulnerability_report,
+    delete_vulnerability_report,
+    get_report,
+    list_reports,
+    update_vulnerability_report,
+)
+from strix.tools.threat_model.tools import (
+    amend_threat_model,
+    get_threat_model,
+    hydrate_threat_models_from_disk,
+    save_threat_model,
+)
+from strix.tools.web_search.tool import web_get_contents, web_search
+
+
+if TYPE_CHECKING:
+    from agents.tool import FunctionTool
+
+
+logger = logging.getLogger(__name__)
+
+SERVER_NAME = "strix"
+_CALL_ID_PREFIX = "mcp"
+
+# Cap on a single tool result sent to the client; larger output is trimmed to a
+# preview and the full text is spilled to the run directory (see ``_invoke``).
+_MAX_RESULT_CHARS = 50_000
+
+# Subdirectory under the run dir where oversized results are spilled, and the
+# name of the exposed tool that reads them back.
+_SPILL_SUBDIR = "mcp_outputs"
+_READ_OUTPUT_TOOL = "read_tool_output"
+
+_INSTRUCTIONS = (
+    "Strix tools over MCP. Shell tools (exec_command, write_stdin) run inside "
+    "an isolated Strix sandbox container that is started on first use. Proxy "
+    "tools expose the HTTP history captured by the sandbox's built-in proxy. "
+    "Reporting tools record CVSS-scored findings. load_skill loads Strix's "
+    "internal knowledge packs. Only operate against targets you are authorized "
+    "to test."
+)
+
+
+def _host_tools() -> list[FunctionTool]:
+    """Host-side tools that need no live sandbox session.
+
+    These read their run state from a plain ``ctx.context`` dict and degrade
+    gracefully when optional keys (coordinator, caido client) are absent, so
+    they are safe to expose with the minimal context this server builds.
+    """
+    return [
+        load_skill,
+        web_search,
+        web_get_contents,
+        list_requests,
+        view_request,
+        repeat_request,
+        list_sitemap,
+        view_sitemap_entry,
+        scope_rules,
+        create_vulnerability_report,
+        update_vulnerability_report,
+        delete_vulnerability_report,
+        create_dependency_report,
+        get_report,
+        list_reports,
+        get_threat_model,
+        save_threat_model,
+        amend_threat_model,
+        list_coverage,
+        record_coverage,
+        update_coverage,
+    ]
+
+
+# Sandbox tools run commands inside the container. Their input schemas are
+# static, so they can be advertised without a live session; only execution
+# needs one. apply_patch is a CustomTool (a different invoke contract) and
+# view_image returns image bytes that don't fit an MCP text result, so neither
+# is exposed here — the shell covers file edits and inspection inside the box.
+def _sandbox_tool_specs() -> list[FunctionTool]:
+    """Unbound sandbox tools, built only to read their name and schema."""
+    return [
+        ExecCommandTool(session=None),  # type: ignore[arg-type]  # schema-only, never invoked
+        WriteStdinTool(session=None),  # type: ignore[arg-type]  # schema-only, never invoked
+    ]
+
+
+SANDBOX_TOOL_NAMES = frozenset(t.name for t in _sandbox_tool_specs())
+
+# Proxy tools read their Caido client from the live sandbox session; the host
+# branch must bring the sandbox up before invoking one, or they report the
+# proxy as unavailable until an unrelated shell call happens to start it.
+PROXY_TOOL_NAMES = frozenset(
+    t.name
+    for t in (
+        list_requests,
+        view_request,
+        repeat_request,
+        list_sitemap,
+        view_sitemap_entry,
+        scope_rules,
+    )
+)
+
+
+def _validate_run_name(run_name: str) -> str:
+    """Return ``run_name`` if it is a single safe path segment, else raise.
+
+    ``run_name`` becomes a directory under ``strix_runs/``. A value with a path
+    separator, a ``..`` segment, an absolute path, or a leading dot could place
+    run files outside that tree or resume an unrelated run, so reject anything
+    that is not a plain name.
+    """
+    name = run_name.strip()
+    if not name:
+        raise ValueError("--run-name must not be empty")
+    if name in {".", ".."} or name.startswith("."):
+        raise ValueError(f"--run-name must not start with a dot: {run_name!r}")
+    if "/" in name or "\\" in name or Path(name).is_absolute():
+        raise ValueError(f"--run-name must be a single path segment: {run_name!r}")
+    if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in name):
+        raise ValueError("--run-name must not contain control characters")
+    return name
+
+
+def _init_run_state(run_name: str) -> Path:
+    """Wire up host-side run state so the reporting, threat-model, and coverage
+    tools persist to a run directory and survive a server restart, mirroring the
+    native CLI. Without it they return success but keep nothing on disk, and
+    ``list_reports`` stays empty.
+
+    ``run_name`` is stable across restarts (it is not the ephemeral sandbox id),
+    so relaunching the server with the same name resumes the same run directory
+    and its reports, coverage, and threat models. Returns that run directory.
+    """
+    run_dir = run_dir_for(run_name)
+    run_dir.mkdir(parents=True, exist_ok=True)
+    state_dir = runtime_state_dir(run_dir)
+    state_dir.mkdir(parents=True, exist_ok=True)
+
+    report_state = ReportState(run_name)
+    report_state.hydrate_from_run_dir()
+    set_global_report_state(report_state)
+
+    hydrate_coverage_from_disk(state_dir)
+    hydrate_threat_models_from_disk(state_dir)
+    logger.info("Run state ready at %s", run_dir)
+    return run_dir
+
+
+def _build_local_sources(target_paths: list[str]) -> list[dict[str, Any]]:
+    """Turn host repo paths into Strix ``local_sources`` mount entries.
+
+    Each path is mounted **read-only** at ``/workspace/<name>`` inside the
+    sandbox. Review needs to read the repository, not change it, and a
+    read-only bind stops a shell tool from modifying or deleting files in the
+    host working tree.
+    """
+    sources: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for raw in target_paths:
+        resolved = Path(raw).expanduser()
+        if not resolved.is_dir():
+            raise ValueError(f"--target-path is not a directory: {raw}")
+        # De-duplicate: two repos whose final path component matches would
+        # otherwise share one /workspace/<name> mount and shadow each other.
+        base = derive_local_base_name(str(resolved))
+        subdir = base
+        suffix = 1
+        while subdir in seen:
+            suffix += 1
+            subdir = f"{base}-{suffix}"
+        seen.add(subdir)
+        sources.append(
+            {
+                "source_path": str(resolved),
+                "workspace_subdir": subdir,
+                "read_only": True,
+            }
+        )
+    return sources
+
+
+class _SandboxTools:
+    """Lazily brings up one Strix sandbox session and binds the shell tools to
+    it. The session is created on the first call that needs it and reused for
+    the life of the server; ``aclose`` tears it down."""
+
+    def __init__(
+        self, local_sources: list[dict[str, Any]] | None = None, scan_id: str | None = None
+    ) -> None:
+        self._scan_id = scan_id or uuid.uuid4().hex[:8]
+        self._local_sources = local_sources or []
+        self._bundle: dict[str, Any] | None = None
+        self._tools: dict[str, FunctionTool] | None = None
+        self._lock = asyncio.Lock()
+
+    @property
+    def workspace_paths(self) -> list[str]:
+        """The ``/workspace/<name>`` paths of the mounted repos, if any."""
+        return [f"/workspace/{s['workspace_subdir']}" for s in self._local_sources]
+
+    async def ensure(self) -> dict[str, FunctionTool]:
+        async with self._lock:
+            if self._tools is not None:
+                return self._tools
+            self._tools = await self._bring_up()
+            return self._tools
+
+    async def _bring_up(self) -> dict[str, FunctionTool]:
+        settings = load_settings()
+        logger.info("Starting Strix sandbox session %s", self._scan_id)
+        self._bundle = await session_manager.create_or_reuse(
+            self._scan_id,
+            image=settings.runtime.image,
+            local_sources=self._local_sources,
+            status_sink=lambda phase: logger.info("sandbox: %s", phase),
+        )
+        session = self._bundle["session"]
+
+        tools: list[FunctionTool] = [ExecCommandTool(session=session)]
+        if session.supports_pty():
+            tools.append(WriteStdinTool(session=session))
+        logger.info("Sandbox ready; bound %d sandbox tool(s)", len(tools))
+        return {t.name: t for t in tools}
+
+    def caido_client(self) -> Any:
+        return self._bundle["caido_client"] if self._bundle else None
+
+    async def aclose(self) -> None:
+        if self._bundle is None:
+            return
+        with contextlib.suppress(Exception):
+            await session_manager.cleanup(self._scan_id)
+        logger.info("Sandbox session %s cleaned up", self._scan_id)
+
+
+def _run_context(sandbox: _SandboxTools) -> dict[str, Any]:
+    """The minimal ``ctx.context`` dict Strix tools read from.
+
+    Only the keys the exposed tools actually consult are populated; the rest of
+    Strix's run context (coordinator, agent graph) is absent and the tools fall
+    back to their no-coordinator paths.
+    """
+    return {
+        "caido_client": sandbox.caido_client(),
+        "agent_id": SERVER_NAME,
+        "parent_id": None,
+        "interactive": False,
+        "scan_targets": sandbox.workspace_paths,
+    }
+
+
+def _spill_large_output(tool_name: str, text: str) -> str | None:
+    """Persist an oversized tool result under the run directory and return the
+    id (file name) the client passes to ``read_tool_output`` to read it back."""
+    report_state = get_global_report_state()
+    if report_state is None:
+        return None
+    safe_tool = "".join(c if c.isalnum() or c in "-_" else "_" for c in tool_name)
+    output_id = f"{safe_tool}-{uuid.uuid4().hex[:8]}.txt"
+    try:
+        out_dir = report_state.get_run_dir() / _SPILL_SUBDIR
+        out_dir.mkdir(parents=True, exist_ok=True)
+        (out_dir / output_id).write_text(text, encoding="utf-8")
+    except OSError:
+        return None
+    return output_id
+
+
+def _read_spilled_output(output_id: str, offset: int, length: int) -> str:
+    """Read a previously spilled tool result back in bounded chunks.
+
+    Exposed as the ``read_tool_output`` MCP tool so a client limited to this
+    server's tools can still retrieve output that ``_invoke`` truncated. The id
+    must be a bare file name within the run's spill directory — no path parts.
+    """
+    if "/" in output_id or "\\" in output_id or ".." in output_id or not output_id.endswith(".txt"):
+        return f"Invalid output id: {output_id!r}"
+    report_state = get_global_report_state()
+    if report_state is None:
+        return "No run state is active, so there is no saved output to read."
+    path = report_state.get_run_dir() / _SPILL_SUBDIR / output_id
+    if not path.is_file():
+        return f"No saved output named {output_id!r}."
+    offset = max(offset, 0)
+    # Cap the window so one call can't return the whole file, and read only that
+    # window off disk (seek) rather than loading the whole file into memory.
+    length = min(length, _MAX_RESULT_CHARS) if length > 0 else _MAX_RESULT_CHARS
+    size = path.stat().st_size
+    with path.open("rb") as handle:
+        handle.seek(offset)
+        raw = handle.read(length)
+    chunk = raw.decode("utf-8", errors="replace")
+    end = offset + len(raw)
+    return f"{chunk}\n\n[bytes {offset}-{end} of {size}; more available: {end < size}]"
+
+
+async def _invoke(tool: FunctionTool, arguments: dict[str, Any], sandbox: _SandboxTools) -> str:
+    """Call a Strix FunctionTool and return its result as text.
+
+    A result larger than ``_MAX_RESULT_CHARS`` is trimmed to a bounded preview
+    so it cannot flood the client's context; the full text is written to the run
+    directory and the preview points the client at it.
+    """
+    ctx: ToolContext[Any] = ToolContext(
+        context=_run_context(sandbox),
+        tool_name=tool.name,
+        tool_call_id=f"{_CALL_ID_PREFIX}_{uuid.uuid4().hex[:12]}",
+        tool_arguments=json.dumps(arguments),
+    )
+    result = await tool.on_invoke_tool(ctx, json.dumps(arguments))
+    text = result if isinstance(result, str) else json.dumps(result, default=str)
+    if len(text) <= _MAX_RESULT_CHARS:
+        return text
+    output_id = _spill_large_output(tool.name, text)
+    if output_id:
+        preview = text[:_MAX_RESULT_CHARS]
+        return (
+            f"{preview}\n\n[truncated to {_MAX_RESULT_CHARS} of {len(text)} chars; "
+            f"call {_READ_OUTPUT_TOOL} with output_id={output_id!r} to read the rest]"
+        )
+    # The save failed (no run dir, disk full, unwritable), so there is nothing to
+    # page back. Show both ends rather than silently dropping the tail.
+    half = _MAX_RESULT_CHARS // 2
+    head = text[:half]
+    tail = text[-half:]
+    return (
+        f"{head}\n\n[could not save the full output ({len(text)} chars); the middle "
+        f"is omitted — showing the first and last {half} chars]\n\n{tail}"
+    )
+
+
+def _server_instructions(sandbox: _SandboxTools) -> str:
+    if not sandbox.workspace_paths:
+        return _INSTRUCTIONS
+    mounted = ", ".join(sandbox.workspace_paths)
+    return (
+        f"{_INSTRUCTIONS} The target repository is mounted in the sandbox at: "
+        f"{mounted}. Run the shell tools there for white-box review."
+    )
+
+
+def _build_server(sandbox: _SandboxTools) -> Any:
+    server: Any = Server(
+        SERVER_NAME, version=get_version(), instructions=_server_instructions(sandbox)
+    )
+
+    host_tools = {t.name: t for t in _host_tools()}
+
+    def _as_mcp_tool(tool: FunctionTool) -> MCPTool:
+        return MCPTool(
+            name=tool.name,
+            description=(tool.description or "").strip() or tool.name,
+            inputSchema=tool.params_json_schema or {"type": "object", "properties": {}},
+        )
+
+    read_output_tool = MCPTool(
+        name=_READ_OUTPUT_TOOL,
+        description=(
+            "Read back a tool result that was truncated, using the output_id "
+            "from the truncation notice. Returns a bounded chunk; page with "
+            f"byte offset and length (length is capped at {_MAX_RESULT_CHARS})."
+        ),
+        inputSchema={
+            "type": "object",
+            "properties": {
+                "output_id": {"type": "string", "description": "Id from the truncation notice."},
+                "offset": {"type": "integer", "minimum": 0, "default": 0},
+                "length": {
+                    "type": "integer",
+                    "minimum": 1,
+                    "maximum": _MAX_RESULT_CHARS,
+                    "default": _MAX_RESULT_CHARS,
+                },
+            },
+            "required": ["output_id"],
+        },
+    )
+
+    @server.list_tools()  # type: ignore[untyped-decorator]  # mcp SDK decorators are untyped
+    async def list_tools() -> list[MCPTool]:
+        # Sandbox tools are advertised from their static schemas, so listing
+        # never starts the container; the session is created on first call.
+        host = [_as_mcp_tool(t) for t in host_tools.values()]
+        return [*host, read_output_tool, *(_as_mcp_tool(t) for t in _sandbox_tool_specs())]
+
+    @server.call_tool()  # type: ignore[untyped-decorator]  # mcp SDK decorators are untyped
+    async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
+        tool: FunctionTool | None
+        if name == _READ_OUTPUT_TOOL:
+            args = arguments or {}
+            text = _read_spilled_output(
+                str(args.get("output_id", "")),
+                int(args.get("offset", 0) or 0),
+                int(args.get("length", _MAX_RESULT_CHARS) or _MAX_RESULT_CHARS),
+            )
+            return [TextContent(type="text", text=text)]
+        if name in host_tools:
+            tool = host_tools[name]
+            # Proxy tools need the live session's Caido client; start it first.
+            if name in PROXY_TOOL_NAMES:
+                await sandbox.ensure()
+        elif name in SANDBOX_TOOL_NAMES:
+            tool = (await sandbox.ensure()).get(name)
+            if tool is None:
+                # Advertised, but this session doesn't provide it (e.g. no PTY).
+                raise ValueError(f"tool not available in this sandbox session: {name}")
+        else:
+            raise ValueError(f"unknown tool: {name}")
+        text = await _invoke(tool, arguments or {}, sandbox)
+        return [TextContent(type="text", text=text)]
+
+    return server
+
+
+async def _serve(local_sources: list[dict[str, Any]], run_name: str) -> None:
+    # Run state is keyed by the stable run_name so restarts resume it; the
+    # sandbox container is ephemeral and keeps its own random id.
+    _init_run_state(run_name)
+    sandbox = _SandboxTools(local_sources=local_sources)
+    server = _build_server(sandbox)
+    init_options = server.create_initialization_options()
+    try:
+        async with stdio_server() as (read_stream, write_stream):
+            await server.run(read_stream, write_stream, init_options)
+    finally:
+        await sandbox.aclose()
+
+
+def _configure_logging(*, verbose: bool) -> None:
+    """Raise Strix's log level for this command and send logs to stderr.
+
+    ``main`` already called ``setup_console_logging``, which attaches a quiet
+    stderr handler (ERROR, with an INFO/DEBUG filter for the TUI) to the
+    ``strix`` logger and sets ``propagate=False`` — so ``logging.basicConfig`` on
+    the root would never change what this command prints. There is no TUI in
+    server mode, so replace that handler with one that actually emits at the
+    requested level. stdout carries the MCP protocol, so logs must use stderr.
+    """
+    level = logging.DEBUG if verbose else logging.INFO
+    strix_logger = logging.getLogger("strix")
+    strix_logger.setLevel(level)
+    for handler in list(strix_logger.handlers):
+        if isinstance(handler, logging.StreamHandler) and not isinstance(
+            handler, logging.FileHandler
+        ):
+            strix_logger.removeHandler(handler)
+    handler = logging.StreamHandler(sys.stderr)
+    handler.setLevel(level)
+    handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
+    strix_logger.addHandler(handler)
+    strix_logger.propagate = False
+
+
+def run_mcp_server(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(
+        prog="strix mcp-server",
+        description="Serve Strix's tools to an MCP client over stdio.",
+    )
+    parser.add_argument(
+        "-t",
+        "--target-path",
+        dest="target_paths",
+        action="append",
+        default=[],
+        metavar="PATH",
+        help=(
+            "Mount a local repository into the sandbox at /workspace/<name> so "
+            "the shell tools can review its code. Repeatable."
+        ),
+    )
+    parser.add_argument(
+        "--run-name",
+        dest="run_name",
+        default=None,
+        metavar="NAME",
+        help=(
+            "Run name for persisted reports, coverage, and threat models under "
+            "strix_runs/<name>. Pass the same name to resume that run; reusing "
+            "it also shares that run's files, so give concurrent servers "
+            "distinct names. Defaults to a unique name per launch."
+        ),
+    )
+    parser.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        help="Log at DEBUG level to stderr.",
+    )
+    args = parser.parse_args(argv)
+
+    _configure_logging(verbose=args.verbose)
+
+    try:
+        local_sources = _build_local_sources(args.target_paths)
+    except ValueError as exc:
+        parser.error(str(exc))
+
+    # A unique default keeps two servers launched from the same directory from
+    # sharing one run and overwriting each other's reports; pass --run-name to
+    # deliberately resume or share a run.
+    if args.run_name is not None:
+        try:
+            run_name = _validate_run_name(args.run_name)
+        except ValueError as exc:
+            parser.error(str(exc))
+    else:
+        run_name = f"mcp-{uuid.uuid4().hex[:8]}"
+
+    try:
+        asyncio.run(_serve(local_sources, run_name))
+    except KeyboardInterrupt:
+        return 0
+    return 0

tests/test_mcp_server.py

@@ -0,0 +1,400 @@
+"""Tests for ``strix mcp-server`` — the MCP stdio bridge.
+
+These cover the parts that need no Docker sandbox: the host-tool registry, the
+MCP tool advertisement, and the invoke path that calls a Strix ``FunctionTool``
+through a minimal run context. Sandbox bring-up is exercised separately and is
+not required here.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import shutil
+import subprocess
+from typing import TYPE_CHECKING
+
+import pytest
+from agents.tool import FunctionTool
+from mcp.shared.memory import create_connected_server_and_client_session
+
+from strix.interface import mcp_server
+from strix.report.state import get_global_report_state, set_global_report_state
+from strix.tools.load_skill.tool import load_skill
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from agents.tool_context import ToolContext
+
+
+def test_host_tools_are_function_tools_with_schemas() -> None:
+    tools = mcp_server._host_tools()
+    assert tools, "expected at least one host tool"
+    for tool in tools:
+        assert isinstance(tool, FunctionTool)
+        assert tool.name
+        assert isinstance(tool.params_json_schema, dict)
+
+
+def test_host_tools_have_unique_names() -> None:
+    names = [t.name for t in mcp_server._host_tools()]
+    assert len(names) == len(set(names))
+
+
+def test_orchestration_tools_are_not_exposed() -> None:
+    names = {t.name for t in mcp_server._host_tools()}
+    # The MCP client owns planning, control flow, and memory; Strix's
+    # equivalents must not be advertised or they fight the client.
+    for banned in ("create_agent", "finish_scan", "wait_for_user", "think", "create_todo"):
+        assert banned not in names
+
+
+def test_run_context_has_no_coordinator() -> None:
+    # The exposed tools must degrade gracefully with no agent graph present.
+    ctx = mcp_server._run_context(mcp_server._SandboxTools())
+    assert ctx["agent_id"] == mcp_server.SERVER_NAME
+    assert "coordinator" not in ctx
+    assert ctx["interactive"] is False
+
+
+@pytest.mark.asyncio
+async def test_invoke_returns_text_for_bad_input() -> None:
+    # load_skill with an invalid argument should come back as a graceful string,
+    # never raise — the invoke path must always yield text for MCP.
+    sandbox = mcp_server._SandboxTools()
+    result = await mcp_server._invoke(load_skill, {"skills": 12345}, sandbox)
+    assert isinstance(result, str)
+    assert result
+
+
+@pytest.mark.asyncio
+async def test_invoke_serializes_non_string_results() -> None:
+    # A tool returning a non-string must be JSON-encoded, not stringified ad hoc.
+    async def _fake_invoke(_ctx: ToolContext[object], _raw: str) -> dict[str, int]:
+        return {"ok": 1}
+
+    fake = FunctionTool(
+        name="fake",
+        description="fake",
+        params_json_schema={"type": "object", "properties": {}},
+        on_invoke_tool=_fake_invoke,
+    )
+    sandbox = mcp_server._SandboxTools()
+    result = await mcp_server._invoke(fake, {}, sandbox)
+    assert json.loads(result) == {"ok": 1}
+
+
+def test_sandbox_specs_readable_without_a_session() -> None:
+    # Advertising sandbox tools must not require a live container.
+    specs = mcp_server._sandbox_tool_specs()
+    names = {t.name for t in specs}
+    assert names == set(mcp_server.SANDBOX_TOOL_NAMES)
+    assert names == {"exec_command", "write_stdin"}
+    for tool in specs:
+        assert isinstance(tool.params_json_schema, dict)
+
+
+@pytest.mark.asyncio
+async def test_list_tools_handshake_does_not_start_sandbox() -> None:
+    # A client lists tools on connect; that must complete over a real MCP
+    # session without ever bringing up the Docker sandbox.
+    class _NoBootSandbox(mcp_server._SandboxTools):
+        async def ensure(self) -> dict[str, FunctionTool]:
+            raise AssertionError("list_tools must not bring up the sandbox")
+
+    server = mcp_server._build_server(_NoBootSandbox())
+    async with create_connected_server_and_client_session(server) as client:
+        listed = await client.list_tools()
+
+    advertised = {t.name for t in listed.tools}
+    assert "exec_command" in advertised  # sandbox tool, advertised statically
+    assert "write_stdin" in advertised
+    assert "load_skill" in advertised  # host tool
+    # Orchestration tools stay hidden end-to-end.
+    assert "finish_scan" not in advertised
+
+
+@pytest.mark.asyncio
+async def test_call_host_tool_over_session_returns_text() -> None:
+    # A full client->server call of a host tool (no sandbox) returns text
+    # content and does not raise, even on invalid input.
+    server = mcp_server._build_server(mcp_server._SandboxTools())
+    async with create_connected_server_and_client_session(server) as client:
+        result = await client.call_tool("load_skill", {"skills": 123})
+    assert result.content
+    assert result.content[0].type == "text"
+    assert result.content[0].text
+
+
+@pytest.mark.asyncio
+async def test_call_unknown_tool_errors() -> None:
+    server = mcp_server._build_server(mcp_server._SandboxTools())
+    async with create_connected_server_and_client_session(server) as client:
+        result = await client.call_tool("no_such_tool", {})
+    assert result.isError
+
+
+def test_build_server_advertises_host_tools() -> None:
+    server = mcp_server._build_server(mcp_server._SandboxTools())
+    assert server.name == mcp_server.SERVER_NAME
+
+
+def test_run_mcp_server_rejects_unknown_flag() -> None:
+    with pytest.raises(SystemExit):
+        mcp_server.run_mcp_server(["--definitely-not-a-flag"])
+
+
+def test_build_local_sources_maps_dirs_to_mounts(tmp_path: Path) -> None:
+    repo = tmp_path / "my-repo"
+    repo.mkdir()
+    sources = mcp_server._build_local_sources([str(repo)])
+    assert len(sources) == 1
+    assert sources[0]["source_path"] == str(repo)
+    assert sources[0]["workspace_subdir"] == "my-repo"
+    # Review mounts are read-only so shell tools cannot alter the host repo.
+    assert sources[0]["read_only"] is True
+
+
+def test_build_local_sources_rejects_missing_dir(tmp_path: Path) -> None:
+    missing = tmp_path / "nope"
+    with pytest.raises(ValueError, match="not a directory"):
+        mcp_server._build_local_sources([str(missing)])
+
+
+def test_mounted_repo_appears_in_context_and_instructions(tmp_path: Path) -> None:
+    repo = tmp_path / "target-app"
+    repo.mkdir()
+    sources = mcp_server._build_local_sources([str(repo)])
+    sandbox = mcp_server._SandboxTools(local_sources=sources)
+    assert sandbox.workspace_paths == ["/workspace/target-app"]
+    # Tools scope to the mounted path, and the client is told where it is.
+    assert mcp_server._run_context(sandbox)["scan_targets"] == ["/workspace/target-app"]
+    assert "/workspace/target-app" in mcp_server._server_instructions(sandbox)
+
+
+def test_no_mount_leaves_instructions_unchanged() -> None:
+    sandbox = mcp_server._SandboxTools()
+    assert sandbox.workspace_paths == []
+    assert mcp_server._server_instructions(sandbox) == mcp_server._INSTRUCTIONS
+
+
+def test_build_local_sources_dedupes_colliding_names(tmp_path: Path) -> None:
+    # Two repos with the same final path component must get distinct mounts,
+    # or one would shadow the other at /workspace/<name>.
+    a = tmp_path / "one" / "api"
+    b = tmp_path / "two" / "api"
+    a.mkdir(parents=True)
+    b.mkdir(parents=True)
+    sources = mcp_server._build_local_sources([str(a), str(b)])
+    subdirs = [s["workspace_subdir"] for s in sources]
+    assert subdirs == ["api", "api-2"]
+    assert len(set(subdirs)) == 2
+
+
+def test_init_run_state_wires_global_report_state(tmp_path: Path, monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    # Without this, reporting/threat-model/coverage tools report success but
+    # persist nothing. _init_run_state must install a global report state and
+    # create the run directory the tools write to.
+    monkeypatch.chdir(tmp_path)
+    # Another test may have left a global state installed; swap ours in and put
+    # theirs back, so this test neither reads nor clobbers suite-wide state.
+    previous = get_global_report_state()
+    set_global_report_state(None)
+    try:
+        returned = mcp_server._init_run_state("mcp-test")
+        state = get_global_report_state()
+        assert state is not None
+        # run_name is used verbatim (stable across restarts), not a random id.
+        run_dir = tmp_path / "strix_runs" / "mcp-test"
+        assert returned == run_dir
+        assert run_dir.is_dir()
+        assert (run_dir / ".state").is_dir()
+    finally:
+        set_global_report_state(previous)
+
+
+@pytest.mark.asyncio
+async def test_proxy_tool_brings_up_sandbox() -> None:
+    # Proxy tools read the Caido client from the live session, so invoking one
+    # must trigger sandbox bring-up even though it is a host tool.
+    class _RecordingSandbox(mcp_server._SandboxTools):
+        def __init__(self) -> None:
+            super().__init__()
+            self.ensure_calls = 0
+
+        async def ensure(self) -> dict[str, FunctionTool]:
+            self.ensure_calls += 1
+            return {}
+
+    sandbox = _RecordingSandbox()
+    server = mcp_server._build_server(sandbox)
+    async with create_connected_server_and_client_session(server) as client:
+        # repeat_request takes a single required arg, so it clears schema
+        # validation and reaches the dispatch that must start the sandbox.
+        result = await client.call_tool("repeat_request", {"request_id": "missing"})
+    assert sandbox.ensure_calls == 1
+    assert result.content and result.content[0].type == "text"
+
+
+def test_proxy_tool_names_are_host_tools() -> None:
+    host_names = {t.name for t in mcp_server._host_tools()}
+    assert mcp_server.PROXY_TOOL_NAMES
+    assert host_names >= mcp_server.PROXY_TOOL_NAMES
+
+
+@pytest.mark.asyncio
+async def test_invoke_truncates_large_output(tmp_path: Path, monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    # A huge tool result is trimmed to a bounded preview, and the full text is
+    # written to the run directory so nothing is silently lost.
+    monkeypatch.chdir(tmp_path)
+    mcp_server._init_run_state("mcp-big")
+    try:
+        big = "x" * (mcp_server._MAX_RESULT_CHARS + 500)
+
+        class _BigTool:
+            name = "big"
+
+            async def on_invoke_tool(self, _ctx, _args):  # type: ignore[no-untyped-def]
+                return big
+
+        out = await mcp_server._invoke(_BigTool(), {}, mcp_server._SandboxTools())  # type: ignore[arg-type]
+        assert len(out) < len(big)
+        assert "truncated" in out
+        # The notice points at the retrieval tool, and the spill is readable.
+        assert mcp_server._READ_OUTPUT_TOOL in out
+        spilled = list((tmp_path / "strix_runs" / "mcp-big" / "mcp_outputs").glob("big-*.txt"))
+        assert spilled and spilled[0].read_text(encoding="utf-8") == big
+        back = mcp_server._read_spilled_output(spilled[0].name, 0, mcp_server._MAX_RESULT_CHARS)
+        assert back.startswith("x")
+        assert "more available: True" in back
+    finally:
+        set_global_report_state(None)
+
+
+def test_read_spilled_output_rejects_traversal() -> None:
+    assert "Invalid output id" in mcp_server._read_spilled_output("../secret.txt", 0, 100)
+    assert "Invalid output id" in mcp_server._read_spilled_output("a/b.txt", 0, 100)
+
+
+def test_validate_run_name_rejects_traversal_and_separators() -> None:
+    # A run name becomes a directory under strix_runs/; path parts must not escape.
+    for bad in ["../evil", "/abs/path", "a/b", "a\\b", "..", ".", ".hidden", ""]:
+        with pytest.raises(ValueError, match="run-name"):
+            mcp_server._validate_run_name(bad)
+    assert mcp_server._validate_run_name("  my-run  ") == "my-run"
+
+
+@pytest.mark.asyncio
+async def test_read_spilled_output_caps_length_and_pages(tmp_path: Path, monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    # An over-cap length request is clamped, and reads return only the window.
+    monkeypatch.chdir(tmp_path)
+    mcp_server._init_run_state("mcp-cap")
+    try:
+        body = "y" * (mcp_server._MAX_RESULT_CHARS * 2)
+        output_id = mcp_server._spill_large_output("demo", body)
+        assert output_id
+        over = mcp_server._read_spilled_output(output_id, 0, mcp_server._MAX_RESULT_CHARS * 10)
+        # Response body (minus the trailing notice line) never exceeds the cap.
+        chunk = over.split("\n\n[", 1)[0]
+        assert len(chunk) == mcp_server._MAX_RESULT_CHARS
+        assert "more available: True" in over
+    finally:
+        set_global_report_state(None)
+
+
+@pytest.mark.asyncio
+async def test_invoke_keeps_both_ends_when_save_fails() -> None:
+    # With no run state, the spill save fails; the tail must not be dropped.
+    set_global_report_state(None)
+    big = "A" * (mcp_server._MAX_RESULT_CHARS + 1000) + "ZZZEND"
+
+    class _BigTool:
+        name = "big"
+
+        async def on_invoke_tool(self, _ctx, _args):  # type: ignore[no-untyped-def]
+            return big
+
+    out = await mcp_server._invoke(_BigTool(), {}, mcp_server._SandboxTools())  # type: ignore[arg-type]
+    assert "could not save" in out
+    assert out.startswith("A")
+    assert out.rstrip().endswith("ZZZEND")  # tail preserved
+
+
+@pytest.mark.asyncio
+async def test_read_tool_output_is_advertised_and_callable(tmp_path: Path, monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    monkeypatch.chdir(tmp_path)
+    mcp_server._init_run_state("mcp-read")
+    try:
+        output_id = mcp_server._spill_large_output("demo", "full-body-content")
+        assert output_id
+        server = mcp_server._build_server(mcp_server._SandboxTools())
+        async with create_connected_server_and_client_session(server) as client:
+            listed = {t.name for t in (await client.list_tools()).tools}
+            assert mcp_server._READ_OUTPUT_TOOL in listed
+            result = await client.call_tool(mcp_server._READ_OUTPUT_TOOL, {"output_id": output_id})
+        assert result.content and "full-body-content" in result.content[0].text
+    finally:
+        set_global_report_state(None)
+
+
+def test_configure_logging_enables_debug_on_stderr() -> None:
+    strix_logger = logging.getLogger("strix")
+    saved = (strix_logger.level, list(strix_logger.handlers), strix_logger.propagate)
+    try:
+        mcp_server._configure_logging(verbose=True)
+        assert strix_logger.level == logging.DEBUG
+        streams = [
+            h
+            for h in strix_logger.handlers
+            if isinstance(h, logging.StreamHandler) and not isinstance(h, logging.FileHandler)
+        ]
+        assert streams
+        assert all(h.level == logging.DEBUG for h in streams)
+    finally:
+        strix_logger.setLevel(saved[0])
+        strix_logger.handlers = saved[1]
+        strix_logger.propagate = saved[2]
+
+
+def _docker_available() -> bool:
+    docker = shutil.which("docker")
+    if docker is None:
+        return False
+    try:
+        completed = subprocess.run(  # noqa: S603  # fixed `docker info`, resolved path
+            [docker, "info"], capture_output=True, timeout=10, check=False
+        )
+    except (OSError, subprocess.SubprocessError):
+        return False
+    return completed.returncode == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(not _docker_available(), reason="needs a running Docker daemon")
+async def test_sandbox_runs_shell_on_readonly_mount(tmp_path: Path, monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    # Real Docker-backed flow: bring the sandbox up with a mounted repo, run a
+    # shell command through the exposed tool, and confirm the mount is read-only.
+    repo = tmp_path / "proj"
+    repo.mkdir()
+    (repo / "hello.txt").write_text("mounted-content", encoding="utf-8")
+    monkeypatch.chdir(tmp_path)
+    mcp_server._init_run_state("mcp-it")
+    sources = mcp_server._build_local_sources([str(repo)])
+    sandbox = mcp_server._SandboxTools(local_sources=sources)
+    server = mcp_server._build_server(sandbox)
+    try:
+        async with create_connected_server_and_client_session(server) as client:
+            read = await client.call_tool("exec_command", {"cmd": "cat /workspace/proj/hello.txt"})
+            assert read.content and "mounted-content" in read.content[0].text
+            wrote = await client.call_tool(
+                "exec_command", {"cmd": "echo tampered > /workspace/proj/hello.txt"}
+            )
+            text = wrote.content[0].text.lower() if wrote.content else ""
+            assert "read-only" in text or "permission denied" in text
+        # Host file is untouched.
+        assert (repo / "hello.txt").read_text(encoding="utf-8") == "mounted-content"
+    finally:
+        await sandbox.aclose()
+        set_global_report_state(None)