Compare commits

..
Author SHA1 Message Date
Alex Schapiro 8e31009c99 docs(skills): correct gRPC guidance, a .proto is not a spec target 2026-08-20 20:40:46 +00:00
Alex Schapiro 91afce6847 docs(skills): document --workspace-file for supporting files 2026-08-20 20:33:32 +00:00
Alex Schapiro 991c017ddf docs(skills): fix nonexistent --mount flag, document real targeting flags, add application-security-testing skill
- Remove --mount from two skills: the flag does not exist in the CLI. Local
  paths are mounted writable when passed with -t.
- Document --target-list, --scope-mode, --diff-base, and OpenAPI/Postman
  targets, so agents stop putting spec URLs in --instruction prose.
- Add the application-security-testing skill as the entry point for
  whole-product AppSec requests, routing each asset to the right workflow.
- Drop contractions and Latin abbreviations across the skill prose.
2026-08-20 20:32:34 +00:00
Alex Schapiro 4b6b2b7920 docs(skills): use current OWASP editions (Top 10:2025, API Top 10 2023) 2026-08-20 20:32:34 +00:00
Alex Schapiro 93e2c3d7e9 fix(skills): avoid unquoted colon in api-security-testing description 2026-08-20 20:32:34 +00:00
Alex Schapiro 614d2bfb22 feat(skills): add target-specific security testing skills (web app, API, OWASP Top 10, code review) 2026-08-20 20:32:34 +00:00
Alex Schapiro d6f2218756 Drop strict tool schemas on Claude routes 2026-08-20 23:12:23 +03:00
29 changed files with 469 additions and 1626 deletions
+8
View File
@@ -15,6 +15,14 @@ npx skills add usestrix/strix
- `fix-security-vulnerabilities-with-strix` — remediate findings and re-run Strix to verify
- `ci-security-scanning-with-strix` — add PR scanning to CI/CD (self-hosted CLI or managed app)
Target-specific workflows built on the same engine:
- `application-security-testing` — whole-product AppSec review: pick the right test per asset, then rank the results
- `web-app-penetration-testing` — black-box pentest of a live web app or staging site
- `api-security-testing` — REST/GraphQL APIs and the OWASP API Security Top 10 (BOLA/IDOR, authz)
- `owasp-top-10-testing` — systematic OWASP Top 10 assessment with honest per-category coverage
- `find-security-vulnerabilities-in-code` — white-box review of a repo or working tree
**Two ways to run, same engine — pick per situation:**
- **Open-source CLI (self-hosted):** free, fully local, BYO LLM key, needs Docker. Best for local dev loops, air-gapped/offline, and full control.
+1 -25
View File
@@ -116,7 +116,7 @@ Strix is agent-ready. Give Claude Code, Cursor, Codex, or any [SKILL.md-compatib
npx skills add usestrix/strix
```
This installs four skills: **penetration-testing-with-strix** (run headless scans and read results), **managed-pentesting-with-strix** (drive the managed [app.strix.ai](https://app.strix.ai) platform via REST — no local Docker or LLM key), **fix-security-vulnerabilities-with-strix** (remediate + re-scan to verify), and **ci-security-scanning-with-strix** (PR scanning in CI). Agents can run Strix two ways with the same engine — the open-source CLI locally, or the managed cloud when there's no local infra — and read [`AGENTS.md`](AGENTS.md) for a quick reference, [docs.strix.ai/llms.txt](https://docs.strix.ai/llms.txt) for the CLI docs, and [docs.app.strix.ai](https://docs.app.strix.ai) for the API.
This installs nine skills: **penetration-testing-with-strix** (run headless scans and read results), **managed-pentesting-with-strix** (drive the managed [app.strix.ai](https://app.strix.ai) platform via REST — no local Docker or LLM key), **fix-security-vulnerabilities-with-strix** (remediate + re-scan to verify), **ci-security-scanning-with-strix** (PR scanning in CI), plus target-specific workflows: **application-security-testing**, **web-app-penetration-testing**, **api-security-testing**, **owasp-top-10-testing**, and **find-security-vulnerabilities-in-code**. Agents can run Strix two ways with the same engine — the open-source CLI locally, or the managed cloud when there's no local infra — and read [`AGENTS.md`](AGENTS.md) for a quick reference, [docs.strix.ai/llms.txt](https://docs.strix.ai/llms.txt) for the CLI docs, and [docs.app.strix.ai](https://docs.app.strix.ai) for the API.
---
@@ -320,30 +320,6 @@ strix auth status # show the active sign-in
strix auth logout # forget the sign-in
```
#### Connect your own MCP servers
Strix can connect to Model Context Protocol (MCP) servers you list and expose their tools to the agent during a run. Create `~/.strix/mcp-servers.json` with a JSON list of servers. Each entry is either a local `stdio` server that Strix launches as a subprocess, or a remote `http` server:
```json
[
{
"name": "local_fs",
"transport": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"]
},
{
"name": "github",
"transport": "http",
"url": "https://api.githubcopilot.com/mcp/",
"auth": { "kind": "bearer", "token": "your-token" },
"allowed_tools": ["list_issues"]
}
]
```
Each server's tools are namespaced by `name` (for example `local_fs.read_file`). Omit `allowed_tools` to expose every tool the server offers, or set it to a list to restrict which tools the agent can call. The file is optional, and a server that fails to connect is skipped without failing the run. You can point Strix at a different file with `STRIX_MCP_CONFIG`.
**Recommended models for best results:**
- [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4`
+1 -2
View File
@@ -47,8 +47,7 @@
"pages": [
"integrations/github-actions",
"integrations/ci-cd",
"integrations/coding-agents",
"integrations/mcp"
"integrations/coding-agents"
]
},
{
+5
View File
@@ -19,6 +19,11 @@ npx skills add usestrix/strix
| `managed-pentesting-with-strix` | Drive the managed [app.strix.ai](https://app.strix.ai) platform over REST — no local Docker or LLM key needed |
| `fix-security-vulnerabilities-with-strix` | Triage findings, fix root causes, and re-run Strix to verify each fix |
| `ci-security-scanning-with-strix` | Add PR security scanning to GitHub Actions or any CI (self-hosted CLI or managed app) |
| `application-security-testing` | Assess a whole product: choose the right test for each asset, then rank the findings into one remediation plan |
| `web-app-penetration-testing` | Black-box pentest of a live web app or staging site — scope, credentials, and multi-account access-control testing |
| `api-security-testing` | Test a REST/GraphQL API against the OWASP API Security Top 10 — schema-driven enumeration, BOLA/IDOR, authz |
| `owasp-top-10-testing` | Systematic OWASP Top 10 assessment with honest per-category coverage |
| `find-security-vulnerabilities-in-code` | White-box security review of a repo or working tree, with exploits to confirm findings |
Install a single skill with `npx skills add usestrix/strix --skill penetration-testing-with-strix`, or use one without installing:
-113
View File
@@ -1,113 +0,0 @@
---
title: "MCP Servers"
description: "Connect your own MCP servers and expose their tools to the agent"
---
Strix can connect to [Model Context Protocol (MCP)](https://modelcontextprotocol.io) servers you list and expose their tools to the agent during a run. Use this to give the agent extra capabilities — reading files, querying an issue tracker, or any other tool a server offers.
## Setup
Create the file `~/.strix/mcp-servers.json`. It holds a JSON list of the servers you want the agent to reach. Each entry is either a local `stdio` server that Strix launches as a subprocess, or a remote `http` server.
Create the directory if it does not exist, then write the file:
```bash
mkdir -p ~/.strix
```
Paste the servers you want into `~/.strix/mcp-servers.json`. The example below shows one of each transport — a local filesystem server over `stdio` and a remote GitHub server over `http` with a bearer token:
```json
[
{
"name": "local_fs",
"transport": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"]
},
{
"name": "github",
"transport": "http",
"url": "https://api.githubcopilot.com/mcp/",
"auth": { "kind": "bearer", "token": "your-token" },
"allowed_tools": ["list_issues"]
}
]
```
Strix reads this file at the start of each run. There is no default file, so no MCP tools are loaded until you create it. Edit `command`, `args`, `url`, and `token` to match your own servers.
## Fields
<ParamField path="name" type="string" required>
A short label for the connection. Each server's tools are namespaced by
`name` (for example `local_fs.read_file`), so two servers can offer the same
tool name without colliding.
</ParamField>
<ParamField path="transport" type="string">
`stdio` for a local subprocess server, or `http` for a remote server.
</ParamField>
<ParamField path="command" type="string">
For `stdio` servers: the executable Strix launches (for example `npx`).
</ParamField>
<ParamField path="args" type="array">
For `stdio` servers: the arguments passed to `command`.
</ParamField>
<ParamField path="url" type="string">
For `http` servers: the server endpoint URL.
</ParamField>
<ParamField path="auth" type="object">
For `http` servers that need a bearer token:
`{ "kind": "bearer", "token": "your-token" }`.
</ParamField>
<ParamField path="allowed_tools" type="array">
Restrict which tools the agent can call. Omit it to expose every tool the
server offers, or set it to a list of tool names to allow only those.
</ParamField>
<ParamField path="notes" type="string">
Free-text notes for the agent about what this connection is and how you want
it used, for example "Staging analytics database, read-only, prefer aggregate
queries." When set, the notes are given to the agent at the start of the run
as a description of the connection.
</ParamField>
## Choosing connections per run
By default every connection in the file is used on each run. To narrow it for a
single run without editing the file, use either flag (both repeatable):
```bash
strix --mcp-server github -t ... # use only the named connection(s)
strix --mcp-exclude staging-db -t ... # use everything except the named one(s)
```
`--mcp-server` keeps only the connections you name; `--mcp-exclude` drops the
ones you name. Connection names must be unique in the file; if two entries share
a name, the first is kept and the rest are ignored.
## Pointing at a different file
To read the config from another path instead of `~/.strix/mcp-servers.json`, either pass `--mcp-config <path>` on the command line:
```bash
strix --mcp-config ./mcp-servers.json -t ...
```
or set the `STRIX_MCP_CONFIG` environment variable to that path. The flag takes precedence when both are given.
## Startup confirmation
When servers are configured, Strix prints a one-line summary at scan startup, for example `MCP: connected 1 server (14 tools): local_fs`, so you can confirm your servers connected.
## Behavior
- The config file is optional. Without it, a run simply gets no MCP tools.
- A server that fails to connect is skipped and logged, and the run continues without it.
- A single malformed entry is skipped without blocking the valid ones.
-2
View File
@@ -241,8 +241,6 @@ ignore = [
"tests/test_stream_idle_timeout.py" = ["N802", "SLF001"]
"tests/test_unknown_tool_recovery.py" = ["N802"]
"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"]
# 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"]
+61
View File
@@ -0,0 +1,61 @@
---
name: api-security-testing
description: Security-test a REST, GraphQL, or gRPC API with Strix — autonomous agents that enumerate endpoints from an OpenAPI/GraphQL schema (or by crawling), then actually exploit the API-specific vulnerability classes in the OWASP API Security Top 10 (2023) — broken object-level authorization (BOLA/IDOR), broken object property level authorization (excessive data exposure and mass assignment), broken function-level authorization, unrestricted resource consumption, SSRF, injection, and auth/token flaws. Every finding comes with a working proof-of-concept request. Use when the user asks to pentest, security-test, audit, or find vulnerabilities in an API, endpoint, or backend service.
license: Apache-2.0
metadata:
author: usestrix
homepage: https://docs.strix.ai
---
# Security-test an API
APIs fail differently from web UIs: there is no rendered surface to crawl, the interesting bugs are authorization-shaped rather than injection-shaped, and the same endpoint behaves differently per token. This workflow targets those specifics with Strix's autonomous agents, using the current [OWASP API Security Top 10 (2023)](https://owasp.org/API-Security/editions/2023/en/0x11-t10/) as the coverage checklist. For the web-app equivalent, the current edition is the OWASP Top 10:2025 — see **owasp-top-10-testing**.
Install, LLM setup, full CLI flags, and the managed-cloud path are in the **penetration-testing-with-strix** skill. Read it if `strix --version` fails or the target is not an API.
## 1. Gather what the agents need
APIs are near-impossible to test blind, so collect first:
| Input | Why it matters |
|---|---|
| **Schema** — OpenAPI/Swagger file, Postman collection, GraphQL endpoint (introspection), or a gRPC `.proto` | Turns guesswork into full endpoint enumeration. Biggest single win in coverage. An OpenAPI/Swagger or Postman spec (`.json`/`.yaml`/`.yml`) is a target Strix takes directly; a `.proto` is not, so pass it with `--workspace-file`. |
| **Two sets of credentials/tokens**, ideally in different tenants | BOLA/IDOR — API1:2023, still the #1 API risk — can only be *proven* by accessing tenant A's objects with tenant B's token. |
| **A low-privilege and a high-privilege token** | Required to prove broken function-level authorization (API5:2023 — a `user` calling admin-only routes). |
| **Example object IDs** | Lets agents test ID tampering immediately instead of hunting for valid identifiers. |
| **Out-of-scope routes** | Payments, mass notification, destructive admin endpoints. |
| **Rate limits / WAF** in front of the API | Avoids agents burning budget on throttled requests; mention them so testing adapts. |
Ask the user for anything missing — do not fabricate tokens or scan an API they do not own.
## 2. Run the scan
Pass the spec as a **target**, not as prose in the instruction — Strix parses OpenAPI/Swagger (`.json`/`.yaml`) and Postman collection exports directly, so the agents start from the real endpoint list:
```bash
strix -n -t ./openapi.yaml -t https://api.staging.example.com --max-budget 20 \
--instruction "Tenant A token: <tokenA> (org 1111, user id 11, order id 501).
Tenant B token: <tokenB> (org 2222, user id 22).
Admin token: <tokenAdmin>.
Focus: BOLA across orgs (API1), function-level authz on /admin/* (API5), object property level authz on PATCH /users/{id} — both mass assignment and over-exposed fields in list responses (API3), unrestricted resource consumption (API4).
Out of scope: POST /billing/*, POST /notifications/broadcast."
```
- **Postman instead of OpenAPI:** a collection export works as a target (`-t ./collection.postman_collection.json`), or pull one live with `-t postman://<collection-uuid>` (optionally `"postman://<collection-uuid>?env=<environment-uuid>"`), which needs `POSTMAN_API_KEY` in the environment.
- **Many services at once:** put one target per line in a file and pass `--target-list ./targets.txt`, repeatable and combinable with `-t`.
- **Add the backend source for depth:** `-t ./services/api -t https://api.staging.example.com`. With code access the agents can reason about authorization checks and object ownership rather than inferring them from responses.
- **gRPC:** target the endpoint and pass the definition as a workspace file, `-t https://grpc.staging.example.com --workspace-file ./service.proto`. Only `.json`, `.yaml`, and `.yml` specs are recognized as targets, so `-t ./service.proto` fails with "Path exists but is not a directory".
- **GraphQL:** point at the GraphQL endpoint and say whether introspection is enabled; call out that you want batching/aliasing abuse, depth/complexity limits, and per-field authorization tested.
- **Internal/private APIs** unreachable from your machine: use the managed platform's network connector — see **managed-pentesting-with-strix**.
- Use `--instruction-file` when the credential/context block gets long, and keep tokens out of shell history and out of committed files.
- **Supporting files** the agents should read but not test, such as an endpoint wordlist or handwritten notes about the tenancy model: pass `--workspace-file ./notes.md`. The file lands read-only in `/workspace`. Add `:DEST` to choose the path, for example `--workspace-file ./wordlist.txt:lists/wordlist.txt`.
## 3. Verify findings
`strix_runs/<run>/penetration_test_report.md` first, then `vulnerabilities/*.md` — each contains the exact request that proved the issue. Replay it (for example, with `curl`) before reporting; for authorization findings, confirm the response really contains the other tenant's data rather than an empty 200.
`findings.sarif` uploads to GitHub code scanning; `vulnerabilities.json` is the structured index for ticketing.
## 4. Fix, re-test, and keep it tested
Remediate with **fix-security-vulnerabilities-with-strix** (fix the authorization check, not the single endpoint), then re-run against the same target to prove the exploit is dead. Wire it into pull-request CI with **ci-security-scanning-with-strix** so new endpoints get tested as they ship.
@@ -0,0 +1,66 @@
---
name: application-security-testing
description: Application security testing (AppSec) across a whole product with Strix — decide which asset needs which test (source code, running web app, API, CI pipeline), run it, and turn the results into a ranked remediation plan. Autonomous agents exploit and prove each issue instead of emitting static-analysis alerts, so the plan is ordered by what is actually reachable. Use when the user asks for an application security review or audit, an appsec assessment, vulnerability scanning across their stack, a security review before a launch or a customer security questionnaire, or does not yet know which kind of security test they need.
license: Apache-2.0
metadata:
author: usestrix
homepage: https://docs.strix.ai
---
# Application security testing
Entry point for "make my application secure" requests, where the target is not yet a single URL or repo. The job here is to pick the right test per asset, run it, and produce one ranked plan — not to run everything at maximum depth.
Install, LLM setup, all CLI flags, and the managed-cloud path live in the **penetration-testing-with-strix** skill. Read it first if `strix --version` fails.
Only test assets the user owns or is authorized to test. Confirm authorization before the first run, and prefer staging over production, because the agents send real exploit payloads and can change data.
## 1. Map the assets
Ask (or read from the repo) and write the answers down before scanning:
- **Source** — one repo, a monorepo, several services? Which languages/frameworks?
- **Running environments** — is there a staging deployment? A public production site? A local dev server only?
- **APIs** — REST, GraphQL, gRPC? Is there an OpenAPI/GraphQL schema?
- **Authentication** — can you get two test accounts in different tenants? Most high-impact bugs need them.
- **Constraints** — out-of-scope paths, whether production may be touched, budget and wall-clock limits.
If there is no staging environment and production is off limits, say so early. A code-only review is still valuable, but it cannot prove exploitability against a live app.
## 2. Pick the right test per asset
| Asset | Skill to use |
| --- | --- |
| Repository or working tree | **find-security-vulnerabilities-in-code** |
| Live web app or staging site | **web-app-penetration-testing** |
| REST/GraphQL/gRPC API | **api-security-testing** |
| Assessment mapped to OWASP categories | **owasp-top-10-testing** |
| Every pull request, continuously | **ci-security-scanning-with-strix** |
| No Docker, no LLM key, or a report an auditor will accept | **managed-pentesting-with-strix** |
Those skills carry the flags, credential handling, and result-reading details. Do not duplicate their instructions here.
Sequence for a first assessment:
1. Review the code. It is the cheapest run and it maps the authorization model.
2. Pentest staging with credentials, and pass the repo as a second target so the agents keep source context.
3. Add CI scanning, so later regressions are caught without another manual pass.
Run one asset at a time and read each report before starting the next. Findings from the code review make the live run sharper.
## 3. Consolidate into one plan
Findings arrive per run in `strix_runs/<run>/`. Merge them into a single list and rank by **proven impact**, not by scanner severity:
1. Validated exploits reachable without authentication.
2. Validated cross-tenant or privilege-escalation issues.
3. Validated issues needing an authenticated account.
4. Unproven observations (configuration, dependency, and hardening notes) — flag as such, and never present them as confirmed vulnerabilities.
Deduplicate: the same root cause often surfaces in both the code review and the live pentest.
## 4. Be honest about coverage
State plainly what was *not* tested — assets with no staging environment, categories a black-box run cannot reach (logging and alerting, supply-chain integrity, insecure design), and any run that hit its budget or turn cap before finishing. Check `run.json` status and cost against `--max-budget` for each run. An empty result set from a truncated scan is not a clean bill of health.
Then remediate with **fix-security-vulnerabilities-with-strix**, which re-runs Strix against each fix to prove the exploit no longer works.
@@ -12,7 +12,7 @@ metadata:
You can gate PRs two ways — pick based on the environment, or combine them:
- **Managed platform (recommended for most teams)** — connect the GitHub/GitLab/Bitbucket app once and Strix reviews every PR with **no workflow file, no runner, no Docker, and no LLM key**. Results post as PR comments and land in the team dashboard. Best when you want zero CI maintenance, central tracking, or your runners lack Docker. See "Managed platform" below and the **managed-pentesting-with-strix** skill.
- **Self-hosted OSS CLI in your runner** — run a diff-scoped scan as a pipeline step. Fully in your infra, free (BYO LLM key), no external account. Requires Docker on the runner. Best for air-gapped/self-hosted CI or when you don't want scans leaving your environment.
- **Self-hosted OSS CLI in your runner** — run a diff-scoped scan as a pipeline step. Fully in your infra, free (BYO LLM key), no external account. Requires Docker on the runner. Best for air-gapped/self-hosted CI or when you do not want scans leaving your environment.
Both fail the build on validated findings and both emit SARIF 2.1.0, so you can start with one and add the other later.
@@ -63,13 +63,13 @@ jobs:
fi
```
Then tell the user to add two repository secrets: `STRIX_LLM` (model id, e.g. `openai/gpt-5.4`) and `LLM_API_KEY` (the provider key). Do not create these values yourself.
Then tell the user to add two repository secrets: `STRIX_LLM` (model id, for example `openai/gpt-5.4`) and `LLM_API_KEY` (the provider key). Do not create these values yourself.
Notes:
- In CI/headless runs Strix automatically scopes to the PR's changed files (`--scope-mode auto`). If diff resolution fails, keep `fetch-depth: 0` or set `--diff-base` to the PR's actual base branch — use `origin/${{ github.base_ref }}` in GitHub Actions rather than a hard-coded `origin/main`, since repos use different default branches.
- Exit codes: `0` pass, `2` vulnerabilities found (fails the job), `1` setup error.
- The runner needs Docker (default GitHub-hosted Ubuntu runners have it).
- **Size the budget so the scan completes — don't let it fail open.** A `0` exit means "no validated vulnerabilities in what was analyzed"; if `--max-budget` is hit before the diff is fully covered, the scan wraps up early and can still exit `0`. The "Fail unless the scan completed" step above narrows the gap: `strix_runs/<run>/run.json` is `"stopped"` when the scan was cut off at the hard budget limit without a final report. It is not a complete guard — the agents get graduated wrap-up warnings before that limit, and a run that wraps up on a warning still calls `finish_scan` and records `"completed"` with partial coverage. So keep that step in any pipeline that gates merges **and** give the scan real headroom (compare `run.json`'s `llm_usage.cost` against `--max-budget`; if it ran right up to the cap, raise it). For a `quick` diff-scoped PR scan `--max-budget 10` is usually ample, raise it for large diffs.
- **Size the budget so the scan completes — do not let it fail open.** A `0` exit means "no validated vulnerabilities in what was analyzed"; if `--max-budget` is hit before the diff is fully covered, the scan wraps up early and can still exit `0`. The "Fail unless the scan completed" step above narrows the gap: `strix_runs/<run>/run.json` is `"stopped"` when the scan was cut off at the hard budget limit without a final report. It is not a complete guard — the agents get graduated wrap-up warnings before that limit, and a run that wraps up on a warning still calls `finish_scan` and records `"completed"` with partial coverage. So keep that step in any pipeline that gates merges **and** give the scan real headroom (compare `run.json`'s `llm_usage.cost` against `--max-budget`; if it ran right up to the cap, raise it). For a `quick` diff-scoped PR scan `--max-budget 10` is usually ample, raise it for large diffs.
### Optional: upload findings to GitHub code scanning
@@ -90,7 +90,7 @@ Any pipeline works the same way — install, set the two env vars, run headless:
```bash
curl -sSL https://strix.ai/install | bash
# Resolve the PR's base branch robustly (use your CI's base-branch variable if it
# has one, e.g. GitHub Actions: origin/${{ github.base_ref }}). Avoid piping the
# has one, for example GitHub Actions: origin/${{ github.base_ref }}). Avoid piping the
# git lookup into another command — a failed lookup would otherwise be masked.
BASE_BRANCH="${CI_MERGE_REQUEST_TARGET_BRANCH_NAME:-}" # GitLab MR target
if [ -z "$BASE_BRANCH" ]; then
@@ -98,7 +98,7 @@ if [ -z "$BASE_BRANCH" ]; then
BASE_BRANCH="${BASE_BRANCH#origin/}"
fi
DIFF_BASE="origin/${BASE_BRANCH:-main}"
# Fail loudly rather than silently narrowing scope (e.g. to HEAD~1, which on a
# Fail loudly rather than silently narrowing scope (for example, to HEAD~1, which on a
# multi-commit branch would scan only the last commit and let earlier ones pass).
if ! git rev-parse --verify --quiet "$DIFF_BASE" >/dev/null; then
echo "Cannot resolve diff base '$DIFF_BASE'. Fetch the base branch (git fetch origin <base>) or set --diff-base explicitly." >&2
@@ -0,0 +1,62 @@
---
name: find-security-vulnerabilities-in-code
description: Find security vulnerabilities in a codebase or repository with Strix — a white-box AI security review that reads your source, reasons about the actual data flow and authorization model, then exploits what it finds in a live sandbox so every reported issue has a working proof-of-concept instead of a noisy static-analysis alert. Covers injection, XSS, SSRF, broken access control and IDOR, insecure deserialization, secrets in code, unsafe dependencies, and business-logic flaws. Use when the user asks to security-scan, security-review, or audit their code, repo, or pull request for vulnerabilities.
license: Apache-2.0
metadata:
author: usestrix
homepage: https://docs.strix.ai
---
# Find security vulnerabilities in code
White-box security review with Strix: the agents read the source to build a model of routes, sinks, and authorization checks, then attempt real exploitation. Findings come with a proof-of-concept, so the output is a short list of proven issues rather than the hundreds of "potential" hits a pattern-matching scanner produces.
Install, LLM setup, all flags, and the managed-cloud path are in the **penetration-testing-with-strix** skill.
## Run it
```bash
# Local working tree
strix -n -t ./ --scan-mode standard --max-budget 15
# A GitHub repo directly
strix -n -t https://github.com/org/app --max-budget 15
# Monorepo: point at the service that matters, not the whole tree
strix -n -t ./services/checkout --max-budget 20
# Only what a branch changed (whole-repo review is wasteful on a large repo)
strix -n -t ./ --scope-mode diff --diff-base origin/main --max-budget 10
```
A local path is mounted into the sandbox **writable**, so the agents can modify it. Run against a clean checkout.
Two things sharply improve results:
1. **Add a running instance of the app.** `-t ./ -t http://host.docker.internal:3000` lets the agents confirm exploitability against live behavior instead of reasoning about it statically — this is the difference between "this looks unsafe" and a validated finding. If nothing is running, static-only findings should be described as unconfirmed.
2. **Scope the review.** Point at the risky subtree and say what matters:
```bash
strix -n -t ./services/api --max-budget 15 \
--instruction "Focus on the authorization layer in src/auth and every route under src/routes/admin. Multi-tenant app: tenant id comes from the JWT. Flag any query that filters by object id without also filtering by tenant."
```
Tenancy model, trust boundaries, and which inputs are attacker-controlled are things the agents cannot infer reliably — tell them.
## Reviewing a pull request instead of the whole repo
For diff-scoped review of a branch or PR (and blocking merges on findings), use **ci-security-scanning-with-strix** — it covers diff scoping, PR comments, and SARIF upload to GitHub code scanning. The managed platform can also review PRs directly via API (**managed-pentesting-with-strix**).
## Read the results
In `strix_runs/<run>/`: `penetration_test_report.md` (start here), `vulnerabilities/*.md` (one per finding, with PoC and remediation), `vulnerabilities.json` / `.csv`, `findings.sarif` (upload to code scanning), `run.json`.
Before reporting to the user, open each finding and check the PoC actually demonstrates impact. Report file and line alongside the exploit so the fix is obvious.
Exit `0` means nothing exploitable was proven in what was analyzed — not that the codebase is clean. Check `run.json` status and cost against `--max-budget`, and note which paths went unreviewed if the run was capped.
## Complementary tooling
This is exploit-validated review, not an exhaustive inventory. Keep a dependency scanner (SCA) and secret scanning in place for complete coverage of known-CVE dependencies and committed credentials; use this for the logic, authorization, and injection bugs those tools structurally cannot find.
## Fix and verify
Hand results to **fix-security-vulnerabilities-with-strix**: patch the root cause (the shared authorization helper, not the one route), then re-run Strix to prove the exploit no longer works.
@@ -27,7 +27,7 @@ Order work by severity: critical → high → medium → low. Every Strix findin
For each finding:
1. Reproduce it with the PoC from the finding file when feasible.
2. Fix the root cause, not the specific payload (e.g. parameterize all queries, don't blocklist one string; enforce authorization in the handler, don't hide the endpoint).
2. Fix the root cause, not the specific payload (parameterize every query instead of blocking one string, and enforce authorization in the handler instead of hiding the endpoint).
3. Prefer the framework's built-in defense (ORM parameterization, template auto-escaping, CSRF middleware, centralized authz) over ad-hoc sanitization.
4. Keep the diff minimal and apply the repo's existing patterns. Finding files often include `fix_before`/`fix_after` snippets — use them as a starting point, not verbatim.
@@ -70,7 +70,7 @@ new_id=$(curl -sS "$BASE/scans/$scan_id/rerun" "${auth[@]}" -X POST | jq -r .sca
Or, if the cloud scan came from a repo/PR, trigger a fresh PR review on the fix branch (`POST /pr-reviews/start`). The platform also retests a single finding directly: `POST /api/v1/vulnerabilities/{vulnerabilityId}/retest`.
- Also re-run the PoC manually when it is a simple request/script — fastest signal.
- Run the project's own test suite to make sure the fix doesn't break behavior.
- Run the project's own test suite to make sure the fix does not break behavior.
## 4. Report
@@ -80,7 +80,7 @@ Useful `CreateScanRequest` fields:
| `domain_ids` / `repository_ids` / `internal_targets` | targets (at least one) |
| `domain_paths` / `repository_branches` | narrow to specific paths / branches |
| `credentials` | authenticated scanning, incl. `mfa_method` (`totp`/`email_otp`/…) + `totp_secret` |
| `headers` | extra HTTP headers (e.g. API keys) for the target |
| `headers` | extra HTTP headers (API keys, for example) for the target |
| `focus` / `concerns` / `context` | steer the agents |
| `upload_ids` | attach uploaded source/docs archives for white-box context |
| `notify_on_completion` / `notification_emails` | email when done |
@@ -89,7 +89,7 @@ Response is `{ scan_id, title, status }` with `status` = `pending`.
## 3. Poll to completion
`GET /scans/{scanId}` (`scans:read`). Status flow: `pending → running → completed` (or `failed` / `cancelled`). Poll on an interval — scans take minutes to hours; don't block.
`GET /scans/{scanId}` (`scans:read`). Status flow: `pending → running → completed` (or `failed` / `cancelled`). Poll on an interval — scans take minutes to hours. Do not block.
```bash
while :; do
@@ -143,10 +143,10 @@ List/inspect via `GET /pr-reviews` and `GET /pr-reviews/{id}`. Repo-level PR-rev
## 7. Continuous testing (schedules & webhooks)
- **Schedules** (`schedules:write`, Pro plan): create recurring scans and trigger them on demand — the managed equivalent of a cron-driven CLI loop.
- **Webhooks** (`webhooks:write`): subscribe to pentest/vulnerability lifecycle events (e.g. `scan.completed`, `vulnerability.created`) to push results into Slack, ticketing, or your own pipeline instead of polling.
- **Webhooks** (`webhooks:write`): subscribe to pentest/vulnerability lifecycle events such as `scan.completed` and `vulnerability.created` to push results into Slack, ticketing, or your own pipeline instead of polling.
See the schedules and webhooks sections at [docs.app.strix.ai](https://docs.app.strix.ai) for payloads.
## Safety
Only scan assets the user's organization owns or is authorized to test. External domain scans require verification (DNS/file/meta-tag) enforced by the platform — don't try to bypass it.
Only scan assets the user's organization owns or is authorized to test. External domain scans require verification (DNS/file/meta-tag) enforced by the platform — do not try to bypass it.
+64
View File
@@ -0,0 +1,64 @@
---
name: owasp-top-10-testing
description: Test an application against the OWASP Top 10 with Strix — autonomous AI agents that attempt real exploits for each category of the current OWASP Top 10:2025 (broken access control including SSRF, security misconfiguration, software supply chain failures, cryptographic failures, injection, insecure design, authentication failures, integrity failures, logging and alerting failures, mishandling of exceptional conditions) and report only what they could actually prove, mapped back to the category with a proof-of-concept. Also covers the OWASP API Security Top 10 (2023). Use when the user asks for an OWASP Top 10 assessment, OWASP compliance testing, or a security review mapped to OWASP categories.
license: Apache-2.0
metadata:
author: usestrix
homepage: https://docs.strix.ai
---
# Test against the OWASP Top 10
The OWASP Top 10 is a taxonomy of risk categories, not a test suite — "OWASP Top 10 testing" means exercising each category against the real application and reporting what's actually exploitable. Strix's agents do the exploitation; this skill covers running it category-by-category and reporting coverage honestly.
**Use the current edition: [OWASP Top 10:2025](https://owasp.org/Top10/)** (8th installment, superseding 2021). Ask the user before targeting an older edition — some compliance checklists still reference 2021, and a report labelled with the wrong edition is misleading. Key differences from 2021: **SSRF is folded into A01**, **A03 Software Supply Chain Failures** expands the old "Vulnerable and Outdated Components", and **A10 Mishandling of Exceptional Conditions** is new; A02 Security Misconfiguration moved 5→2.
Install, LLM setup, and the managed-cloud alternative: **penetration-testing-with-strix**.
## What is and is not testable by an agent
Be straight with the user about this — claiming a clean sweep of all ten is misleading.
| Category (2025) | Coverage |
|---|---|
| A01 Broken Access Control (incl. SSRF) | **Strong** — cross-user/tenant access, privilege escalation, IDOR, and SSRF (including blind, via out-of-band callbacks) are all exploit-validated. Needs two accounts plus a privileged one to prove the authorization half. |
| A02 Security Misconfiguration | **Strong** — debug endpoints, verbose errors, permissive CORS, missing hardening, default credentials, exposed admin surfaces. |
| A03 Software Supply Chain Failures | **Partial** — version fingerprinting, and vulnerable/outdated dependency review when source is supplied. Build-system and distribution-infrastructure compromise (the broader half of this category) is out of scope for a runtime scan — pair with SCA plus build-provenance controls. |
| A04 Cryptographic Failures | **Partial** — transport config, unencrypted data in transit, secrets and tokens leaked in responses. At-rest crypto and key management need source or infra review. |
| A05 Injection | **Strong** — SQL/NoSQL/command/template injection and XSS, exploit-validated. |
| A06 Insecure Design | **Partial** — business-logic abuse (price/quantity tampering, workflow skipping, race conditions) is found where reachable; design intent still needs human review and threat modelling. |
| A07 Authentication Failures | **Strong** — auth bypass, weak session/token handling, password-reset and MFA flaws. |
| A08 Software or Data Integrity Failures | **Partial** — insecure deserialization and unsigned-update paths where reachable; CI/CD trust boundaries are not runtime-testable. |
| A09 Security Logging & Alerting Failures | **Not testable from outside** — requires reviewing the logging and alerting pipeline. State this rather than reporting it as passed. |
| A10 Mishandling of Exceptional Conditions | **Partial** — agents actively probe error handling and fail-open behavior (malformed input, forced errors, race and timeout conditions) and report what leaks or bypasses a control; exhaustive coverage of internal error paths needs source review. |
For APIs, run the same exercise against the **OWASP API Security Top 10 (2023)** — API1 BOLA, API3 Broken Object Property Level Authorization (2019's excessive data exposure + mass assignment merged), API5 broken function-level authorization — using the **api-security-testing** skill.
## Run it
Maximum category coverage comes from giving the agents both the source and a running instance, plus credentials at two privilege levels:
```bash
strix -n \
-t https://github.com/org/app \
-t https://staging.example.com \
--scan-mode deep --max-budget 30 \
--instruction "OWASP Top 10:2025 assessment. Cover every category systematically and map each finding to its 2025 category id.
Accounts: userA@example.com/<pw> (org 1), userB@example.com/<pw> (org 2), admin@example.com/<pw>.
Prioritise A01 (cross-org access, privilege escalation, SSRF), A02, A05, A07, A10.
Out of scope: /billing/*, outbound email."
```
- `--scan-mode deep` matters here: systematically walking ten categories is not a quick scan.
- Without a second account, A01 results are structurally incomplete — say so in the report rather than leaving it implied.
- Need an auditor-facing PDF? Run it through the managed platform and pull the technical report (**managed-pentesting-with-strix**).
## Report honestly
From `strix_runs/<run>/`, group `vulnerabilities/*.md` by category and state, per category: what was attempted, what was proven, and what could not be assessed (A09 always; A03/A04/A06/A08/A10 partially). Label the report with the edition used. Verify each PoC yourself before it goes in front of the user.
A `0` exit code means nothing exploitable was proven **in what was analyzed** — check `run.json` status and cost against `--max-budget`; a budget-capped run is not a completed assessment.
## Then fix and re-test
Remediate with **fix-security-vulnerabilities-with-strix** and re-run to prove each exploit is closed. For ongoing coverage as the app changes, gate pull requests using **ci-security-scanning-with-strix**.
+20 -8
View File
@@ -14,14 +14,14 @@ Strix runs autonomous AI pentesting agents that dynamically exploit a target and
- **Open-source CLI** (self-hosted) — runs on your machine in a Docker sandbox with your own LLM key. Free, fully local, BYO-LLM, air-gap capable. Docs: [docs.strix.ai](https://docs.strix.ai).
- **Cloud API** (managed) — runs on Strix's infrastructure via `https://app.strix.ai/api/v1`. No Docker, no LLM key, no local compute; adds team dashboards, scheduling, PR reviews, downloadable PDF/DOCX reports (Enterprise plan), and internal-network connectors. Docs: [docs.app.strix.ai](https://docs.app.strix.ai). Full workflow in the **managed-pentesting-with-strix** skill.
## Which one? (decide, don't default)
## Which one? (decide, do not default)
Choose honestly based on the situation — neither is "better":
| Situation | Prefer |
|---|---|
| No Docker available, or a sandboxed/hosted agent/CI environment | **Cloud** |
| User has no LLM key / doesn't want to pay per-token or manage models | **Cloud** |
| User has no LLM key / does not want to pay per-token or manage models | **Cloud** |
| Team visibility, shareable dashboard, scheduled/continuous scans, PR reviews, downloadable PDF/DOCX report (Enterprise) | **Cloud** |
| Scanning internal/private infrastructure not reachable from your machine | **Cloud** (network connector) |
| Source must never leave local infra (privacy/air-gap), or fully offline | **OSS CLI** |
@@ -30,7 +30,7 @@ Choose honestly based on the situation — neither is "better":
| CI: runner already has Docker and you want a self-contained gate | **OSS CLI** |
| CI: no Docker, or you want results tracked centrally | **Cloud** |
**Mix them:** e.g. use the OSS CLI for the fast local dev-loop while writing/fixing code, and the Cloud for the authoritative, team-visible scan + report + tracking; or gate PRs with the OSS CLI in CI while the Cloud runs scheduled deep scans and PR reviews across the org. Both emit the same SARIF 2.1.0, so findings line up across environments.
**Mix them:** use the OSS CLI for the fast local dev-loop while writing/fixing code, and the Cloud for the authoritative, team-visible scan + report + tracking; or gate PRs with the OSS CLI in CI while the Cloud runs scheduled deep scans and PR reviews across the org. Both emit the same SARIF 2.1.0, so findings line up across environments.
If unsure and the user has (or will create) an app.strix.ai account, prefer **Cloud** — it avoids all local-infra friction. If they want zero signup / full local control, use the **OSS CLI**.
@@ -70,21 +70,33 @@ strix -n -t https://github.com/org/app -t https://staging.example.com
strix -n -t https://app.example.com \
--instruction "Use credentials user@example.com:pass123. Focus on IDOR and auth bypass."
# Large monorepo: bind-mount instead of copying
strix -n --mount ./huge-monorepo
# API spec as a first-class target (OpenAPI/Swagger or a Postman collection export)
strix -n -t ./openapi.yaml -t https://api.staging.example.com
# Many targets from a file, one per line
strix -n --target-list ./targets.txt --max-budget 30
# Give the agents a file to work with (wordlist, spec, notes) without making it a target
strix -n -t https://staging.example.com --workspace-file ./wordlist.txt --max-budget 20
```
A local path passed with `-t` is mounted into the sandbox **writable** — the agents can read and modify it, so point at a clean checkout, not uncommitted work you care about.
Key flags:
| Flag | Meaning |
|---|---|
| `-t, --target` | URL, repo URL, local path, domain, or IP. Repeatable. |
| `-t, --target` | URL, repo URL, local path, domain, IP, OpenAPI/Postman spec, or `postman://<uuid>`. Repeatable. |
| `--target-list PATH` | File of targets, one per line (`#` comments allowed). Repeatable, combines with `-t`. |
| `-n, --non-interactive` | Headless, exits on completion. Required for agents. |
| `-m, --scan-mode` | `quick` (minutes) / `standard` (~30 min) / `deep` (hours, default). |
| `--instruction` / `--instruction-file` | Credentials, focus areas, scope rules. |
| `--workspace-file PATH[:DEST]` | Place a file from this machine into `/workspace` read-only before the scan, for a wordlist, a spec, or notes. Repeatable. |
| `--max-budget USD` | Hard LLM spend cap; scan wraps up cleanly at the limit. |
| `--max-turns N` | Per-agent turn cap (default 500). |
| `--resume RUN_NAME` | Resume a prior run from `strix_runs/`. |
| `--resume RUN_NAME` | Resume a prior run from `strix_runs/`, with its agent history and targets. Cannot be combined with `-t`. |
| `--scope-mode` | For code targets: `auto` (diff-scope in CI/headless), `diff` (force changed files only), `full` (whole tree). |
| `--diff-base REF` | Branch or commit that `diff` scope compares against. Defaults to the repo's default branch. |
Scans take minutes (`quick`) to hours (`deep`). Run them in the background and poll for completion rather than blocking.
@@ -130,7 +142,7 @@ curl -sS "$BASE/scans/$scan_id" -H "Authorization: Bearer $STRIX_API_TOKEN" | jq
curl -sS "$BASE/scans/$scan_id/sarif" -H "Authorization: Bearer $STRIX_API_TOKEN" -o findings.sarif
```
Ask the user to create the token (and register the target as a domain/repository asset) if they haven't. If Docker/local prerequisites aren't already satisfied, use this path instead of trying to install infra.
Ask the user to create the token (and register the target as a domain/repository asset) if they have not. If Docker/local prerequisites are not already satisfied, use this path instead of trying to install infra.
---
@@ -0,0 +1,54 @@
---
name: web-app-penetration-testing
description: Pentest a web app or website end to end — black-box testing of a live URL, staging environment, or local dev server that finds and exploits real vulnerabilities (auth bypass, broken access control, IDOR, injection, XSS, SSRF, business logic) and proves each one with a working proof-of-concept instead of a signature match. Runs with Strix, either the self-hosted open-source CLI or the managed app.strix.ai cloud. Use when the user asks to pentest, hack, security-test, or audit their web app, website, web application, or staging site.
license: Apache-2.0
metadata:
author: usestrix
homepage: https://docs.strix.ai
---
# Pentest a web application
Black-box (and optionally source-assisted) penetration testing of a running web app with Strix's autonomous agents. Every reported finding is validated with a working exploit, so there are no signature-based false positives to triage.
Install, LLM setup, all CLI flags, and the managed-cloud alternative are covered in the **penetration-testing-with-strix** skill — read it if the target is not a running web app, or if `strix --version` fails. This skill is the web-app-specific workflow.
## 1. Confirm authorization and scope
Before running anything, establish:
- **The target is the user's** (or they are explicitly authorized to test it). Never pentest a third-party site on a hunch.
- **Which environment.** Prefer staging over production; agents send real exploit payloads and will create/modify data.
- **Out-of-scope paths** — payment flows, mass-email endpoints, admin destructive actions, third-party SSO providers.
- **Credentials.** Most real vulnerabilities live behind login. Without a test account, the agents only ever see the marketing surface.
Ask for anything missing rather than guessing.
## 2. Run the scan
```bash
strix -n -t https://staging.example.com --max-budget 20 \
--instruction "Test account: qa@example.com / <password>. In scope: /app/*, /api/*. Do not touch /billing or send email. Focus on access control between the two seeded orgs."
```
Notes that matter for web apps specifically:
- **Give it credentials via `--instruction`** (or `--instruction-file` for anything long), including how to log in if the flow is unusual (magic link, SSO, MFA-exempt test user).
- **Two accounts beat one.** Multi-tenant IDOR and broken-access-control bugs — consistently the highest-impact class in web apps — can only be proven when the agent can attempt cross-account access.
- **Add the repo for white-box depth** when you have the source: `-t https://github.com/org/app -t https://staging.example.com` (or a local path). Source access materially improves coverage of business-logic and authorization flaws.
- **Localhost works.** Point at `http://host.docker.internal:3000` (Docker Desktop) so the sandbox can reach a dev server on the host.
- `--scan-mode quick` for a fast dev-loop pass, `standard` (~30 min) for a normal review, `deep` for pre-release assurance. Always set `--max-budget`.
For a hosted run with no Docker/LLM key, or when the user wants a shareable dashboard and an auditor-ready PDF, use the cloud path in **managed-pentesting-with-strix** instead — same engine, same findings.
## 3. Review results
Read `strix_runs/<run>/penetration_test_report.md` first, then per-finding files in `vulnerabilities/`. Each contains the PoC — re-run it yourself to confirm before reporting to the user.
Exit codes: `0` no validated vulns in what was analyzed, `2` vulnerabilities found, `1` fatal error. A `0` is not proof of full coverage — if the budget or turn cap was hit the scan wraps up early, so check `run.json` status and cost against `--max-budget` before calling the app clean.
## 4. Fix and verify
Hand findings to the **fix-security-vulnerabilities-with-strix** skill: patch the root cause, then re-run Strix against the same target to prove the exploit no longer works. Re-testing is the only reliable confirmation a fix landed.
To keep the app tested on every change rather than once, wire Strix into CI with **ci-security-scanning-with-strix**.
+47 -10
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import dataclasses
import inspect
import json
import logging
@@ -222,6 +223,17 @@ def _with_coerced_arguments(tool: FunctionTool) -> FunctionTool:
return tool
def _with_strictness(tool: FunctionTool, strict_schemas: bool) -> FunctionTool:
"""Drop strict JSON-schema mode when the route can't take it (see
``supports_strict_tool_schemas``); the tool stays functionally identical.
Returns a copy so the shared tool singletons keep their declared mode.
"""
if strict_schemas or not tool.strict_json_schema:
return tool
return dataclasses.replace(tool, strict_json_schema=False)
def _function_tool_with_error_result(tool: FunctionTool) -> FunctionTool:
invoke_tool = tool.on_invoke_tool
@@ -285,24 +297,38 @@ def _bound_custom_tool(tool: CustomTool) -> CustomTool:
return tool
def _configure_filesystem_tools(toolset: Any, *, chat_completions: bool) -> None:
def _configure_filesystem_tools(
toolset: Any, *, chat_completions: bool, strict_schemas: bool = True
) -> None:
for name, tool in vars(toolset).items():
if chat_completions:
if isinstance(tool, CustomTool):
setattr(toolset, name, _custom_tool_as_function_tool(tool))
elif isinstance(tool, FunctionTool):
setattr(
toolset, name, _function_tool_with_error_result(_with_coerced_arguments(tool))
toolset,
name,
_function_tool_with_error_result(
_with_strictness(_with_coerced_arguments(tool), strict_schemas)
),
)
elif isinstance(tool, CustomTool):
setattr(toolset, name, _bound_custom_tool(tool))
elif isinstance(tool, FunctionTool):
setattr(toolset, name, _with_bounded_result(_with_coerced_arguments(tool)))
setattr(
toolset,
name,
_with_bounded_result(
_with_strictness(_with_coerced_arguments(tool), strict_schemas)
),
)
def _make_filesystem_configurator(*, chat_completions: bool) -> Any:
def _make_filesystem_configurator(*, chat_completions: bool, strict_schemas: bool) -> Any:
def configure(toolset: Any) -> None:
_configure_filesystem_tools(toolset, chat_completions=chat_completions)
_configure_filesystem_tools(
toolset, chat_completions=chat_completions, strict_schemas=strict_schemas
)
return configure
@@ -406,11 +432,13 @@ def _wrap_write_stdin(tool: FunctionTool) -> FunctionTool:
return tool
def _configure_shell_tools(toolset: Any, *, chat_completions: bool) -> None:
def _configure_shell_tools(
toolset: Any, *, chat_completions: bool, strict_schemas: bool = True
) -> None:
for name, tool in vars(toolset).items():
if not isinstance(tool, FunctionTool):
continue
wrapped = _with_coerced_arguments(tool)
wrapped = _with_strictness(_with_coerced_arguments(tool), strict_schemas)
if tool.name == "exec_command":
wrapped = _wrap_exec_command(wrapped)
elif tool.name == "write_stdin":
@@ -420,9 +448,11 @@ def _configure_shell_tools(toolset: Any, *, chat_completions: bool) -> None:
setattr(toolset, name, wrapped)
def _make_shell_configurator(*, chat_completions: bool) -> Any:
def _make_shell_configurator(*, chat_completions: bool, strict_schemas: bool) -> Any:
def configure(toolset: Any) -> None:
_configure_shell_tools(toolset, chat_completions=chat_completions)
_configure_shell_tools(
toolset, chat_completions=chat_completions, strict_schemas=strict_schemas
)
return configure
@@ -568,6 +598,7 @@ def build_strix_agent(
is_whitebox: bool = False,
interactive: bool = False,
chat_completions_tools: bool = False,
strict_tool_schemas: bool = True,
system_prompt_context: dict[str, Any] | None = None,
extra_tools: Sequence[Tool] | None = None,
instructions_override: str | None = None,
@@ -577,6 +608,8 @@ def build_strix_agent(
Args:
chat_completions_tools: Wrap SDK custom tools as function tools
when the selected backend cannot accept Responses custom tools.
strict_tool_schemas: Send function tools as strict-schema tools. Off
for routes that reject a toolset this size as strict.
extra_tools: Additional tools for this scan agent only, on top of any
registered via ``register_agent_tools``.
instructions_override: Use this verbatim as the system prompt instead
@@ -604,7 +637,7 @@ def build_strix_agent(
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
_ensure_unique_tool_names(tools)
tools = [
_with_bounded_result(_with_coerced_arguments(tool))
_with_bounded_result(_with_strictness(_with_coerced_arguments(tool), strict_tool_schemas))
if isinstance(tool, FunctionTool)
else tool
for tool in tools
@@ -630,11 +663,13 @@ def build_strix_agent(
Filesystem(
configure_tools=_make_filesystem_configurator(
chat_completions=chat_completions_tools,
strict_schemas=strict_tool_schemas,
),
),
Shell(
configure_tools=_make_shell_configurator(
chat_completions=chat_completions_tools,
strict_schemas=strict_tool_schemas,
),
),
],
@@ -647,6 +682,7 @@ def make_child_factory(
is_whitebox: bool = False,
interactive: bool = False,
chat_completions_tools: bool = False,
strict_tool_schemas: bool = True,
system_prompt_context: dict[str, Any] | None = None,
) -> Any:
"""Return the runner-owned builder used by ``spawn_child_agent``.
@@ -665,6 +701,7 @@ def make_child_factory(
is_whitebox=is_whitebox,
interactive=interactive,
chat_completions_tools=chat_completions_tools,
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=system_prompt_context,
)
+15
View File
@@ -749,6 +749,18 @@ def uses_chat_completions_tool_schema(model_name: str, settings: Settings) -> bo
return not model_supports_reasoning(model_name)
def supports_strict_tool_schemas(model_name: str) -> bool:
"""Return whether the route accepts strict tool schemas for Strix's toolset.
Claude caps a request at 20 strict tools and 16 union-typed parameters
across all strict schemas. Strix ships ~30 tools and the strict dialect
turns every optional parameter into a nullable union, so both caps are
exceeded and the request is rejected outright.
"""
name = model_name.strip().lower()
return not any(marker in name for marker in _ANTHROPIC_MODEL_MARKERS)
def model_supports_reasoning(model_name: str) -> bool:
import litellm
@@ -845,6 +857,9 @@ def is_known_openai_bare_model(model_name: str) -> bool:
return bool(entry and entry.get("litellm_provider") == "openai")
_ANTHROPIC_MODEL_MARKERS = ("anthropic", "claude", "sonnet", "opus", "haiku")
def is_claude_model(model_name: str) -> bool:
return "claude" in (model_name or "").strip().lower()
+6 -51
View File
@@ -22,6 +22,7 @@ from strix.config import load_settings
from strix.config.models import (
StrixProvider,
configure_sdk_model_defaults,
supports_strict_tool_schemas,
uses_chat_completions_tool_schema,
)
from strix.config.settings import DEFAULT_MAX_TURNS
@@ -51,12 +52,10 @@ from strix.tools.output_store import (
if TYPE_CHECKING:
from agents.mcp import MCPServer
from agents.memory import SQLiteSession
from agents.result import RunResultBase
from strix.runtime.status import StatusSink
from strix.tools.mcp import ConnectedMcpServer
logger = logging.getLogger(__name__)
@@ -64,33 +63,6 @@ logger = logging.getLogger(__name__)
StreamEventSink = Callable[[str, Any], None]
def _mcp_startup_summary(connections: list[ConnectedMcpServer]) -> str:
"""One user-facing line summarizing the MCP servers that connected."""
server_count = len(connections)
tool_count = sum(c.tool_count for c in connections)
servers_word = "server" if server_count == 1 else "servers"
tools_word = "tool" if tool_count == 1 else "tools"
names = ", ".join(c.name for c in connections)
return f"MCP: connected {server_count} {servers_word} ({tool_count} {tools_word}): {names}"
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,
@@ -204,6 +176,9 @@ async def run_strix_scan(
)
logger.info("LLM model resolved: %s", resolved_model)
chat_completions_tools = uses_chat_completions_tool_schema(resolved_model, settings)
strict_tool_schemas = supports_strict_tool_schemas(resolved_model)
if not strict_tool_schemas:
logger.info("Sending non-strict tool schemas: %s caps strict tools", resolved_model)
if coordinator is None:
coordinator = AgentCoordinator()
@@ -282,7 +257,6 @@ async def run_strix_scan(
configure_spill_writer(_spill_to_workspace)
sessions_to_close: list[SQLiteSession] = []
mcp_servers: list[MCPServer] = []
try:
targets = scan_config.get("targets") or []
@@ -328,24 +302,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]
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,
@@ -354,6 +310,7 @@ async def run_strix_scan(
is_whitebox=is_whitebox,
interactive=interactive,
chat_completions_tools=chat_completions_tools,
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=root_context,
instructions_override=root_instructions,
)
@@ -372,6 +329,7 @@ async def run_strix_scan(
is_whitebox=is_whitebox,
interactive=interactive,
chat_completions_tools=chat_completions_tools,
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=scope_context,
)
@@ -520,9 +478,6 @@ async def run_strix_scan(
for s in sessions_to_close:
with contextlib.suppress(Exception):
s.close()
for mcp_server in mcp_servers:
with contextlib.suppress(Exception):
await mcp_server.cleanup() # type: ignore[no-untyped-call]
with contextlib.suppress(Exception):
await coordinator._maybe_snapshot()
if cleanup_on_exit:
-39
View File
@@ -3,7 +3,6 @@
from __future__ import annotations
import argparse
import os
import sys
from pathlib import Path
@@ -220,30 +219,6 @@ Examples:
help="Path to a custom config file (JSON) to use instead of ~/.strix/cli-config.json",
)
parser.add_argument(
"--mcp-config",
type=str,
metavar="PATH",
help="Path to an MCP servers JSON file to use instead of ~/.strix/mcp-servers.json.",
)
parser.add_argument(
"--mcp-server",
dest="mcp_server",
action="append",
metavar="NAME",
help="Use only this MCP connection for the run, by its config name "
"(repeatable). Every other configured connection is skipped.",
)
parser.add_argument(
"--mcp-exclude",
dest="mcp_exclude",
action="append",
metavar="NAME",
help="Skip this MCP connection for the run, by its config name (repeatable).",
)
parser.add_argument(
"--max-budget",
"--max-budget-usd",
@@ -292,20 +267,6 @@ Examples:
if args.config:
apply_config_override(validate_config_file(args.config))
if args.mcp_config:
mcp_config_path = Path(args.mcp_config).expanduser()
if not mcp_config_path.is_file():
parser.error(f"--mcp-config file not found: {args.mcp_config}")
# The MCP loader reads this env var as its config-path override, so
# setting it here makes the flag win over the default location.
os.environ["STRIX_MCP_CONFIG"] = str(mcp_config_path)
# The MCP loader reads these as its per-run include/exclude selection.
if args.mcp_server:
os.environ["STRIX_MCP_ONLY"] = ",".join(args.mcp_server)
if args.mcp_exclude:
os.environ["STRIX_MCP_EXCLUDE"] = ",".join(args.mcp_exclude)
if args.update:
sys.exit(0 if self_update() else 1)
@@ -22,20 +22,19 @@ func statusIcon(status string) (string, lipgloss.Style) {
return "○ Unknown", Dim()
}
// renderGenericTool ports registry._render_default_tool_widget. It shows the
// tool name, its arguments, and a status line only. The raw result is
// deliberately not rendered: a generic/MCP result (e.g. a multi-kilobyte JSON
// payload from a database query tool) is noise on screen, and the agent narrates
// what it got in its next message. The full result still lives in the event
// data, the run log, and the `strix view` viewer.
func renderGenericTool(name string, args map[string]any, status string) string {
// renderGenericTool ports registry._render_default_tool_widget.
func renderGenericTool(name string, args map[string]any, result any, status string) string {
var b strings.Builder
b.WriteString(Dim().Render("→ Using tool ") + Bold(Blue).Render(name) + "\n")
for _, k := range SortedKeys(args) {
b.WriteString(" " + Dim().Render(k) + ": " + StringValue(args[k]) + "\n")
}
icon, style := statusIcon(status)
b.WriteString(style.Render(icon))
if (status == "completed" || status == "failed" || status == "error") && result != nil {
b.WriteString(lipgloss.NewStyle().Bold(true).Render("Result: ") + StringValue(result))
} else {
icon, style := statusIcon(status)
b.WriteString(style.Render(icon))
}
return b.String()
}
@@ -88,7 +87,7 @@ func Tool(data map[string]any) string {
case "list_requests", "view_request", "repeat_request", "list_sitemap", "view_sitemap_entry", "scope_rules":
return renderProxyTool(name, args, result, status)
}
return renderGenericTool(name, args, status)
return renderGenericTool(name, args, result, status)
}
// ---------------------------------------------------------------------------
@@ -203,7 +203,7 @@ func TestToolDispatchCoversKnownTools(t *testing.T) {
{
"unknown tool falls back to generic",
tool("brand_new_tool", map[string]any{"alpha": "1"}, "done", "completed"),
[]string{"brand_new_tool", "alpha", "Done"},
[]string{"brand_new_tool", "alpha", "Result:", "done"},
},
}
@@ -214,18 +214,6 @@ func TestToolDispatchCoversKnownTools(t *testing.T) {
}
}
func TestGenericToolOmitsRawResult(t *testing.T) {
// The generic/MCP renderer shows tool name, args, and a status line only —
// never the raw result payload.
long := strings.Repeat("x", 5000)
out := ansi.Strip(Tool(tool("db_query", map[string]any{"query": "select 1"}, long, "completed")))
requireContains(t, out, "db_query", "query", "Done")
if strings.Contains(out, "Result:") || strings.Contains(out, strings.Repeat("x", 20)) {
t.Fatalf("generic result body must not be rendered:\n%s", out)
}
}
func TestCollapseToolShellPreviewAndExpand(t *testing.T) {
lines := make([]string, 16)
for i := range lines {
-21
View File
@@ -1,21 +0,0 @@
"""Generic MCP client: connect MCP servers and expose their tools."""
from __future__ import annotations
from strix.tools.mcp.client import ConnectedMcpServer, connect_mcp_servers
from strix.tools.mcp.config import (
BearerAuth,
McpAuth,
McpConnectionConfig,
)
from strix.tools.mcp.loader import load_user_mcp_configs
__all__ = [
"BearerAuth",
"ConnectedMcpServer",
"McpAuth",
"McpConnectionConfig",
"connect_mcp_servers",
"load_user_mcp_configs",
]
-367
View File
@@ -1,367 +0,0 @@
"""Connect to MCP servers and expose their tools to the agent.
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.
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.
"""
from __future__ import annotations
import contextlib
import json
import logging
import re
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
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
# 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.
ResultTransform = Callable[[str, Any], Any]
logger = logging.getLogger(__name__)
# A tool name offered to a model has to be letters, digits, underscores or
# hyphens; anything else is rejected outright by the model APIs. Three things can
# put a stray character in one: the separator between the connection and the tool
# name, a name the server chose for its own tool (servers commonly namespace
# theirs), and the connection name out of the user's config file. Sanitizing the
# finished name covers all three rather than only the separator.
_INVALID_TOOL_NAME_CHARS = re.compile(r"[^a-zA-Z0-9_-]")
def _namespaced_tool_name(connection: str, tool: str) -> str:
"""The name a connection's tool is offered to the model under.
Only the model-facing name is rewritten. Every call to the server uses the
tool name the server itself reported, so sanitizing here can never change
which tool is invoked.
"""
return _INVALID_TOOL_NAME_CHARS.sub("_", f"{connection}_{tool}")
class ConnectedMcpServer(NamedTuple):
"""One successfully connected MCP server and how many tools it registered.
``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;
``notes`` carries the connection's optional free-text description so the
caller can surface it to the agent as context about the connection.
"""
server: MCPServer
name: str
tool_count: int
notes: str | None = None
def _auth_headers(config: McpConnectionConfig) -> dict[str, str]:
"""Build the per-server request headers from the connection's auth."""
auth = config.auth
if auth is None:
return {}
return {"Authorization": f"Bearer {auth.token}"}
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.
"""
tool_filter = (
create_static_tool_filter(allowed_tool_names=config.allowed_tools)
if config.allowed_tools is not None
else None
)
if config.transport == "stdio":
stdio_params: MCPServerStdioParams = {
"command": cast("str", config.command),
"args": config.args,
"env": config.env,
}
return MCPServerStdio(
params=stdio_params,
name=config.name,
tool_filter=tool_filter,
cache_tools_list=True,
)
http_params: MCPServerStreamableHttpParams = {
"url": cast("str", config.url),
"headers": _auth_headers(config),
}
return MCPServerStreamableHttp(
params=http_params,
name=config.name,
tool_filter=tool_filter,
cache_tools_list=True,
)
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
its own call, this is what makes the agent see byte-identical content to what
the SDK would have produced on its own.
"""
if getattr(server, "use_structured_content", False) and result.structuredContent:
return json.dumps(result.structuredContent)
outputs: list[dict[str, Any]] = []
for item in result.content:
if item.type == "text":
outputs.append({"type": "text", "text": item.text})
elif item.type == "image":
outputs.append(
{"type": "image", "image_url": f"data:{item.mimeType};base64,{item.data}"}
)
else:
outputs.append({"type": "text", "text": str(item.model_dump(mode="json"))})
if len(outputs) == 1:
return outputs[0]
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,
server: MCPServer,
result_transform: ResultTransform | None = None,
) -> list[Tool]:
"""List a connected server's tools, prefix + filter them, and register them.
``allowed_tools`` of ``None`` registers every listed tool; a list restricts
to exactly those names.
"""
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
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.
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.
"""
connected: list[ConnectedMcpServer] = []
for config in configs:
server: MCPServer | None = None
try:
server = _build_server(config)
await server.connect() # type: ignore[no-untyped-call]
tools = await _register_server_tools(config, server, result_transform)
except Exception:
logger.exception("Skipping MCP connection %r", config.name)
if server is not None:
with contextlib.suppress(Exception):
await server.cleanup() # type: ignore[no-untyped-call]
continue
except BaseException:
# A cancellation (or other non-Exception failure) mid-connect must not
# orphan MCP subprocesses or HTTP sessions. Clean up the server being
# connected and every server already connected, then re-raise so the
# caller still stops. The runner only receives the list on a clean
# return, so on an abnormal exit this function owns the cleanup.
if server is not None:
with contextlib.suppress(Exception):
await server.cleanup() # type: ignore[no-untyped-call]
for established in connected:
with contextlib.suppress(Exception):
await established.server.cleanup() # type: ignore[no-untyped-call]
raise
logger.info("Connected MCP server %r (%d tools)", config.name, len(tools))
connected.append(
ConnectedMcpServer(
server=server, name=config.name, tool_count=len(tools), notes=config.notes
)
)
return connected
-73
View File
@@ -1,73 +0,0 @@
"""The connection-config contract for the MCP client.
Describes one MCP server the client can connect to: its transport, endpoint or
launch command, optional auth, and an optional tool allowlist. Field names are
stable; callers build against them.
"""
from __future__ import annotations
from typing import Annotated, Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
class BearerAuth(BaseModel):
"""Header-token auth, sent as ``Authorization: Bearer <token>``."""
model_config = ConfigDict(extra="forbid")
kind: Literal["bearer"] = "bearer"
token: str = Field(min_length=1, repr=False)
McpAuth = Annotated[BearerAuth, Field(discriminator="kind")]
class McpConnectionConfig(BaseModel):
"""One MCP server the client can connect to.
Two transports are supported: streamable ``http`` (a remote endpoint) and
``stdio`` (a local server launched as a subprocess).
"""
model_config = ConfigDict(extra="forbid")
name: str = Field(min_length=1)
"""Namespaced tool prefix, unique per run (e.g. ``github``)."""
transport: Literal["http", "stdio"] = "http"
"""``http`` for a streamable HTTP endpoint, ``stdio`` for a local subprocess."""
url: str | None = Field(default=None, min_length=1)
"""The MCP server endpoint. Required for ``http``."""
auth: McpAuth | None = None
"""Bearer token for the server. Optional; a local stdio server usually
needs none."""
command: str | None = Field(default=None, min_length=1)
"""The executable to launch for ``stdio``. Required for ``stdio``."""
args: list[str] = Field(default_factory=list)
"""Arguments passed to ``command`` (stdio only)."""
env: dict[str, str] = Field(default_factory=dict)
"""Extra environment variables for the stdio subprocess."""
allowed_tools: list[str] | None = None
"""Tool allowlist, applied after the server lists its tools. ``None`` (the
default) exposes every tool the server lists; a list restricts to it."""
notes: str | None = None
"""Free-text notes for the agent describing what this connection is and how
to use it. When set, they are prefixed onto each of the connection's tool
descriptions so the agent sees them."""
@model_validator(mode="after")
def _check_transport_fields(self) -> McpConnectionConfig:
if self.transport == "http" and not self.url:
raise ValueError("an http MCP connection requires 'url'")
if self.transport == "stdio" and not self.command:
raise ValueError("a stdio MCP connection requires 'command'")
return self
-132
View File
@@ -1,132 +0,0 @@
"""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.
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
or unreadable file yields an empty list.
"""
from __future__ import annotations
import json
import logging
import os
from pathlib import Path
from typing import cast
from pydantic import ValidationError
from strix.tools.mcp.config import McpConnectionConfig
logger = logging.getLogger(__name__)
_DEFAULT_PATH: Path = Path.home() / ".strix" / "mcp-servers.json"
_PATH_ENV_VAR = "STRIX_MCP_CONFIG"
# Per-run selection, set by the --mcp-server / --mcp-exclude CLI flags. Each is a
# comma-separated list of connection names.
_ONLY_ENV_VAR = "STRIX_MCP_ONLY"
_EXCLUDE_ENV_VAR = "STRIX_MCP_EXCLUDE"
def _resolve_path(path: Path | None) -> Path:
if path is not None:
return path
override = os.environ.get(_PATH_ENV_VAR)
if override:
return Path(override)
return _DEFAULT_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.
"""
seen: set[str] = set()
unique: list[McpConnectionConfig] = []
for config in configs:
if config.name in seen:
logger.warning(
"Ignoring MCP server %r: another connection already uses that name "
"(names must be unique because they namespace the server's tools).",
config.name,
)
continue
seen.add(config.name)
unique.append(config)
return unique
def _parse_names(env_var: str) -> set[str]:
return {name.strip() for name in os.environ.get(env_var, "").split(",") if name.strip()}
def _apply_run_selection(configs: list[McpConnectionConfig]) -> list[McpConnectionConfig]:
"""Restrict this run's connections to an optional include/exclude selection.
``STRIX_MCP_ONLY`` (if set) keeps only the named connections; then
``STRIX_MCP_EXCLUDE`` drops any named connection. With neither set, every
connection is kept.
"""
only = _parse_names(_ONLY_ENV_VAR)
exclude = _parse_names(_EXCLUDE_ENV_VAR)
if not only and not exclude:
return configs
available = {config.name for config in configs}
for name in sorted((only | exclude) - available):
logger.warning(
"MCP connection selection named %r, which is not configured; ignoring it", name
)
selected: list[McpConnectionConfig] = []
for config in configs:
if only and config.name not in only:
continue
if config.name in exclude:
continue
selected.append(config)
return selected
def load_user_mcp_configs(path: Path | None = None) -> list[McpConnectionConfig]:
"""Load MCP connection configs from the user's JSON file.
The path is ``path`` if given, else ``$STRIX_MCP_CONFIG``, else
``~/.strix/mcp-servers.json``. The file is a JSON list of server entries.
A missing file returns ``[]``; an unreadable or non-list file is logged and
returns ``[]``; individual entries that fail validation are logged and
skipped. Connections sharing a name are de-duplicated (first wins), and an
optional per-run include/exclude selection is applied last.
"""
source = _resolve_path(path)
if not source.exists():
return []
try:
raw = json.loads(source.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
logger.exception("Could not read MCP config at %s; ignoring it", source)
return []
if not isinstance(raw, list):
logger.warning("MCP config at %s is not a JSON list; ignoring it", source)
return []
entries = cast("list[object]", raw)
configs: list[McpConnectionConfig] = []
for index, entry in enumerate(entries):
try:
configs.append(McpConnectionConfig.model_validate(entry))
except ValidationError as exc:
logger.warning("Skipping invalid MCP server entry #%d in %s: %s", index, source, exc)
return _apply_run_selection(_dedupe_by_name(configs))
+16
View File
@@ -112,3 +112,19 @@ def test_wait_for_agents_is_available_in_both_modes() -> None:
for interactive in (True, False):
agent = factory.build_strix_agent(is_root=True, interactive=interactive)
assert "wait_for_agents" in [t.name for t in agent.tools]
def test_strict_tool_schemas_can_be_disabled_per_route() -> None:
"""Claude routes cap strict tools; the toolset must be sendable without strict."""
agent = factory.build_strix_agent(is_root=True, strict_tool_schemas=False)
function_tools = [t for t in agent.tools if isinstance(t, FunctionTool)]
assert function_tools
assert not any(t.strict_json_schema for t in function_tools)
def test_disabling_strict_leaves_shared_tools_untouched() -> None:
factory.build_strix_agent(is_root=True, strict_tool_schemas=False)
agent = factory.build_strix_agent(is_root=True)
assert any(t.strict_json_schema for t in agent.tools if isinstance(t, FunctionTool))
-88
View File
@@ -1,88 +0,0 @@
"""Tests for the --mcp-config CLI flag."""
from __future__ import annotations
import importlib
import os
import sys
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
import pytest
if TYPE_CHECKING:
from pathlib import Path
cli_main: Any = importlib.import_module("strix.interface.main")
def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
cli_main,
"load_settings",
lambda: SimpleNamespace(runtime=SimpleNamespace(max_local_copy_mb=1024)),
)
def test_mcp_config_flag_sets_loader_override(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
config = tmp_path / "servers.json"
config.write_text("[]", encoding="utf-8")
_stub_settings(monkeypatch)
# delenv records "originally absent" so monkeypatch removes whatever the
# parser sets, keeping the override from leaking into other tests.
monkeypatch.delenv("STRIX_MCP_CONFIG", raising=False)
monkeypatch.setattr(
sys, "argv", ["strix", "-t", "https://test.com/", "-n", "--mcp-config", str(config)]
)
args = cli_main.parse_arguments()
assert args.mcp_config == str(config)
assert os.environ["STRIX_MCP_CONFIG"] == str(config)
def test_mcp_config_flag_rejects_missing_file(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
_stub_settings(monkeypatch)
monkeypatch.delenv("STRIX_MCP_CONFIG", raising=False)
missing = tmp_path / "nope.json"
monkeypatch.setattr(
sys, "argv", ["strix", "-t", "https://test.com/", "-n", "--mcp-config", str(missing)]
)
with pytest.raises(SystemExit):
cli_main.parse_arguments()
assert "--mcp-config file not found" in capsys.readouterr().err
def test_mcp_server_flags_set_selection_env(monkeypatch: pytest.MonkeyPatch) -> None:
_stub_settings(monkeypatch)
monkeypatch.delenv("STRIX_MCP_ONLY", raising=False)
monkeypatch.delenv("STRIX_MCP_EXCLUDE", raising=False)
monkeypatch.setattr(
sys,
"argv",
[
"strix",
"-t",
"https://test.com/",
"-n",
"--mcp-server",
"a",
"--mcp-server",
"b",
"--mcp-exclude",
"c",
],
)
cli_main.parse_arguments()
assert os.environ["STRIX_MCP_ONLY"] == "a,b"
assert os.environ["STRIX_MCP_EXCLUDE"] == "c"
-661
View File
@@ -1,661 +0,0 @@
"""Tests for the generic MCP client: config contract, namespacing, and filtering."""
from __future__ import annotations
import asyncio
import json
from typing import TYPE_CHECKING, Any
import re
import pytest
from agents.mcp import MCPServer, MCPServerStdio, MCPServerStreamableHttp
from mcp.types import CallToolResult, TextContent
from mcp.types import Tool as MCPTool
from pydantic import ValidationError
from strix.agents import factory
from strix.core.runner import _mcp_connection_notes
from strix.interface.tui.live_view import _tool_status_from_result
from strix.tools.mcp import (
BearerAuth,
ConnectedMcpServer,
McpConnectionConfig,
load_user_mcp_configs,
)
from strix.tools.mcp import client as mcp_client
from strix.tools.mcp.client import _auth_headers, _build_server, _register_server_tools
if TYPE_CHECKING:
from pathlib import Path
from agents.tool import Tool
class FakeMCPServer(MCPServer):
"""A connected MCP server stand-in, so tests never touch the network."""
def __init__(self, name: str, tools: list[MCPTool]) -> None:
super().__init__()
self._name = name
self._tools = tools
self.calls: list[tuple[str, dict[str, Any] | None]] = []
@property
def name(self) -> str:
return self._name
async def connect(self) -> None:
return None
async def cleanup(self) -> None:
return None
async def list_tools(
self,
run_context: Any = None,
agent: Any = None,
) -> list[MCPTool]:
return list(self._tools)
async def call_tool(
self,
tool_name: str,
arguments: dict[str, Any] | None,
meta: dict[str, Any] | None = None,
) -> CallToolResult:
self.calls.append((tool_name, arguments))
return CallToolResult(content=[TextContent(type="text", text=f"routed:{tool_name}")])
async def list_prompts(self) -> Any:
raise NotImplementedError
async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> Any:
raise NotImplementedError
def _mcp_tool(name: str) -> MCPTool:
return MCPTool(
name=name,
description=f"remote tool {name}",
inputSchema={"type": "object", "properties": {}},
)
def _config(name: str, allowed_tools: list[str]) -> McpConnectionConfig:
return McpConnectionConfig(
name=name,
url="https://mcp.example.com",
auth=BearerAuth(token="run-token"),
allowed_tools=allowed_tools,
)
@pytest.fixture(autouse=True)
def _reset_registry() -> Any:
saved = list(factory._EXTRA_TOOLS)
factory._EXTRA_TOOLS.clear()
try:
yield
finally:
factory._EXTRA_TOOLS[:] = saved
# --- config contract ---------------------------------------------------------
def test_bearer_config_parses_from_dict() -> None:
config = McpConnectionConfig.model_validate(
{
"name": "files_main",
"transport": "http",
"url": "https://mcp.example.com",
"auth": {"kind": "bearer", "token": "abc"},
"allowed_tools": ["list_files"],
}
)
assert isinstance(config.auth, BearerAuth)
assert config.auth.token == "abc"
assert config.allowed_tools == ["list_files"]
def test_unknown_auth_kind_is_rejected() -> None:
with pytest.raises(ValidationError):
McpConnectionConfig.model_validate(
{
"name": "x",
"url": "https://mcp.example.com",
"auth": {"kind": "oauth", "token": "abc"},
}
)
def test_stdio_config_parses_from_dict() -> None:
config = McpConnectionConfig.model_validate(
{
"name": "local_fs",
"transport": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/srv/data"],
"env": {"FOO": "bar"},
}
)
assert config.transport == "stdio"
assert config.command == "npx"
assert config.args == ["-y", "@modelcontextprotocol/server-filesystem", "/srv/data"]
assert config.env == {"FOO": "bar"}
# A local stdio server needs no auth, and omitting allowed_tools means "all".
assert config.auth is None
assert config.allowed_tools is None
def test_http_config_without_url_is_rejected() -> None:
with pytest.raises(ValidationError):
McpConnectionConfig.model_validate(
{
"name": "x",
"transport": "http",
"auth": {"kind": "bearer", "token": "abc"},
}
)
def test_stdio_config_without_command_is_rejected() -> None:
with pytest.raises(ValidationError):
McpConnectionConfig.model_validate(
{
"name": "x",
"transport": "stdio",
}
)
def test_empty_name_is_rejected() -> None:
with pytest.raises(ValidationError):
McpConnectionConfig.model_validate(
{
"name": "",
"url": "https://mcp.example.com",
"auth": {"kind": "bearer", "token": "abc"},
}
)
def test_unknown_field_is_rejected() -> None:
with pytest.raises(ValidationError):
McpConnectionConfig.model_validate(
{
"name": "x",
"url": "https://mcp.example.com",
"auth": {"kind": "bearer", "token": "abc"},
"surprise": True,
}
)
# --- auth headers ------------------------------------------------------------
def test_bearer_auth_builds_authorization_header() -> None:
headers = _auth_headers(_config("files_main", []))
assert headers == {"Authorization": "Bearer run-token"}
# --- namespacing and filtering -----------------------------------------------
def _registered_names() -> list[str]:
return [tool.name for tool in factory.registered_agent_tools()]
@pytest.mark.asyncio
async def test_tools_are_namespaced_per_connection() -> None:
server_a = FakeMCPServer("conn_a", [_mcp_tool("describe")])
server_b = FakeMCPServer("conn_b", [_mcp_tool("describe")])
await _register_server_tools(_config("conn_a", ["describe"]), server_a)
await _register_server_tools(_config("conn_b", ["describe"]), server_b)
# Same remote tool name on two connections does not collide.
assert _registered_names() == ["conn_a_describe", "conn_b_describe"]
@pytest.mark.asyncio
async def test_registered_names_are_valid_tool_names() -> None:
# Model APIs reject a tool name containing anything but letters, digits,
# underscores and hyphens, and reject the whole request rather than the one
# tool. A server naming its own tools with dots, or a connection named with
# a space in the user's config, must not be able to break a run.
server = FakeMCPServer("my server", [_mcp_tool("db.query"), _mcp_tool("ok_tool")])
await _register_server_tools(_config("my server", None), server)
names = _registered_names()
assert names == ["my_server_db_query", "my_server_ok_tool"]
assert all(re.fullmatch(r"[a-zA-Z0-9_-]{1,128}", name) for name in names)
@pytest.mark.asyncio
async def test_a_rename_does_not_change_which_tool_is_called() -> None:
# Only the model-facing name is sanitized; the server is always asked for the
# tool name it reported.
server = FakeMCPServer("my server", [_mcp_tool("db.query")])
tools = await _register_server_tools(_config("my server", None), server)
assert tools[0].name == "my_server_db_query"
await tools[0].on_invoke_tool(None, "{}")
assert server.calls == [("db.query", {})]
@pytest.mark.asyncio
async def test_disallowed_tool_is_not_registered() -> None:
server = FakeMCPServer(
"files_main",
[_mcp_tool("list_files"), _mcp_tool("search")],
)
await _register_server_tools(_config("files_main", ["list_files"]), server)
names = _registered_names()
assert "files_main_list_files" in names
assert "files_main_search" not in names
@pytest.mark.asyncio
async def test_allowed_tools_none_registers_every_listed_tool() -> None:
server = FakeMCPServer(
"local_fs",
[_mcp_tool("read_file"), _mcp_tool("write_file")],
)
config = McpConnectionConfig(name="local_fs", url="https://mcp.example.com", allowed_tools=None)
await _register_server_tools(config, server)
names = _registered_names()
assert "local_fs_read_file" in names
assert "local_fs_write_file" in names
@pytest.mark.asyncio
async def test_allowed_tools_list_restricts_registration() -> None:
server = FakeMCPServer(
"local_fs",
[_mcp_tool("read_file"), _mcp_tool("write_file")],
)
await _register_server_tools(_config("local_fs", ["read_file"]), server)
names = _registered_names()
assert names == ["local_fs_read_file"]
@pytest.mark.asyncio
async def test_registered_tool_routes_to_its_server_with_the_original_name() -> None:
server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
tools: list[Tool] = await _register_server_tools(
_config("files_main", ["list_files"]), server
)
tool = tools[0]
output = await tool.on_invoke_tool(None, "{}") # type: ignore[union-attr]
# The call reaches the right server, addressed by the unprefixed remote name.
assert server.calls == [("list_files", {})]
assert output == {"type": "text", "text": "routed:list_files"}
# --- result transform --------------------------------------------------------
@pytest.mark.asyncio
async def test_result_transform_receives_namespaced_name_and_structured_result() -> None:
server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
seen: list[tuple[str, Any]] = []
def transform(name: str, structured: Any) -> Any:
seen.append((name, structured))
return {"kept": structured["content"][0]["text"]}
tools: list[Tool] = await _register_server_tools(
_config("files_main", ["list_files"]), server, result_transform=transform
)
output = await tools[0].on_invoke_tool(None, "{}") # type: ignore[union-attr]
# The underlying MCP call still routes by the unprefixed remote name.
assert server.calls == [("list_files", {})]
# The transform is called with the namespaced name and the parsed result.
assert len(seen) == 1
name, structured = seen[0]
assert name == "files_main_list_files"
# A parsed CallToolResult (dict/list), not a pre-serialized string.
assert structured["content"][0]["text"] == "routed:list_files"
assert structured["isError"] is False
# The transform's return value is exactly what the tool yields.
assert output == {"kept": "routed:list_files"}
@pytest.mark.asyncio
async def test_result_transform_can_rewrite_the_tool_output() -> None:
server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
def transform(_name: str, structured: Any) -> Any:
# Keep only a truncated view of the text field.
return structured["content"][0]["text"][:6]
tools: list[Tool] = await _register_server_tools(
_config("files_main", ["list_files"]), server, result_transform=transform
)
output = await tools[0].on_invoke_tool(None, "{}") # type: ignore[union-attr]
assert output == "routed"
@pytest.mark.asyncio
async def test_without_result_transform_output_is_unchanged() -> None:
server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
tools: list[Tool] = await _register_server_tools(
_config("files_main", ["list_files"]), server, result_transform=None
)
output = await tools[0].on_invoke_tool(None, "{}") # type: ignore[union-attr]
# Same shape the SDK produces today: no transform in the path.
assert server.calls == [("list_files", {})]
assert output == {"type": "text", "text": "routed:list_files"}
# --- error status capture ----------------------------------------------------
class ErroringMCPServer(FakeMCPServer):
"""A connected server whose calls come back as MCP errors (isError=True)."""
async def call_tool(
self,
tool_name: str,
arguments: dict[str, Any] | None,
meta: dict[str, Any] | None = None,
) -> CallToolResult:
self.calls.append((tool_name, arguments))
return CallToolResult(
content=[TextContent(type="text", text=f"boom:{tool_name}")],
isError=True,
)
@pytest.mark.asyncio
async def test_errored_mcp_result_is_flagged_failed_for_the_tui() -> None:
server = ErroringMCPServer("files_main", [_mcp_tool("list_files")])
tools: list[Tool] = await _register_server_tools(
_config("files_main", ["list_files"]), server
)
output = await tools[0].on_invoke_tool(None, "{}") # type: ignore[union-attr]
# The error text stays exactly what the agent gets today; a success:False tag
# rides alongside it purely so the TUI can tell the call apart from a success.
assert output == {"type": "text", "text": "boom:list_files", "success": False}
assert _tool_status_from_result(output) == "failed"
@pytest.mark.asyncio
async def test_successful_mcp_result_stays_completed_for_the_tui() -> None:
server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
tools: list[Tool] = await _register_server_tools(
_config("files_main", ["list_files"]), server
)
output = await tools[0].on_invoke_tool(None, "{}") # type: ignore[union-attr]
# A non-error result is untouched and keeps rendering as done.
assert output == {"type": "text", "text": "routed:list_files"}
assert _tool_status_from_result(output) == "completed"
# --- server build branch -----------------------------------------------------
def test_build_server_stdio_branch() -> None:
config = McpConnectionConfig(
name="local_fs",
transport="stdio",
command="my-server",
args=["--flag", "value"],
env={"TOKEN": "x"},
)
server = _build_server(config)
# Built, not connected: no subprocess is launched here.
assert isinstance(server, MCPServerStdio)
assert server.name == "local_fs"
assert server.params.command == "my-server"
assert server.params.args == ["--flag", "value"]
assert server.params.env == {"TOKEN": "x"}
def test_build_server_http_branch() -> None:
server = _build_server(_config("files_main", ["list_files"]))
assert isinstance(server, MCPServerStreamableHttp)
assert server.name == "files_main"
# --- loader ------------------------------------------------------------------
def test_loader_parses_stdio_and_http_entries(tmp_path: Path) -> None:
config_file = tmp_path / "mcp-servers.json"
config_file.write_text(
json.dumps(
[
{
"name": "local_fs",
"transport": "stdio",
"command": "npx",
"args": ["-y", "server-filesystem"],
},
{
"name": "files_main",
"transport": "http",
"url": "https://mcp.example.com",
"auth": {"kind": "bearer", "token": "abc"},
"allowed_tools": ["list_files"],
},
]
),
encoding="utf-8",
)
configs = load_user_mcp_configs(config_file)
assert [c.name for c in configs] == ["local_fs", "files_main"]
assert configs[0].transport == "stdio"
assert configs[1].allowed_tools == ["list_files"]
def test_loader_skips_bad_entry_but_keeps_good_ones(tmp_path: Path) -> None:
config_file = tmp_path / "mcp-servers.json"
config_file.write_text(
json.dumps(
[
{"name": "broken", "transport": "http"}, # missing url
{
"name": "local_fs",
"transport": "stdio",
"command": "npx",
},
]
),
encoding="utf-8",
)
configs = load_user_mcp_configs(config_file)
assert [c.name for c in configs] == ["local_fs"]
def test_loader_returns_empty_when_file_absent(tmp_path: Path) -> None:
assert load_user_mcp_configs(tmp_path / "does-not-exist.json") == []
def test_loader_reads_env_var_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_file = tmp_path / "from-env.json"
config_file.write_text(
json.dumps([{"name": "local_fs", "transport": "stdio", "command": "npx"}]),
encoding="utf-8",
)
monkeypatch.setenv("STRIX_MCP_CONFIG", str(config_file))
configs = load_user_mcp_configs()
assert [c.name for c in configs] == ["local_fs"]
# --- connection notes --------------------------------------------------------
@pytest.mark.asyncio
async def test_connection_notes_are_carried_on_the_connection(
monkeypatch: pytest.MonkeyPatch,
) -> None:
server = FakeMCPServer("db", [_mcp_tool("query")])
monkeypatch.setattr(mcp_client, "_build_server", lambda _config: server)
config = McpConnectionConfig(
name="db",
url="https://mcp.example.com",
notes="Staging analytics DB; read-only.",
allowed_tools=["query"],
)
connections = await mcp_client.connect_mcp_servers([config])
# Notes ride on the connection (surfaced once), not stapled onto each tool.
assert connections[0].notes == "Staging analytics DB; read-only."
def test_connection_notes_block_lists_only_noted_connections() -> None:
connections = [
ConnectedMcpServer(
server=FakeMCPServer("db", []), name="db", tool_count=2, notes="staging, read-only"
),
ConnectedMcpServer(server=FakeMCPServer("fs", []), name="fs", tool_count=1, notes=None),
]
block = _mcp_connection_notes(connections)
assert block is not None
assert "db" in block
assert "staging, read-only" in block
# A connection without notes is not listed.
assert "fs" not in block
def test_connection_notes_block_is_none_without_notes() -> None:
connections = [
ConnectedMcpServer(server=FakeMCPServer("db", []), name="db", tool_count=1, notes=None)
]
assert _mcp_connection_notes(connections) is None
# --- cancellation cleanup ----------------------------------------------------
@pytest.mark.asyncio
async def test_connect_cleans_up_when_cancelled_mid_connect(
monkeypatch: pytest.MonkeyPatch,
) -> None:
cleaned: list[str] = []
class _Tracking(FakeMCPServer):
def __init__(self, name: str, *, fail_connect: bool = False) -> None:
super().__init__(name, [_mcp_tool("t")])
self._fail_connect = fail_connect
async def connect(self) -> None:
if self._fail_connect:
raise asyncio.CancelledError
async def cleanup(self) -> None:
cleaned.append(self._name)
servers = {"good": _Tracking("good"), "bad": _Tracking("bad", fail_connect=True)}
monkeypatch.setattr(mcp_client, "_build_server", lambda config: servers[config.name])
configs = [
McpConnectionConfig(name="good", url="https://mcp.example.com", allowed_tools=["t"]),
McpConnectionConfig(name="bad", url="https://mcp.example.com", allowed_tools=["t"]),
]
with pytest.raises(asyncio.CancelledError):
await mcp_client.connect_mcp_servers(configs)
# The server being connected when cancelled, and the one already connected,
# are both cleaned up rather than orphaned.
assert cleaned == ["bad", "good"]
# --- duplicate names and run selection ---------------------------------------
def _names_file(tmp_path: Path, *names: str) -> Path:
config_file = tmp_path / "mcp-servers.json"
config_file.write_text(
json.dumps([{"name": n, "transport": "stdio", "command": "npx"} for n in names]),
encoding="utf-8",
)
return config_file
def test_loader_drops_duplicate_named_connections(tmp_path: Path) -> None:
config_file = tmp_path / "mcp-servers.json"
config_file.write_text(
json.dumps(
[
{"name": "dup", "transport": "stdio", "command": "first"},
{"name": "dup", "transport": "stdio", "command": "second"},
{"name": "other", "transport": "stdio", "command": "npx"},
]
),
encoding="utf-8",
)
configs = load_user_mcp_configs(config_file)
# Duplicate name is dropped; the first entry wins.
assert [c.name for c in configs] == ["dup", "other"]
assert configs[0].command == "first"
def test_loader_include_selection_keeps_only_named(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
config_file = _names_file(tmp_path, "a", "b", "c")
monkeypatch.setenv("STRIX_MCP_ONLY", "a,c")
configs = load_user_mcp_configs(config_file)
assert [c.name for c in configs] == ["a", "c"]
def test_loader_exclude_selection_drops_named(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
config_file = _names_file(tmp_path, "a", "b", "c")
monkeypatch.setenv("STRIX_MCP_EXCLUDE", "b")
configs = load_user_mcp_configs(config_file)
assert [c.name for c in configs] == ["a", "c"]
+22
View File
@@ -9,6 +9,7 @@ from strix.config.models import (
RECOMMENDED_MODEL_NAMES,
is_recommended_or_frontier_model,
request_timeout_extra_args,
supports_strict_tool_schemas,
)
@@ -90,3 +91,24 @@ def test_frontier_model_families_are_accepted(model_name: str) -> None:
)
def test_non_frontier_models_are_rejected(model_name: str) -> None:
assert not is_recommended_or_frontier_model(model_name)
@pytest.mark.parametrize(
"model_name",
[
"anthropic/claude-sonnet-4-6",
"bedrock/anthropic.claude-opus-4-8-v1:0",
"vertex_ai/claude-sonnet-5",
"Sonnet-5",
],
)
def test_claude_routes_reject_strict_tool_schemas(model_name: str) -> None:
assert not supports_strict_tool_schemas(model_name)
@pytest.mark.parametrize(
"model_name",
["openai/gpt-5.4", "gpt-5.4", "gemini/gemini-3.1-pro-preview", "deepseek/deepseek-v4"],
)
def test_other_routes_keep_strict_tool_schemas(model_name: str) -> None:
assert supports_strict_tool_schemas(model_name)