feat(mcp): add strix mcp-server to serve Strix tools to MCP clients

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/<name> 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/<name> 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.
This commit is contained in:
IshanKulkarni02 2026-10-06 13:57:45 +05:30
parent f1386cad37
commit 67d6e8355c
4 changed files with 927 additions and 0 deletions

View file

@ -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/<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 OpenRouter](https://openrouter.ai/z-ai/glm-5.3) - `openrouter/z-ai/glm-5.3` (the default pick)

View file

@ -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()

View file

@ -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/<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)
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/<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.
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

353
tests/test_mcp_server.py Normal file
View file

@ -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/<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)
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)