Harden agent-native MCP proof of concept

pyproject.toml

@@ -60,6 +60,7 @@ dev = [
   "pyinstaller>=6.17.0; python_version >= '3.12' and python_version < '3.15'",
   "pytest>=8.3",
   "pytest-asyncio>=0.24",
+  "types-pygments>=2.20.0.20260518",
 ]
 
 [tool.pytest.ini_options]
@@ -118,6 +119,16 @@ disable_error_code = ["import-untyped"]
 [[tool.mypy.overrides]]
 module = ["tests.*"]
 disallow_untyped_decorators = false
+# Tests intentionally monkeypatch imported module collaborators that are not
+# part of those modules' public export surface.
+disable_error_code = ["attr-defined"]
+
+# Textual's runtime-defined widget/event APIs do not currently match its strict
+# type surface. Keep the legacy TUI isolated until it receives a dedicated
+# typing migration; all other Strix modules remain under strict checking.
+[[tool.mypy.overrides]]
+module = ["strix.interface.tui.app"]
+ignore_errors = true
 
 # ============================================================================
 # Ruff Configuration (Fast Python Linter & Formatter)

strix/core/hooks.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import math
 from typing import TYPE_CHECKING, Any
 
 from agents.lifecycle import RunHooks
@@ -27,8 +28,9 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
     """Persist SDK-native usage after every model response."""
 
     def __init__(self, *, model: str, max_budget_usd: float | None = None) -> None:
-        import math
-        if max_budget_usd is not None and (not math.isfinite(max_budget_usd) or max_budget_usd <= 0):
+        if max_budget_usd is not None and (
+            not math.isfinite(max_budget_usd) or max_budget_usd <= 0
+        ):
             raise ValueError("max_budget_usd must be a finite number greater than 0")
         self._model = model
         self._max_budget_usd = max_budget_usd

strix/interface/main.py

@@ -310,6 +310,7 @@ def _positive_budget(value: str) -> float:
     except ValueError as exc:
         raise argparse.ArgumentTypeError(f"invalid float value: {value!r}") from exc
     import math
+
     if not math.isfinite(budget) or budget <= 0:
         raise argparse.ArgumentTypeError("must be a finite number greater than 0")
     return budget

strix/mcp/server.py

@@ -2,8 +2,8 @@
 
 from __future__ import annotations
 
-import os
 import logging
+import os
 from contextlib import asynccontextmanager
 from typing import Any
 
@@ -15,10 +15,20 @@ from mcp.shared.message import SessionMessage
 from strix.mcp.service import StrixMCPService
 from strix.tools.proxy.tools import (
     list_requests as _list_requests,
+)
+from strix.tools.proxy.tools import (
     list_sitemap as _list_sitemap,
+)
+from strix.tools.proxy.tools import (
     repeat_request as _repeat_request,
+)
+from strix.tools.proxy.tools import (
     scope_rules as _scope_rules,
+)
+from strix.tools.proxy.tools import (
     view_request as _view_request,
+)
+from strix.tools.proxy.tools import (
     view_sitemap_entry as _view_sitemap_entry,
 )
 
