scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-08-28 05:25:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'upstream/main' into feat/i18n-spanish

Browse source 
# Conflicts:
#	strix/agents/prompts/system_prompt.jinja
This commit is contained in:
criss717 2026-08-27 17:57:56 +02:00
commit fe0ea5316f
33 changed files with 1980 additions and 961 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

@ -243,6 +243,8 @@ ignore = [
"tests/test_report_pdf.py" = ["S105", "S106"]
# Fake MCP server matches the SDK's MCPServer signature; its args are unused.
"tests/test_mcp_client.py" = ["S105", "S106", "ARG002"]
# MCP connection request in a test carries a dummy bearer token.
"tests/test_runner_root_prompt.py" = ["S106"]
# Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a
# circular dependency with strix.telemetry / strix.interface.viewer.report_pdf.
"strix/interface/viewer/server.py" = ["N802", "PLC0415"]

4
strix/agents/factory.py
View file

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

18
strix/agents/prompts/system_prompt.jinja
View file

@ -80,6 +80,22 @@ AUTHORIZED TARGETS:
{% endfor %}
{% endif %}
{% if system_prompt_context and system_prompt_context.mcp_available %}
MCP CONNECTIONS (available this run):
- The user connected one or more MCP (Model Context Protocol) servers — external tool providers you can reach on demand. Their individual tools do NOT appear in your tool list; three dispatch tools are the only way in.
{% if system_prompt_context.mcp_connections %}
- Connected this run (call describe_mcp on one to see its tools):
{% for connection in system_prompt_context.mcp_connections %}
- {{ connection.name }} ({{ connection.tool_count }} tools){% if connection.purpose %}: {{ connection.purpose }}{% endif %}
{% endfor %}
{% endif %}
- Reach for a connection whenever the target itself cannot give you information a connection could: its database schema and access policies, real deployment or infrastructure configuration, known issues or prior findings, or server logs. In those cases call list_mcps early to see what is available, and prefer a connection's authoritative data over inferring from the target's responses. Do not wait to be told a connection exists.
1. Call list_mcps() to discover the available connections.
2. Call describe_mcp(connection="<name>") to inspect one connection's tools, each with its name, description, and JSON input schema.
3. Call call_mcp(connection="<name>", tool="<tool>", arguments={...}) to run one, passing an arguments object that matches the schema (omit arguments for a tool that takes none).
- Do not assume a connection or tool exists; discover it with list_mcps and describe it with describe_mcp before calling.
{% endif %}
AUTHORIZATION STATUS:
- You have FULL AUTHORIZATION for authorized security validation on in-scope targets to help secure the target systems/app
- All permission checks have been COMPLETED and APPROVED - never question your authority
@ -224,7 +240,7 @@ VALIDATION REQUIREMENTS:
- CLOSURE DISCIPLINE: every candidate you open ends in exactly one explicit state — `confirmed` (working PoC, or a complete source→control→sink→impact trace that is reachable), `ruled_out` (you can name the SPECIFIC control, at a location, that runs on every attacker-reachable path before the sink), or `open_proof_gap` (plausible, unconfirmed, and you could NOT name such a control). "I moved on" is not a closure state. Silently dropping an uncertain candidate is mislabelling an `open_proof_gap` as `ruled_out` and is how real bugs get missed.
- Missing information is NOT proof of safety: no caller found, can't tell if deployed/exposed, couldn't stand up the service, build failed — each is an `open_proof_gap`, never a reason to mark a candidate clean. Difficulty is a reason to defer, not to suppress.
- COVERAGE: record every surface you assess with `record_coverage` (surface + risk area + outcome + evidence), including the ones that came back clean — a report that only lists findings cannot say what was reviewed and cleared. Use the `needs_follow_up` outcome for anything left in an `open_proof_gap` state, and carry the same items up in `agent_finish(open_items=[...])`. The ledger is shared and mutable: when you resolve a surface another agent left open — or find that a closed one is not — move that entry with `update_coverage` instead of recording a second one for the same surface. The root agent reconciles all of it via `list_coverage` before `finish_scan`.
- THREAT MODEL: before you start testing, call `get_threat_model` on the target you were pointed at — it is the scan's shared answer to who the attacker is, where the trust boundaries sit, and what counts as critical here, and it is cached per target rather than per scan. Read it instead of re-deriving trust boundaries yourself; where your testing disproves it — a boundary it calls trusted turns out to be attacker-reachable, a role it did not know about, a host or endpoint it never listed — record that with `amend_threat_model` so the agents after you inherit the correction. Amending is not optional politeness: a model nobody corrects turns the first agent's guesses into everyone's assumptions.
- THREAT MODEL: before you start testing, call `get_threat_model` on the target you were pointed at — it is the scan's shared answer to who the attacker is, where the trust boundaries sit, and what counts as critical here. It is scoped to this scan and nothing carries over from an earlier run, so `found: false` means no agent on this run has derived one yet. Read it instead of re-deriving trust boundaries yourself; where your testing disproves it — a boundary it calls trusted turns out to be attacker-reachable, a role it did not know about, a host or endpoint it never listed — record that with `amend_threat_model` so the agents after you inherit the correction. Amending is not optional politeness: a model nobody corrects turns the first agent's guesses into everyone's assumptions.
- Before filing any report, run the counterevidence pass: argue the strongest case AGAINST the finding, record what you found in the `counterevidence` field, set `confidence` honestly (a static-only trace you couldn't execute is at best `medium`), and state what evidence would change the severity. See the counterevidence and severity-calibration knowledge above.
- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)

104
strix/core/runner.py
View file

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

14
strix/interface/tui/internal/app/model_test.go
View file

@ -865,6 +865,20 @@ func TestPanelPaddingResetsLeakingLineBackground(t *testing.T) {
}
}
func TestFillBackgroundRestoresBaseForegroundAfterReset(t *testing.T) {
const textFG = "\x1b[38;2;212;212;212m"
view := "\x1b[38;2;167;139;250m◈ \x1b[0m\x1b[2mspawning\x1b[0m"
filled := fillBackground(view)
baseStyle := blackBG + textFG
if !strings.HasPrefix(filled, baseStyle) {
t.Fatalf("frame does not set its base colors: %q", filled)
}
if got, want := strings.Count(filled, "\x1b[0m"+baseStyle), 2; got != want {
t.Fatalf("base colors restored after %d resets, want %d: %q", got, want, filled)
}
}
func TestMainTraceTreeAndFindingsRenderScrollbars(t *testing.T) {
model := New(nil)
model.width, model.height = 150, 35

21
strix/interface/tui/internal/app/view.go
View file

@ -352,21 +352,28 @@ func (m Model) toastOverlay(view string) string {
return strings.Join(bg, "\n")
}
// blackBG is the SGR that selects a solid black background.
const blackBG = "\x1b[48;2;0;0;0m"
// Base frame colors are reapplied after full SGR resets so the TUI does not
// inherit an unreadable foreground from the user's terminal profile.
const (
blackBG = "\x1b[48;2;0;0;0m"
textFG = "\x1b[38;2;212;212;212m"
baseFrameColors = blackBG + textFG
)
// fillBackground paints the whole frame black like Textual's Screen background.
// Bubble Tea has no screen compositor, so any cell the view does not explicitly
// color shows the terminal's default background. lipgloss emits a full reset
// (\x1b[0m) at the end of every styled span, which also clears the background, so
// we reassert black after each reset (and at the start). Spans that set their own
// background — inline code, selected rows, buttons — keep it, because their color
// is emitted before the reset.
// (\x1b[0m) at the end of every styled span, which clears both foreground and
// background. Reasserting only black made uncolored and faint text inherit the
// terminal profile's foreground; light profiles therefore rendered that text
// black-on-black. Reapply both base colors after each reset (and at the start).
// Spans that set their own colors — inline code, selected rows, buttons — keep
// them, because their color is emitted after the base style.
func fillBackground(view string) string {
if view == "" {
return view
}
return blackBG + strings.ReplaceAll(view, "\x1b[0m", "\x1b[0m"+blackBG)
return baseFrameColors + strings.ReplaceAll(view, "\x1b[0m", "\x1b[0m"+baseFrameColors)
}
func (m Model) splashView() string {

16
strix/interface/tui/internal/render/coverage_test.go
View file

@ -96,14 +96,12 @@ func TestCoverageDuplicateRejectionSurfacesTheError(t *testing.T) {
requireContains(t, out, "/login", "already has coverage entry a1b2c3")
}
func TestGetThreatModelRendersStalenessAndAmendments(t *testing.T) {
func TestGetThreatModelRendersAmendments(t *testing.T) {
out := ansi.Strip(Tool(tool("get_threat_model",
map[string]any{"target": "https://app.example.com"},
map[string]any{
"success": true,
"found": true,
"stale": true,
"cached_revision": "0123456789abcdef",
"success": true,
"found": true,
"content": "# Overview\nMulti-tenant billing app.\n\n" +
"## Trust Boundaries and Assumptions\n\n## Attack Surface\n",
"amendments": []any{
@ -116,7 +114,6 @@ func TestGetThreatModelRendersStalenessAndAmendments(t *testing.T) {
"completed")))
requireContains(t, out,
"Threat Model", "https://app.example.com",
"stale", "01234567",
"1 amendment(s)", "ReconAgent", "staging host shares the production database",
"Multi-tenant billing app.", "Overview", "Trust Boundaries and Assumptions",
)
@ -126,7 +123,7 @@ func TestGetThreatModelMissingModelIsExplicit(t *testing.T) {
out := ansi.Strip(Tool(tool("get_threat_model",
map[string]any{"target": "10.0.0.5"},
map[string]any{"success": true, "found": false}, "completed")))
requireContains(t, out, "No model cached for this target yet")
requireContains(t, out, "No model derived for this target yet")
}
func TestSaveThreatModelWarnsWhenAmendmentsAreCleared(t *testing.T) {
@ -134,15 +131,10 @@ func TestSaveThreatModelWarnsWhenAmendmentsAreCleared(t *testing.T) {
map[string]any{"target": "app.example.com", "content": "# Overview\nA thing.\n"},
map[string]any{
"success": true,
"revision": "unversioned",
"amendments_cleared": 2,
},
"completed")))
requireContains(t, out, "Threat Model Saved", "saved", "cleared 2 amendment(s)")
// An unversioned target has no revision worth printing.
if strings.Contains(out, "unversioned") {
t.Fatalf("unversioned revision should not be rendered:\n%s", out)
}
}
func TestAmendThreatModelRendersAddendum(t *testing.T) {

12
strix/interface/tui/internal/render/mcp.go
View file

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

5
strix/interface/tui/internal/render/registry.go
View file

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

25
strix/interface/tui/internal/render/render_test.go
View file

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

21
strix/interface/tui/internal/render/threat_model.go
View file

@ -56,9 +56,6 @@ func renderThreatModel(name string, args map[string]any, result any) string {
threatModelBody(&b, StringValue(args["addendum"]))
default:
b.WriteString("\n " + Col(Green).Render("✓ saved"))
if revision := shortRevision(StringValue(m["revision"])); revision != "" {
b.WriteString(Dim().Render(" at " + revision))
}
// Saving folds amendments away, so the count that vanished is worth
// stating: it is the one destructive thing this tool does.
if cleared, ok := NumericValue(m["amendments_cleared"]); ok && cleared > 0 {
@ -72,15 +69,9 @@ func renderThreatModel(name string, args map[string]any, result any) string {
func threatModelReadBody(b *strings.Builder, result map[string]any) {
if !truthy(result["found"]) {
b.WriteString("\n " + Dim().Render("No model cached for this target yet"))
b.WriteString("\n " + Dim().Render("No model derived for this target yet"))
return
}
if truthy(result["stale"]) {
b.WriteString("\n " + Col(AmberY).Render("⚠ stale"))
if cached := shortRevision(StringValue(result["cached_revision"])); cached != "" {
b.WriteString(Dim().Render(" (written at " + cached + ")"))
}
}
if amendments, ok := result["amendments"].([]any); ok && len(amendments) > 0 {
b.WriteString("\n " + Col(Gold).Render("+ "+strconv.Itoa(len(amendments))+
" amendment(s)") + Dim().Render(" — later statements win"))
@ -126,13 +117,3 @@ func threatModelBody(b *strings.Builder, content string) {
b.WriteString("\n " + Dim().Render(strings.Join(headings, " · ")))
}
}
// shortRevision abbreviates a git sha; "unversioned" targets have no revision
// worth showing.
func shortRevision(revision string) string {
revision = strings.TrimSpace(revision)
if revision == "" || revision == "unversioned" {
return ""
}
return firstN(revision, 8)
}

49
strix/interface/tui/live_view.py
View file

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

14
strix/interface/tui/runtime.py
View file

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

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

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

18
strix/interface/viewer/frontend/src/components/live/tool-renderers/ThreatModelRenderer.tsx
View file

@ -16,13 +16,6 @@ const ACTION_LABELS: Record<string, { label: string; Icon: typeof Crosshair }> =
amend_threat_model: { label: "Threat model amended", Icon: Plus },
};
/** A git sha is noise past its first bytes, and "unversioned" is not a revision. */
function shortRevision(revision: unknown): string {
const value = typeof revision === "string" ? revision.trim() : "";
if (!value || value === "unversioned") return "";
return value.slice(0, 8);
}
export default function ThreatModelRenderer({ toolName, args, result }: ToolRendererProps) {
const action = ACTION_LABELS[toolName] ?? { label: "Threat model", Icon: Crosshair };
const ActionIcon = action.Icon;
@ -59,22 +52,15 @@ export default function ThreatModelRenderer({ toolName, args, result }: ToolRend
return (
<div>
{header}
<div className="mt-1.5 text-[#555] text-xs">No model cached for this target yet</div>
<div className="mt-1.5 text-[#555] text-xs">No model derived for this target yet</div>
</div>
);
}
const rawAmendments = structured?.amendments;
const amendments: Amendment[] = Array.isArray(rawAmendments) ? (rawAmendments as Amendment[]) : [];
const cachedRevision = shortRevision(structured?.cached_revision);
return (
<div>
{header}
{structured?.stale === true && (
<div className="mt-1.5 flex items-center gap-1.5 text-yellow-400/80 text-xs">
<AlertTriangle className="w-3 h-3 shrink-0" />
<span>stale{cachedRevision ? ` — written at ${cachedRevision}` : ""}</span>
</div>
)}
{amendments.length > 0 && (
<div className="mt-2">
<span className="text-amber-400/70 text-xs font-semibold">
@ -119,12 +105,10 @@ export default function ThreatModelRenderer({ toolName, args, result }: ToolRend
}
const cleared = (structured?.amendments_cleared as number | undefined) ?? 0;
const revision = shortRevision(structured?.revision);
const content = (args.content as string) ?? "";
return (
<div>
{header}
{revision && <div className="mt-1.5 text-[#666] font-mono text-xs">at {revision}</div>}
{/* Saving folds amendments away — the one destructive thing this tool does. */}
{cleared > 0 && (
<div className="mt-1.5 flex items-center gap-1.5 text-yellow-400/80 text-xs">

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

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

7
strix/interface/viewer/frontend/src/types/events.ts
View file

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

2
strix/interface/viewer/static/assets/index-C9c1WbvP.js → strix/interface/viewer/static/assets/index-CYf9nnT3.js
View file

File diff suppressed because one or more lines are too long

2
strix/interface/viewer/static/index.html
View file

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

2
strix/skills/coordination/root_agent.md
View file

@ -27,7 +27,7 @@ Before spawning agents, analyze the target from the scan config/scope and any pr
## Establish the Threat Model
Every scan needs one shared answer to "who is the attacker here, and what are they attacking" — black-box or white-box. Without it, five agents derive five different answers and their findings cannot be reconciled. Call `get_threat_model` on the target (a host, a URL, or a repository path) before you spawn hunters; if nothing is cached, derive one and persist it with `save_threat_model`. It is cached per target, so a later scan of the same host or tree reads it back instead of paying for it twice, and a model written from source is read back by an agent testing the deployment.
Every scan needs one shared answer to "who is the attacker here, and what are they attacking" — black-box or white-box. Without it, five agents derive five different answers and their findings cannot be reconciled. Call `get_threat_model` on the target (a host, a URL, or a repository path) before you spawn hunters; if no model exists yet, derive one and share it with `save_threat_model`. It lives for this scan only — nothing carries over from an earlier run, so every scan derives its own — but within the run every agent reads the same document, and a model written from source is read back by an agent testing the deployment.
**When the target includes a repository**, derive it up front: the code tells you the boundaries, entrypoints, and controls before you send a single request.

39
strix/tools/mcp/__init__.py
View file

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

173
strix/tools/mcp/agent_tools.py Normal file
View file

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

360
strix/tools/mcp/client.py
View file

@ -1,14 +1,15 @@
"""Connect to MCP servers and expose their tools to the agent.
"""Connect to MCP servers so a run can reach their tools on demand.
Given one :class:`McpConnectionConfig` per server, :func:`connect_mcp_servers`
lists each server's tools, keeps the ones on the connection's allowlist (or all
of them when none is set), prefixes each with the connection name so servers do
not collide, and registers them through the agent factory. The factory applies
output bounding, per-call timeouts, and structured errors to every registered
tool, so this layer does not reimplement them.
connects each server, counts the tools it offers (honoring the connection's
allowlist), and returns the live sessions. It does NOT register anything as an
agent tool: under the generic-dispatch model the run holds these sessions in a
per-run :class:`~strix.tools.mcp.registry.McpRegistry`, and the agent reaches
them through the two dispatch tools (``describe_mcp`` / ``call_mcp``), which call
:func:`dispatch_mcp_call` here to run one tool and serialize its result.
A server that cannot connect, or a tool set that cannot be registered, is logged
and skipped, so one bad connection never fails the run.
A server that cannot connect is logged and skipped, so one bad connection never
fails the run.
"""
from __future__ import annotations
@ -16,36 +17,34 @@ from __future__ import annotations
import contextlib
import json
import logging
import os
from pathlib import Path
from typing import TYPE_CHECKING, Any, NamedTuple, cast
from agents.exceptions import ModelBehaviorError
from agents.mcp import (
MCPServer,
MCPServerStdio,
MCPServerStdioParams,
MCPServerStreamableHttp,
MCPServerStreamableHttpParams,
MCPUtil,
create_static_tool_filter,
)
from strix.agents.factory import register_agent_tools
from strix.tools.mcp.naming import namespaced_tool_name
from mcp.client.stdio import stdio_client
if TYPE_CHECKING:
from collections.abc import Callable
from agents.tool import FunctionTool, Tool
from mcp.types import Tool as MCPTool
from strix.tools.mcp.config import McpConnectionConfig
from strix.tools.mcp.registry import McpConnectionRequest, McpRegistry
# Runs on each tool's structured result before it reaches the agent. Called
# ``result_transform(namespaced_tool_name, structured_result)`` and its return
# value becomes the tool's output. ``structured_result`` is the parsed
# ``CallToolResult`` as a dict (not a serialized string), so the transform can
# project or drop individual fields.
# Runs on one tool call's structured result before it reaches the agent.
# Called ``result_transform(label, structured_result)`` and its return value
# becomes the tool's output. ``label`` is the model-facing
# ``<connection>_<tool>`` name so a transform keyed on names still resolves
# the same way it did under per-tool registration; ``structured_result`` is
# the parsed ``CallToolResult`` as a dict (not a serialized string), so the
# transform can project or drop individual fields.
ResultTransform = Callable[[str, Any], Any]
@ -53,12 +52,14 @@ logger = logging.getLogger(__name__)
class ConnectedMcpServer(NamedTuple):
"""One successfully connected MCP server and how many tools it registered.
"""One successfully connected MCP server and how many tools it offers.
``server`` is kept so the caller can clean it up when the run ends;
``name`` and ``tool_count`` let the caller show the user a startup summary;
``server`` is kept so the caller can clean it up when the run ends, and so
the caller can hand the live session to the run's
:class:`~strix.tools.mcp.registry.McpRegistry`; ``name`` and ``tool_count``
let the caller show the user a startup summary and fill the prompt inventory;
``notes`` carries the connection's optional free-text description so the
caller can surface it to the agent as context about the connection.
caller can surface it as the connection's purpose in the inventory.
"""
server: MCPServer
@ -75,13 +76,42 @@ def _auth_headers(config: McpConnectionConfig) -> dict[str, str]:
return {"Authorization": f"Bearer {auth.token}"}
@contextlib.asynccontextmanager
async def _quiet_stdio_streams(params: Any) -> Any:
"""Run a stdio MCP server with its stderr sent to the void.
A stdio MCP server chats on stderr as it boots (the filesystem server, for
one, prints ``Allowed directories: [ ... ]``). The mcp library forwards that
stderr to the parent's ``sys.stderr`` by default, which is the terminal the
TUI is drawing on, so the banner corrupts the display. Pointing ``errlog`` at
``os.devnull`` drops that chatter. Connection failures are unaffected: they
still raise from ``connect`` and are logged by :func:`connect_mcp_servers`.
"""
with Path(os.devnull).open("w", encoding="utf-8") as errlog:
async with stdio_client(params, errlog=errlog) as streams:
yield streams
class _QuietMCPServerStdio(MCPServerStdio):
"""``MCPServerStdio`` whose subprocess stderr is kept off the terminal.
The SDK's ``create_streams`` calls ``stdio_client(self.params)`` with no
``errlog``, so the subprocess stderr defaults to ``sys.stderr`` and paints
server banners over the running TUI. Overriding it lets us redirect that
stream; everything else about the stdio transport is unchanged.
"""
def create_streams(self) -> Any:
return _quiet_stdio_streams(self.params)
def _build_server(config: McpConnectionConfig) -> MCPServer:
"""Construct (but do not connect) the SDK server for one connection.
When ``allowed_tools`` is a list the static filter means the server will not
even list tools outside it; :func:`_register_server_tools` re-applies the
same allowlist as the authoritative gate on what gets registered. When it is
``None`` no filter is applied and every listed tool is registered.
even list tools outside it, so it is the authoritative gate on what
``describe_mcp`` and ``call_mcp`` can see. When it is ``None`` no filter is
applied and every listed tool is reachable.
"""
tool_filter = (
create_static_tool_filter(allowed_tool_names=config.allowed_tools)
@ -95,7 +125,7 @@ def _build_server(config: McpConnectionConfig) -> MCPServer:
"args": config.args,
"env": config.env,
}
return MCPServerStdio(
return _QuietMCPServerStdio(
params=stdio_params,
name=config.name,
tool_filter=tool_filter,
@ -114,105 +144,14 @@ def _build_server(config: McpConnectionConfig) -> MCPServer:
)
def _build_tool(
config: McpConnectionConfig,
server: MCPServer,
mcp_tool: MCPTool,
result_transform: ResultTransform | None,
) -> FunctionTool:
"""Build one namespaced FunctionTool from a listed MCP tool.
The SDK builds the tool (so name override, input schema, approval policy,
error-as-result handling, and tool-origin metadata are unchanged). With a
``result_transform`` we route the underlying MCP call through
:func:`_install_result_transform` so the transform sees the structured result
and decides the tool's output. Without one (the stock path), we still route
the call, through :func:`_install_error_status_capture`, so an errored result
reads as failed in the TUI while the agent's content is unchanged.
"""
namespaced_name = namespaced_tool_name(config.name, mcp_tool.name)
tool = MCPUtil.to_function_tool(
mcp_tool,
server,
convert_schemas_to_strict=False,
tool_name_override=namespaced_name,
)
if result_transform is not None:
_install_result_transform(tool, server, mcp_tool.name, namespaced_name, result_transform)
else:
_install_error_status_capture(tool, server, mcp_tool.name, namespaced_name)
return tool
def _install_result_transform(
tool: FunctionTool,
server: MCPServer,
base_tool_name: str,
namespaced_name: str,
result_transform: ResultTransform,
) -> None:
"""Route a tool's MCP call through ``result_transform``, innermost.
``MCPUtil.to_function_tool`` serializes the result inside its own invoke, so
the structured result cannot be intercepted through it. Instead we call
``server.call_tool`` ourselves, hand the parsed :class:`CallToolResult` to the
transform, and return the transform's output as the tool result.
This runs INSIDE the tool's invoke. The agent factory wraps a registered
tool's ``on_invoke_tool`` with output bounding, disk spill, and tracing at
agent-build time, which is OUTSIDE this invoke, so the transform is genuinely
the innermost step: nothing sees the raw result before the transform does.
``to_function_tool`` wraps the real invoke in the SDK's failure-handling
invoker, which stores the inner coroutine on ``_invoke_tool_impl`` and calls
it inside its try/except. Swapping that inner impl keeps the SDK's
error-as-result handling and all tool metadata while inserting the transform.
If the SDK ever renames that attribute we fail loudly rather than silently
skip the transform.
"""
async def _invoke(_ctx: Any, input_json: str) -> Any:
parsed: Any = json.loads(input_json) if input_json else {}
if not isinstance(parsed, dict):
raise ModelBehaviorError(
f"Invalid JSON input for tool {namespaced_name}: expected a JSON object"
)
args = cast("dict[str, Any]", parsed)
result = await server.call_tool(base_tool_name, args)
structured_result = result.model_dump(mode="json")
return result_transform(namespaced_name, structured_result)
_replace_tool_invoke(tool, _invoke)
def _replace_tool_invoke(tool: FunctionTool, invoke: Callable[[Any, str], Any]) -> None:
"""Swap a FunctionTool's inner invoke, failing loudly if the SDK shape changed.
``to_function_tool`` wraps the real invoke in the SDK's failure-handling
invoker, which stores the inner coroutine on ``_invoke_tool_impl`` and calls
it inside its own try/except. Swapping that inner impl keeps the SDK's
error-as-result handling and every piece of tool metadata intact. It is a
plain object with the coroutine as an attribute, not a function, so we treat
it as untyped to swap it. If the SDK ever renames that attribute we raise
rather than silently leave the swap un-applied.
"""
invoker = cast("Any", tool.on_invoke_tool)
if not hasattr(invoker, "_invoke_tool_impl"):
raise RuntimeError(
"agents SDK FunctionTool invoker shape changed: cannot swap the tool "
"invoke without risking it being silently skipped."
)
invoker._invoke_tool_impl = invoke
def _mcp_result_to_tool_output(server: MCPServer, result: Any) -> Any:
"""Serialize a ``CallToolResult`` to a tool output, mirroring the agents SDK.
This reproduces the serialization in ``agents.mcp.util.MCPUtil.invoke_mcp_tool``
(structured-content JSON when the server asks for it, otherwise text/image
content blocks, unwrapping a single block). Because the stock path now routes
content blocks, unwrapping a single block). Because the dispatch tool routes
its own call, this is what makes the agent see byte-identical content to what
the SDK would have produced on its own.
the SDK would have produced building the tool itself.
"""
if getattr(server, "use_structured_content", False) and result.structuredContent:
return json.dumps(result.structuredContent)
@ -232,85 +171,88 @@ def _mcp_result_to_tool_output(server: MCPServer, result: Any) -> Any:
return outputs
def _install_error_status_capture(
tool: FunctionTool,
server: MCPServer,
base_tool_name: str,
namespaced_name: str,
) -> None:
"""Make an errored MCP result read as failed in the TUI, agent content unchanged.
The stock SDK invoke returns only the text/image tool output and drops the
``CallToolResult.isError`` flag, so the TUI cannot tell an errored MCP call
(which it renders as a green "done") from a successful one. We route the call
the same way :func:`_install_result_transform` does, read ``isError`` off the
full result, and on an error tag the returned output dict with
``success: False``.
That tag reaches the human-facing status but not the agent. The SDK stores the
raw return value on the run item's ``output`` (which the TUI reads to derive a
tool's status), but hands the agent the value re-projected through its
ToolOutput schema, which keeps only the known ``type``/``text`` fields and
drops the extra ``success`` key. So the status flips to failed while the agent
still receives exactly the same error content it does today. Non-error calls
return the stock output unchanged and keep rendering as done.
"""
async def _invoke(_ctx: Any, input_json: str) -> Any:
parsed: Any = json.loads(input_json) if input_json else {}
if not isinstance(parsed, dict):
raise ModelBehaviorError(
f"Invalid JSON input for tool {namespaced_name}: expected a JSON object"
)
args = cast("dict[str, Any]", parsed)
result = await server.call_tool(base_tool_name, args)
tool_output = _mcp_result_to_tool_output(server, result)
if getattr(result, "isError", False) and isinstance(tool_output, dict):
return {**tool_output, "success": False}
return tool_output
_replace_tool_invoke(tool, _invoke)
async def _register_server_tools(
config: McpConnectionConfig,
async def dispatch_mcp_call(
server: MCPServer,
tool_name: str,
arguments: dict[str, Any],
*,
label: str,
result_transform: ResultTransform | None = None,
) -> list[Tool]:
"""List a connected server's tools, prefix + filter them, and register them.
) -> Any:
"""Run one MCP tool call and convert its result to a tool output.
``allowed_tools`` of ``None`` registers every listed tool; a list restricts
to exactly those names.
Shared single dispatch point for the generic ``call_mcp`` tool. Calls
``server.call_tool`` with the tool's unprefixed name, then:
- with a ``result_transform`` (strix-pro's sanitizer), hands the parsed
:class:`CallToolResult` to it as ``result_transform(label, structured)`` and
returns whatever the transform returns; or
- without one, serializes the result the way the agents SDK does (see
:func:`_mcp_result_to_tool_output`) and, when the result is an MCP error,
normalizes it through :func:`_errored_tool_output` so the failure reaches
the interfaces (see that function for the representation and why it does not
corrupt the content the agent receives).
"""
result = await server.call_tool(tool_name, arguments)
if result_transform is not None:
return result_transform(label, result.model_dump(mode="json"))
tool_output = _mcp_result_to_tool_output(server, result)
if getattr(result, "isError", False):
return _errored_tool_output(tool_output)
return tool_output
def _errored_tool_output(tool_output: Any) -> dict[str, Any]:
"""Tag a serialized MCP error so the interfaces render it as failed.
Both the TUI and the run viewer decide a tool call failed by reading a
``success`` key off a top-level dict in the result (``success is False`` means
failed). :func:`_mcp_result_to_tool_output` returns a dict only for a single
content block; a structured-content result comes back as a string and a
multi-block result as a list, and on those the failure flag had nowhere to
ride, so the interfaces showed a failed call as done. This normalizes every
errored result to a top-level dict carrying ``success: False``:
- a single content block (already a dict) keeps its ``type``/``text`` and gains
``success: False`` alongside. The SDK's ToolOutput projection keeps the known
``type``/``text`` fields and drops ``success`` before the agent sees it, so
the agent still receives exactly the error content;
- a list (multiple blocks) or a string (structured content) is placed under a
stable ``content`` key so the flag has a top-level dict to ride on. The agent
still receives the full error content, under ``content``, rather than losing
it.
"""
if isinstance(tool_output, dict):
return {**tool_output, "success": False}
return {"success": False, "content": tool_output}
async def _count_server_tools(config: McpConnectionConfig, server: MCPServer) -> int:
"""Count a connected server's reachable tools for the startup summary.
``allowed_tools`` of ``None`` counts every listed tool; a list counts only
those names. The count matches what ``describe_mcp`` will show, because the
static tool filter built in :func:`_build_server` restricts the server's own
``list_tools`` to the same allowlist.
"""
allowed = config.allowed_tools
mcp_tools = await server.list_tools()
tools: list[Tool] = [
_build_tool(config, server, mcp_tool, result_transform)
for mcp_tool in mcp_tools
if allowed is None or mcp_tool.name in allowed
]
register_agent_tools(*tools)
return tools
return sum(1 for mcp_tool in mcp_tools if allowed is None or mcp_tool.name in allowed)
async def connect_mcp_servers(
configs: list[McpConnectionConfig],
result_transform: ResultTransform | None = None,
) -> list[ConnectedMcpServer]:
"""Connect to each MCP server and register its tools.
When ``result_transform`` is given, every registered tool routes its result
through it before the result reaches the agent (see
:func:`_install_result_transform`). When it is ``None`` the tools behave
exactly as the SDK builds them.
"""Connect to each MCP server and return its live session.
Returns one :class:`ConnectedMcpServer` per server that connected, carrying
the SDK server (so the caller can clean it up when the run ends) plus the
server name and how many tools it registered (so the caller can show the
user a startup summary). Connections that fail are skipped rather than
raised.
the SDK server (so the caller can clean it up when the run ends and hand it to
the run's registry) plus the server name, how many tools it offers, and the
connection's notes. Connections that fail are skipped rather than raised.
Nothing is registered as an agent tool: the caller builds a per-run
:class:`~strix.tools.mcp.registry.McpRegistry` from these sessions, and the
agent reaches each tool on demand through ``describe_mcp`` / ``call_mcp``.
"""
connected: list[ConnectedMcpServer] = []
for config in configs:
@ -318,7 +260,7 @@ async def connect_mcp_servers(
try:
server = _build_server(config)
await server.connect() # type: ignore[no-untyped-call]
tools = await _register_server_tools(config, server, result_transform)
tool_count = await _count_server_tools(config, server)
except Exception:
logger.exception("Skipping MCP connection %r", config.name)
if server is not None:
@ -339,11 +281,47 @@ async def connect_mcp_servers(
await established.server.cleanup() # type: ignore[no-untyped-call]
raise
logger.info("Connected MCP server %r (%d tools)", config.name, len(tools))
logger.info("Connected MCP server %r (%d tools)", config.name, tool_count)
connected.append(
ConnectedMcpServer(
server=server, name=config.name, tool_count=len(tools), notes=config.notes
server=server, name=config.name, tool_count=tool_count, notes=config.notes
)
)
return connected
async def attach_mcp_requests(
requests: list[McpConnectionRequest],
registry: McpRegistry,
) -> list[ConnectedMcpServer]:
"""Connect a caller's MCP requests and populate the run's registry.
The one shared attach-and-populate path both the command-line and the
SaaS/pro product go through, so all connecting and cleanup lives in one owner.
The caller supplies inert :class:`McpConnectionRequest` objects (a config plus
a provider label, an optional per-connection ``result_transform``, and an
optional ``purpose``) and never a live session: the engine connects each
config here, reusing :func:`connect_mcp_servers` so the fail-open behavior (a
connection that will not connect is logged and skipped) and the cancellation
cleanup are preserved unchanged.
For each connection that came up, this registers it under its config name with
its tool count, its ``provider`` label, its ``result_transform``, and a purpose
of ``request.purpose`` when set else the connection's notes. Returns the
connected servers (the runner records them and cleans them up when the run
ends).
"""
request_by_name = {request.config.name: request for request in requests}
connections = await connect_mcp_servers([request.config for request in requests])
for connection in connections:
request = request_by_name[connection.name]
registry.add(
name=connection.name,
server=connection.server,
tool_count=connection.tool_count,
purpose=request.purpose or connection.notes,
provider=request.provider,
result_transform=request.result_transform,
)
return connections

6
strix/tools/mcp/config.py
View file

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

13
strix/tools/mcp/loader.py
View file

@ -1,9 +1,10 @@
"""Read the open-source user's MCP servers from ``~/.strix/mcp-servers.json``.
An open-source user lists the MCP servers they want the agent to reach in a
small JSON file. Strix reads it at the start of a run, connects to each server,
and registers its tools. The file is optional; without it the run simply gets
no MCP tools.
small JSON file. Strix reads it at the start of a run and connects to each
server, holding the live sessions in the run's registry for the agent to reach
on demand. The file is optional; without it the run simply gets no MCP
connections.
Parsing is fail-open. A single malformed entry is logged and skipped rather than
raising, so one bad row never blocks the servers that are valid, and a missing
@ -46,9 +47,9 @@ def _resolve_path(path: Path | None) -> Path:
def _dedupe_by_name(configs: list[McpConnectionConfig]) -> list[McpConnectionConfig]:
"""Keep the first connection of each name, dropping later duplicates.
Names namespace a server's tools (``<name>.<tool>``), so two connections
sharing a name would collide and the second's tools would be silently
rejected at registration. Drop the duplicate here, with a warning, instead.
A connection's name is its key in the run's registry, so two connections
sharing a name would collide and the second would overwrite the first. Drop
the duplicate here, with a warning, instead.
"""
seen: set[str] = set()
unique: list[McpConnectionConfig] = []

59
strix/tools/mcp/naming.py
View file

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

216
strix/tools/mcp/registry.py Normal file
View file

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

313
strix/tools/threat_model/tools.py
View file

@ -1,28 +1,29 @@
"""Target-scoped threat models — cached under ``~/.strix/threat-models``.
"""Run-scoped threat models — mirrored to ``{state_dir}/threat_models.json``.
A threat model describes the target, not the scan: a host, an application, an
API, a repository, or whatever else the engagement is pointed at. It stays
valid across unrelated runs against the same target, so it is keyed by target
identity rather than by run id — one agent derives it, every later agent in
this run and in future runs against the same target reads it back instead of
A threat model is the scan's shared answer to who the attacker is, where the
trust boundaries sit, and what counts as critical for the target. One agent
derives it and every other agent on the same run reads it back instead of
re-deriving trust boundaries from scratch.
Where the target is a checkout, the model is additionally pinned to the git
revision, so a moved ``HEAD`` marks it stale. Black-box targets have no
revision to pin to; those age out instead.
It does not outlive the scan. The mirror lives in the run's own state directory
and exists only so a resumed scan keeps the baseline its earlier agents agreed
on; a new scan against the same host or checkout starts with no model and
derives its own. Agents do spell one target several ways within a run — the URL
they were handed, the page they happen to be testing, a checkout path — so a
model is keyed by a normalized target identity to keep them converging on one
document instead of each starting a fresh one.
"""
from __future__ import annotations
import asyncio
import hashlib
import json
import logging
import re
import subprocess
import tempfile
import threading
from datetime import UTC, datetime, timedelta
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit
@ -35,16 +36,19 @@ from strix.core.agents import AgentCoordinator
logger = logging.getLogger(__name__)
_CACHE_DIR = Path.home() / ".strix" / "threat-models"
_MAX_MODEL_BYTES = 512 * 1024
_MIN_MODEL_CHARS = 400
_MIN_AMENDMENT_CHARS = 80
_MAX_AMENDMENTS = 40
_GIT_TIMEOUT_SECONDS = 10
_UNVERSIONED = "unversioned"
_MAX_AGE_DAYS = 14
_DEFAULT_PORTS = {"http": "80", "https": "443"}
_cache_lock = threading.RLock()
_store_lock = threading.RLock()
# The whole store: target identity -> model. It holds exactly the models this
# scan derived, and is mirrored to the run's state directory for resume.
_MODELS: dict[str, dict[str, Any]] = {}
_store_path: Path | None = None
_REQUIRED_SECTIONS = (
"overview",
@ -95,7 +99,7 @@ def _remote_authority(target: str) -> str:
def _normalize_remote_target(target: str) -> str:
"""Collapse the spellings of one remote target onto a single cache key."""
"""Collapse the spellings of one remote target onto a single key."""
authority = _remote_authority(target)
if not authority:
return re.sub(r"\s+", " ", target.lower()).strip()
@ -132,30 +136,23 @@ def _normalize_git_remote(remote: str) -> str:
return normalized.removesuffix(".git")
def _target_identity(target: str) -> tuple[str, str]:
"""Return the (stable identity, revision) pair a cached model is keyed on.
def _target_identity(target: str) -> str:
"""Return the stable identity a model is stored under.
A checkout is keyed on its remote (so the same repository cloned to two
paths shares one model, and a subdirectory resolves to the whole tree) and
pinned to ``HEAD``. Everything else — a host, a URL, an API base, a named
scope — is keyed on its normalized form and carries no revision. Both
routes run through the same normalization, so a checkout and the URL it
was cloned from land on one key.
A checkout is keyed on its remote, so the same repository checked out at
two paths shares one model and a subdirectory resolves to the whole tree.
Everything else — a host, a URL, an API base, a named scope — is keyed on
its normalized form. Both routes run through the same normalization, so a
checkout and the URL it was cloned from land on one key.
"""
directory = _local_directory(target)
if directory is None:
return _normalize_remote_target(target).removesuffix(".git"), _UNVERSIONED
return _normalize_remote_target(target).removesuffix(".git")
remote = _git(directory, ["config", "--get", "remote.origin.url"])
revision = _git(directory, ["rev-parse", "HEAD"]) or _UNVERSIONED
if remote:
return _normalize_git_remote(remote), revision
return _normalize_git_remote(remote)
toplevel = _git(directory, ["rev-parse", "--show-toplevel"])
return toplevel or str(directory), revision
def _cache_path(identity: str) -> Path:
digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:16]
return _CACHE_DIR / f"{digest}.json"
return toplevel or str(directory)
def _snap_to_scan_target(raw: str, scan_targets: list[str]) -> str:
@ -163,13 +160,13 @@ def _snap_to_scan_target(raw: str, scan_targets: list[str]) -> str:
Agents name the same target differently — one passes the URL it was given,
the next the page it happens to be testing, a third the checkout path. Left
alone those become separate cache keys, every lookup misses, and each agent
alone those become separate keys, every lookup misses, and each agent
quietly derives its own model, which is the exact failure the shared model
exists to prevent. So a target that is recognisably one of the scan's own
targets is resolved to that target instead.
"""
identity, _ = _target_identity(raw)
scoped = [(target, _target_identity(target)[0]) for target in scan_targets]
identity = _target_identity(raw)
scoped = [(target, _target_identity(target)) for target in scan_targets]
if any(known == identity for _, known in scoped):
return raw
@ -208,38 +205,51 @@ def _resolve_target(
return (_snap_to_scan_target(raw, known) if known else raw), None
def _is_expired(created_at: str | None) -> bool:
if not created_at:
return True
try:
created = datetime.fromisoformat(created_at)
except ValueError:
return True
if created.tzinfo is None:
created = created.replace(tzinfo=UTC)
return datetime.now(UTC) - created > timedelta(days=_MAX_AGE_DAYS)
def _missing_sections(content: str) -> list[str]:
lowered = content.lower()
return [section for section in _REQUIRED_SECTIONS if section not in lowered]
def _read_cache(path: Path) -> dict[str, Any] | None:
"""Load a cached model. Callers must already hold ``_cache_lock``."""
if not path.is_file():
return None
try:
cached = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
logger.exception("threat model cache at %s is unreadable", path)
return None
return cached if isinstance(cached, dict) else None
def _write_cache(path: Path, payload: dict[str, Any]) -> str | None:
"""Atomically persist a model. Callers must already hold ``_cache_lock``."""
def hydrate_threat_models_from_disk(state_dir: Path) -> None:
"""Point the store at this run's mirror and load whatever it already holds.
A resumed scan is the same scan, so its agents have to keep the baseline
the earlier ones agreed on. The mirror lives under the run directory, so a
different scan never reads it.
"""
global _store_path # noqa: PLW0603
_store_path = state_dir / "threat_models.json"
with _store_lock:
_MODELS.clear()
if not _store_path.is_file():
return
try:
data = json.loads(_store_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
logger.exception(
"threat_models.json at %s is unreadable; starting with no models",
_store_path,
)
return
if not isinstance(data, dict):
return
_MODELS.update(
{
identity: model
for identity, model in data.items()
if isinstance(identity, str) and isinstance(model, dict)
}
)
logger.info("threat models hydrated from %s (%d)", _store_path, len(_MODELS))
def _persist_locked() -> None:
"""Mirror the store to disk. Callers must already hold ``_store_lock``.
Serializing and renaming in one critical section keeps a writer holding an
older serialization from winning the rename and dropping a concurrent
agent's model or amendment.
"""
path = _store_path
if path is None:
return
try:
payload = json.dumps(_MODELS, ensure_ascii=False, default=str)
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
mode="w",
@ -249,85 +259,61 @@ def _write_cache(path: Path, payload: dict[str, Any]) -> str | None:
suffix=".tmp",
delete=False,
) as tmp:
tmp.write(json.dumps(payload, ensure_ascii=False))
tmp.write(payload)
tmp_path = Path(tmp.name)
tmp_path.replace(path)
except OSError as exc:
logger.exception("threat model persist to %s failed", path)
return f"Failed to persist threat model: {exc}"
return None
except OSError:
logger.exception("threat model mirror to %s failed", path)
def _amendments_of(cached: dict[str, Any]) -> list[dict[str, Any]]:
raw = cached.get("amendments")
def _missing_sections(content: str) -> list[str]:
lowered = content.lower()
return [section for section in _REQUIRED_SECTIONS if section not in lowered]
def _amendments_of(model: dict[str, Any]) -> list[dict[str, Any]]:
raw = model.get("amendments")
if not isinstance(raw, list):
return []
return [item for item in raw if isinstance(item, dict)]
def _not_found(identity: str, revision: str) -> dict[str, Any]:
def _not_found(identity: str) -> dict[str, Any]:
return {
"success": True,
"found": False,
"target": identity,
"revision": revision,
"message": (
"No threat model cached for this target. Derive one — from the code if "
"you have it, from recon output if you do not — and persist it with "
"save_threat_model, so every agent on this scan shares one view of the "
"trust boundaries instead of each inventing their own."
"No threat model for this target on this scan. Nothing carries over "
"from other scans, so derive one — from the code if you have it, from "
"recon output if you do not — and share it with save_threat_model, so "
"every agent on this scan works from one view of the trust boundaries "
"instead of each inventing their own."
),
}
def _staleness(cached: dict[str, Any], revision: str) -> tuple[bool, str | None]:
"""Decide whether a cached model can still be trusted, and why not."""
if revision != _UNVERSIONED:
if cached.get("revision") == revision:
return False, None
return True, (
"This model was derived against a different revision. Use it as a "
"starting point, re-check the boundaries it names against the current "
"tree, and save the corrected version."
)
created_at = cached.get("created_at")
if not _is_expired(created_at if isinstance(created_at, str) else None):
return False, None
return True, (
f"This model is more than {_MAX_AGE_DAYS} days old and there is no revision "
"to pin it to, so the target may have moved under it. Treat its surface "
"inventory as a lead list to re-confirm during recon, not as fact, and save "
"the corrected version."
)
def _get_impl(target: str, scan_targets: list[str] | None = None) -> dict[str, Any]:
resolved, error = _resolve_target(target, scan_targets)
if resolved is None:
return {"success": False, "error": error}
identity, revision = _target_identity(resolved)
path = _cache_path(identity)
with _cache_lock:
cached = _read_cache(path)
if cached is None:
return _not_found(identity, revision)
content = cached.get("content")
identity = _target_identity(resolved)
with _store_lock:
model = _MODELS.get(identity)
if model is None:
return _not_found(identity)
content = model.get("content")
amendments = list(_amendments_of(model))
if not isinstance(content, str) or not content.strip():
return _not_found(identity, revision)
return _not_found(identity)
stale, stale_message = _staleness(cached, revision)
result: dict[str, Any] = {
"success": True,
"found": True,
"target": identity,
"revision": revision,
"cached_revision": cached.get("revision"),
"created_at": cached.get("created_at"),
"stale": stale,
"content": content,
}
amendments = _amendments_of(cached)
if amendments:
result["amendments"] = amendments
result["amendments_note"] = (
@ -335,8 +321,6 @@ def _get_impl(target: str, scan_targets: list[str] | None = None) -> dict[str, A
"correct or extend it and have not been folded in yet - read them as "
"part of the model, and prefer the later one where they conflict."
)
if stale_message:
result["message"] = stale_message
return result
@ -376,25 +360,21 @@ def _save_impl(
),
}
identity, revision = _target_identity(resolved)
path = _cache_path(identity)
payload: dict[str, Any] = {
"target": identity,
"revision": revision,
"created_at": datetime.now(UTC).isoformat(),
"created_by": agent_name,
"content": body,
}
with _cache_lock:
existing = _read_cache(path)
identity = _target_identity(resolved)
with _store_lock:
existing = _MODELS.get(identity)
folded = len(_amendments_of(existing)) if existing else 0
error = _write_cache(path, payload)
if error:
return {"success": False, "error": error}
_MODELS[identity] = {
"target": identity,
"written_at": datetime.now(UTC).isoformat(),
"written_by": agent_name,
"content": body,
}
_persist_locked()
message = (
"Threat model saved. Subagents should call get_threat_model before they "
"start, and treat its trust boundaries as the shared baseline."
"Threat model shared with this scan. Subagents should call get_threat_model "
"before they start, and treat its trust boundaries as the shared baseline."
)
if folded:
message += (
@ -404,34 +384,35 @@ def _save_impl(
return {
"success": True,
"target": identity,
"revision": revision,
"amendments_cleared": folded,
"message": message,
}
def _append_amendment(
path: Path, amendment: dict[str, Any]
identity: str, amendment: dict[str, Any]
) -> tuple[list[dict[str, Any]] | None, str | None]:
"""Add an amendment to the cached model. Returns (amendments, error)."""
with _cache_lock:
cached = _read_cache(path)
if cached is None or not str(cached.get("content", "")).strip():
"""Add an amendment to the stored model. Returns (amendments, error)."""
with _store_lock:
model = _MODELS.get(identity)
if model is None or not str(model.get("content", "")).strip():
return None, (
"No threat model exists for this target yet, so there is nothing to "
"amend. Derive the base model and call save_threat_model instead."
)
amendments = _amendments_of(cached)
amendments = _amendments_of(model)
if len(amendments) >= _MAX_AMENDMENTS:
return None, (
f"This model already carries {len(amendments)} amendments. Fold them "
"into the base model with save_threat_model before adding more."
)
amendments.append(amendment)
cached["amendments"] = amendments
if len(json.dumps(cached, ensure_ascii=False).encode("utf-8")) > _MAX_MODEL_BYTES:
candidate = [*amendments, amendment]
sized = {**model, "amendments": candidate}
if len(json.dumps(sized, ensure_ascii=False).encode("utf-8")) > _MAX_MODEL_BYTES:
return None, "Threat model with this amendment exceeds 512KB; tighten it."
return amendments, _write_cache(path, cached)
model["amendments"] = candidate
_persist_locked()
return candidate, None
def _amend_impl(
@ -455,13 +436,12 @@ def _amend_impl(
),
}
identity, revision = _target_identity(resolved)
identity = _target_identity(resolved)
amendments, amend_error = _append_amendment(
_cache_path(identity),
identity,
{
"at": datetime.now(UTC).isoformat(),
"by": agent_name,
"revision": revision,
"content": body,
},
)
@ -471,7 +451,6 @@ def _amend_impl(
return {
"success": True,
"target": identity,
"revision": revision,
"amendment_count": len(amendments),
"message": (
"Amendment recorded. Agents calling get_threat_model will now see it "
@ -500,25 +479,25 @@ def _scan_targets(ctx: RunContextWrapper) -> list[str]:
@function_tool(timeout=30)
async def get_threat_model(ctx: RunContextWrapper, target: str) -> str:
"""Read the cached threat model for a target, if one exists.
"""Read this scan's threat model for a target, if an agent has derived one.
A threat model belongs to the target, not to this scan — the same
trust boundaries hold across unrelated runs against the same host
or application. Call this before you start hunting so you inherit
the shared view instead of re-deriving it, and so every agent on
this run agrees on what "attacker-controlled" means here.
The threat model is this run's shared answer to who the attacker
is, where the trust boundaries sit, and what counts as critical
here. Call it before you start hunting so you inherit the shared
view instead of re-deriving it, and so every agent on this run
agrees on what "attacker-controlled" means.
It is scoped to this scan and nothing is carried over from an
earlier run, so an empty result means no agent has derived one yet.
Works black-box or white-box. The target can be a host, a URL, an
API base, or a repository path; equivalent spellings of the same
host resolve to the same model, and a checkout resolves to its
remote, so a model derived white-box is read back by a black-box
agent testing the deployment.
remote, so a model derived white-box by one agent is read back by
another testing the deployment.
Returns ``found: false`` when nothing is cached — derive one and
persist it with ``save_threat_model``. ``stale: true`` means the
checkout moved to a different revision, or that a model with no
revision to pin to has aged out: use it as a starting point,
re-confirm what it claims, and save the corrected version.
Returns ``found: false`` when nothing has been derived yet — derive
one and share it with ``save_threat_model``.
Any ``amendments`` in the response are corrections other agents
recorded after the base model was written. They are part of the
@ -540,10 +519,10 @@ async def get_threat_model(ctx: RunContextWrapper, target: str) -> str:
@function_tool(timeout=30)
async def save_threat_model(ctx: RunContextWrapper, target: str, content: str) -> str:
"""Persist a target-scoped threat model for reuse by other agents.
"""Share a target-scoped threat model with the other agents on this scan.
Keyed by target identity, so a later scan of the same host or tree
reads it back instead of paying to derive it again.
The model lives for this run only — it is not written to disk and a
later scan of the same host or tree starts without it.
**This replaces the whole document, and clears any amendments** —
it is for the agent establishing the baseline (normally root,
@ -561,9 +540,9 @@ async def save_threat_model(ctx: RunContextWrapper, target: str, content: str) -
necessarily provisional — say which parts are inferred rather than
observed, and let later agents amend it as the picture fills in.
**Scope it to the target, not to this scan.** Do not centre it on
the diff you were handed, the subsystem you were assigned, or the
one host that happened to answer first. With source, distinguish
**Scope it to the target, not to your slice of it.** Do not centre
it on the diff you were handed, the subsystem you were assigned, or
the one host that happened to answer first. With source, distinguish
real product and runtime surfaces from test, docs, example, and
developer-tooling paths — in a monorepo, do not let ``tests/`` or
one-off scripts become the centre of gravity unless the code shows

26
tests/conftest.py Normal file
View file

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

1083
tests/test_mcp_client.py
View file

File diff suppressed because it is too large Load diff

142
tests/test_runner_mcp.py Normal file
View file

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

68
tests/test_runner_root_prompt.py
View file

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

80
tests/test_threat_model_tool.py
View file

@ -1,10 +1,8 @@
"""Tests for the target-scoped threat model cache."""
"""Tests for the run-scoped threat model store."""
from __future__ import annotations
import json
import subprocess
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING
import pytest
@ -17,6 +15,7 @@ from strix.tools.threat_model.tools import (
_save_impl,
amend_threat_model,
get_threat_model,
hydrate_threat_models_from_disk,
save_threat_model,
)
@ -64,8 +63,10 @@ def _make_repo(tmp_path: Path, name: str = "repo") -> Path:
@pytest.fixture(autouse=True)
def _isolated_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(threat_model_tools, "_CACHE_DIR", tmp_path / "cache")
def _empty_store() -> None:
"""Each test is its own run, so it starts with an empty, unmirrored store."""
threat_model_tools._MODELS.clear()
threat_model_tools._store_path = None
def test_missing_model_reports_not_found(tmp_path: Path) -> None:
@ -85,11 +86,53 @@ def test_saved_model_round_trips(tmp_path: Path) -> None:
result = _get_impl(str(repo))
assert result["found"] is True
assert result["stale"] is False
assert "multi-tenant billing API" in result["content"]
def test_model_is_stale_after_new_revision(tmp_path: Path) -> None:
def test_nothing_is_written_outside_the_run(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The model must not outlive the scan, so nothing may land in the home dir."""
home = tmp_path / "home"
home.mkdir()
monkeypatch.setenv("HOME", str(home))
repo = _make_repo(tmp_path)
_save_impl(str(repo), _MODEL, "root")
_amend_impl(str(repo), _ADDENDUM, "agent-a")
assert list(home.rglob("*")) == []
def test_a_new_run_starts_without_the_model(tmp_path: Path) -> None:
"""A later scan of the same target inherits nothing from this one."""
repo = _make_repo(tmp_path)
hydrate_threat_models_from_disk(tmp_path / "first-run")
_save_impl(str(repo), _MODEL, "root")
hydrate_threat_models_from_disk(tmp_path / "second-run") # a different scan
assert _get_impl(str(repo))["found"] is False
def test_resuming_the_same_run_keeps_the_model(tmp_path: Path) -> None:
"""A resumed scan is the same scan, so its agents keep the shared baseline."""
state_dir = tmp_path / "state"
repo = _make_repo(tmp_path)
hydrate_threat_models_from_disk(state_dir)
_save_impl(str(repo), _MODEL, "root")
_amend_impl(str(repo), _ADDENDUM, "agent-a")
threat_model_tools._MODELS.clear() # what the resuming process starts from
hydrate_threat_models_from_disk(state_dir)
result = _get_impl(str(repo))
assert result["found"] is True
assert [a["content"] for a in result["amendments"]] == [_ADDENDUM]
def test_model_survives_a_new_revision_within_the_run(tmp_path: Path) -> None:
"""The model is not pinned to a revision; a commit mid-run does not drop it."""
repo = _make_repo(tmp_path)
_save_impl(str(repo), _MODEL, None)
@ -100,11 +143,10 @@ def test_model_is_stale_after_new_revision(tmp_path: Path) -> None:
result = _get_impl(str(repo))
assert result["found"] is True
assert result["stale"] is True
assert result["content"]
assert "multi-tenant billing API" in result["content"]
def test_cache_is_keyed_per_repository(tmp_path: Path) -> None:
def test_store_is_keyed_per_repository(tmp_path: Path) -> None:
first = _make_repo(tmp_path, "first")
second = _make_repo(tmp_path, "second")
_save_impl(str(first), _MODEL, None)
@ -218,8 +260,6 @@ def test_blackbox_target_round_trips() -> None:
result = _get_impl(target)
assert result["found"] is True
assert result["stale"] is False, "a fresh model with no revision is not stale"
assert result["revision"] == "unversioned"
assert "Inferred from recon" in result["content"]
@ -232,22 +272,6 @@ def test_blackbox_target_spellings_share_one_model() -> None:
assert _get_impl("https://other.example.com")["found"] is False
def test_blackbox_model_goes_stale_with_age() -> None:
target = "https://app.example.com"
_save_impl(target, _BLACKBOX_MODEL, "recon")
aged = (datetime.now(UTC) - timedelta(days=threat_model_tools._MAX_AGE_DAYS + 1)).isoformat()
path = threat_model_tools._cache_path("app.example.com:443")
payload = json.loads(path.read_text(encoding="utf-8"))
payload["created_at"] = aged
path.write_text(json.dumps(payload), encoding="utf-8")
result = _get_impl(target)
assert result["stale"] is True
assert "re-confirm" in result["message"]
def test_blackbox_target_can_be_amended() -> None:
target = "https://app.example.com"
_save_impl(target, _BLACKBOX_MODEL, "recon")