From 67d6e8355c5430eca7569d504a9ea64c6aa8301d Mon Sep 17 00:00:00 2001 From: IshanKulkarni02 Date: Tue, 6 Oct 2026 13:57:45 +0530 Subject: [PATCH] feat(mcp): add `strix mcp-server` to serve Strix tools to MCP clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expose Strix's own tools over MCP stdio so a client such as Claude Code, Cursor, or Zed drives them with its own model and agent loop. This mode needs no STRIX_LLM and no LLM_API_KEY — the client supplies the model, so an existing coding-agent subscription can run Strix. Strix still provides its sandbox container, HTTP proxy, skill packs, and CVSS-scored reporting tools. - `strix mcp-server`: 21 host tools plus lazily-bound sandbox shell/FS tools, served over stdio. The sandbox session starts on the first tool call that needs it and is reused for the life of the server. - `--target-path` (repeatable): mount a local repository read-only at /workspace/ for white-box review. Repos whose final path component collides are de-duplicated (api, api-2, ...). - `--run-name`: run state (reporting, threat-model, coverage) persists under strix_runs/ and hydrates on startup, so passing a name resumes that run. The default is unique per launch so two servers started from the same directory do not share a run and overwrite each other's reports. - Proxy tools bring the sandbox up on first use so their Caido client is available without an unrelated shell call first. - Oversized tool results are trimmed to a bounded preview; the full text is spilled to the run directory and is retrievable through the exposed `read_tool_output` tool, so nothing is lost and the client is not flooded. - `--verbose` adjusts the installed stderr handlers directly (basicConfig could not), so DEBUG output shows; stdout stays the MCP transport. - Orchestration-owning tools (agent graph, finish_scan, wait_for_user, notes, todos, think) are intentionally not exposed; the client owns planning and memory. - Tests cover discovery, invocation, context wiring, mounting, de-dup, run-state setup, proxy bring-up, output truncation + retrieval, log config, and a Docker-gated integration test of the sandbox read-only mount. No new dependency: the `mcp` SDK is already transitive via openai-agents. --- README.md | 23 ++ strix/interface/main.py | 9 + strix/interface/mcp_server.py | 542 ++++++++++++++++++++++++++++++++++ tests/test_mcp_server.py | 353 ++++++++++++++++++++++ 4 files changed, 927 insertions(+) create mode 100644 strix/interface/mcp_server.py create mode 100644 tests/test_mcp_server.py diff --git a/README.md b/README.md index f93b2376..d6518df6 100644 --- a/README.md +++ b/README.md @@ -338,6 +338,29 @@ 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. + +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/` 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 OpenRouter](https://openrouter.ai/z-ai/glm-5.3) - `openrouter/z-ai/glm-5.3` (the default pick) diff --git a/strix/interface/main.py b/strix/interface/main.py index cc3a0321..31d4ad19 100644 --- a/strix/interface/main.py +++ b/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 """ @@ -468,6 +469,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() diff --git a/strix/interface/mcp_server.py b/strix/interface/mcp_server.py new file mode 100644 index 00000000..6ce7feb5 --- /dev/null +++ b/strix/interface/mcp_server.py @@ -0,0 +1,542 @@ +"""``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 _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/`` 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/ 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/`` 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) + length = length if length > 0 else _MAX_RESULT_CHARS + data = path.read_text(encoding="utf-8", errors="replace") + chunk = data[offset : offset + length] + end = offset + len(chunk) + return f"{chunk}\n\n[chars {offset}-{end} of {len(data)}; more available: {end < len(data)}]" + + +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) + how = ( + f"; call {_READ_OUTPUT_TOOL} with output_id={output_id!r} to read the rest" + if output_id + else "" + ) + preview = text[:_MAX_RESULT_CHARS] + return f"{preview}\n\n[truncated to {_MAX_RESULT_CHARS} of {len(text)} chars{how}]" + + +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 " + "offset and length." + ), + 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, "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/ 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/. 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. + run_name = args.run_name or f"mcp-{uuid.uuid4().hex[:8]}" + + try: + asyncio.run(_serve(local_sources, run_name)) + except KeyboardInterrupt: + return 0 + return 0 diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py new file mode 100644 index 00000000..9a70a4e4 --- /dev/null +++ b/tests/test_mcp_server.py @@ -0,0 +1,353 @@ +"""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/. + 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) + try: + assert get_global_report_state() is None + 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(None) + + +@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) + + +@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)