@@ -263,8 +273,8 @@ async def _event_loop_stdio() -> Any:
     and writing the pipe file descriptors through the event loop keeps the MCP
     transport portable while preserving newline-delimited JSON-RPC semantics.
     """
-    read_writer, read_stream = anyio.create_memory_object_stream(0)
-    write_stream, write_reader = anyio.create_memory_object_stream(0)
+    read_writer, read_stream = anyio.create_memory_object_stream[SessionMessage | Exception](0)
+    write_stream, write_reader = anyio.create_memory_object_stream[SessionMessage](0)
 
     async def read_stdin() -> None:
         buffered = b""
@@ -290,8 +300,7 @@ async def _event_loop_stdio() -> Any:
         async with write_reader:
             async for session_message in write_reader:
                 payload = (
-                    session_message.message.model_dump_json(by_alias=True, exclude_none=True)
-                    + "\n"
+                    session_message.message.model_dump_json(by_alias=True, exclude_none=True) + "\n"
                 ).encode()
                 while payload:
                     await anyio.wait_writable(1)
@@ -305,10 +314,10 @@ async def _event_loop_stdio() -> Any:
 
 async def _run_stdio() -> None:
     async with _event_loop_stdio() as (read_stream, write_stream):
-        await mcp._mcp_server.run(  # noqa: SLF001 - mirrors FastMCP.run_stdio_async.
+        await mcp._mcp_server.run(
             read_stream,
             write_stream,
-            mcp._mcp_server.create_initialization_options(),  # noqa: SLF001
+            mcp._mcp_server.create_initialization_options(),
         )
 
 

strix/mcp/service.py

@@ -49,7 +49,7 @@ class StrixMCPService:
         self.report_state: ReportState | None = None
         self.bundle: dict[str, Any] | None = None
 
-    async def start_scan(
+    async def start_scan(  # noqa: PLR0911 - lifecycle validation has distinct failure results.
         self,
         targets: list[str],
         *,
@@ -135,7 +135,7 @@ class StrixMCPService:
                 image=load_settings().runtime.image,
                 local_sources=local_sources,
             )
-        except Exception as exc:  # noqa: BLE001 - return a clean MCP error.
+        except Exception as exc:
             state.save_run_data(status="failed")
             logger.exception("Could not create Strix sandbox")
             return {"success": False, "error": f"Could not create Strix sandbox: {exc}"}
@@ -182,12 +182,12 @@ class StrixMCPService:
         max_mb = load_settings().runtime.max_local_copy_mb
         oversized = find_oversized_local_targets(targets_info, max_mb * 1024 * 1024)
         if oversized:
-            details = "; ".join(
+            oversized_details = "; ".join(
                 f"{path} ({size / (1024 * 1024):.0f} MB)" for path, size in oversized
             )
             raise ValueError(
-                f"Local target too large to copy: {details}. Use the mounts argument or raise "
-                "STRIX_MAX_LOCAL_COPY_MB."
+                f"Local target too large to copy: {oversized_details}. "
+                "Use the mounts argument or raise STRIX_MAX_LOCAL_COPY_MB."
             )
 
         scan_name = run_name or generate_run_name(targets_info)
@@ -221,7 +221,7 @@ class StrixMCPService:
             str(destination),
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
-        )  # noqa: S603 - argv is fixed apart from the authorized URL/path.
+        )
         _stdout, stderr = await process.communicate()
         if process.returncode != 0:
             detail = stderr.decode("utf-8", errors="replace")[-2000:]
@@ -283,7 +283,7 @@ class StrixMCPService:
         try:
             raw = await tool.on_invoke_tool(context, json.dumps(arguments))
             return json.loads(raw) if isinstance(raw, str) else {"success": True, "result": raw}
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             logger.exception("Proxy tool call failed")
             return {"success": False, "error": str(exc)}
 

strix/runtime/docker_client.py

@@ -48,7 +48,8 @@ logger = logging.getLogger(__name__)
 class StrixDockerSandboxClient(DockerSandboxClient):
     # Host directories to bind-mount into the container, set by the docker
     # backend before ``create()``. Each item is ``{source, target, read_only}``.
-    strix_bind_mounts: list[dict[str, Any]] = []  # overridden per-instance in backends.py
+    # The backend replaces this on every instance before ``create``.
+    strix_bind_mounts: list[dict[str, Any]] = []  # noqa: RUF012
 
     async def _create_container(
         self,

tests/test_agent_launcher.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from argparse import Namespace
-from pathlib import Path
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
 
@@ -49,9 +48,7 @@ def test_explicit_legacy_mode_does_not_launch_agent(monkeypatch: pytest.MonkeyPa
     assert agent_launcher.uses_coding_agent(_args(agent="legacy")) is False
 
 
-def test_codex_command_embeds_ephemeral_mcp_config(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(Path, "cwd", classmethod(lambda cls: Path("/tmp/project")))
-
+def test_codex_command_embeds_ephemeral_mcp_config() -> None:
     command = agent_launcher.build_agent_command(_args(non_interactive=True), "codex")
 
     assert command[:2] == ["codex", "exec"]

uv.lock

@@ -2233,6 +2233,7 @@ dev = [
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "ruff" },
+    { name = "types-pygments" },
 ]
 
 [package.metadata]
@@ -2259,6 +2260,7 @@ dev = [
     { name = "pytest", specifier = ">=8.3" },
     { name = "pytest-asyncio", specifier = ">=0.24" },
     { name = "ruff", specifier = ">=0.11.13" },
+    { name = "types-pygments", specifier = ">=2.20.0.20260518" },
 ]
 
 [[package]]
@@ -2379,6 +2381,27 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/3f/f9/2b3ff4e56e5fa7debfaf9eb135d0da96f3e9a1d5b27222223c7296336e5f/typer-0.25.1-py3-none-any.whl", hash = "sha256:75caa44ed46a03fb2dab8808753ffacdbfea88495e74c85a28c5eefcf5f39c89", size = 58409, upload-time = "2026-04-30T19:32:18.271Z" },
 ]
 
+[[package]]
+name = "types-docutils"
+version = "0.22.3.20260518"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/fd/91/81b520d0869a41aa0d86e713417b63e8604b4280c304bc09b02d74c7efd4/types_docutils-0.22.3.20260518.tar.gz", hash = "sha256:2c45ba63a9ac64246335359b68fe9c27602926499c9b67caec3780745f6aadee", size = 57504, upload-time = "2026-05-18T06:03:53.325Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/01/9b/243fb84ede4f987f303a89e734c5def35ead2f8bd962ad301c1e99680958/types_docutils-0.22.3.20260518-py3-none-any.whl", hash = "sha256:9c4cbc37d9e37f47dfe971ac09e9880380dc948ff1f23956892e810d0bf08676", size = 91970, upload-time = "2026-05-18T06:03:52.388Z" },
+]
+
+[[package]]
+name = "types-pygments"
+version = "2.20.0.20260518"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "types-docutils" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/db/66/3e27e8dbe72947d51355d1fcba222a55bf5f0770ee660e9cc9df0d1ce5d7/types_pygments-2.20.0.20260518.tar.gz", hash = "sha256:bcab233d0389cb0a91146eb860e7bdcbceaa0f30bee73cefc7b367129cc4a330", size = 21152, upload-time = "2026-05-18T06:07:26.927Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/66/d6/5f19f5b633af6a7666c6096fdd06b70f0d4a8f585af5e18a3ef58ad47ec1/types_pygments-2.20.0.20260518-py3-none-any.whl", hash = "sha256:e40728efd00c9da5936366648d5b3c55e72ed52bdd80495a96577b4d717c4fcb", size = 29002, upload-time = "2026-05-18T06:07:26.054Z" },
+]
+
 [[package]]
 name = "types-requests"
 version = "2.33.0.20260518"