Simplify Python proxy automation

containers/Dockerfile

@@ -78,13 +78,6 @@ RUN pipx install arjun && \
     pipx inject dirsearch setuptools && \
     pipx install wafw00f
 
-# ``python_action`` ships a fresh ``caido-sdk-client`` per call into /tmp.
-# Install it system-wide so the driver's ``import caido_sdk_client`` resolves
-# without venv activation. The helper *logic* lives host-side in
-# ``strix/tools/proxy/_calls.py`` (shipped at runtime) — only the SDK dep
-# is image-baked.
-RUN pip install --break-system-packages --no-cache-dir caido-sdk-client
-
 ENV NPM_CONFIG_PREFIX=/home/pentester/.npm-global
 RUN mkdir -p /home/pentester/.npm-global
 
@@ -198,9 +191,13 @@ RUN mkdir -p /workspace && chown -R pentester:pentester /workspace /app
 
 USER pentester
 RUN python3 -m venv /app/.venv && \
+    /app/.venv/bin/pip install --no-cache-dir caido-sdk-client && \
     /app/.venv/bin/pip install --no-cache-dir -r /home/pentester/tools/jwt_tool/requirements.txt && \
     ln -s /home/pentester/tools/jwt_tool/jwt_tool.py /home/pentester/.local/bin/jwt_tool
 
+COPY --chown=pentester:pentester strix/tools/proxy/caido_api.py /opt/strix-python/caido_api.py
+ENV PYTHONPATH=/opt/strix-python
+
 RUN echo 'export PATH="/home/pentester/go/bin:/home/pentester/.local/bin:/home/pentester/.npm-global/bin:$PATH"' >> /home/pentester/.bashrc && \
     echo 'export PATH="/home/pentester/go/bin:/home/pentester/.local/bin:/home/pentester/.npm-global/bin:$PATH"' >> /home/pentester/.profile
 

docs/tools/proxy.mdx

@@ -30,23 +30,33 @@ The agent can take any captured request and replay it with modifications:
 
 ## Python Integration
 
-All proxy functions are automatically available in Python sessions. This enables powerful scripted security testing:
+Proxy helpers are available to sandbox Python scripts through the image-baked `caido_api` module. This enables powerful scripted security testing:
 
 ```python
-# List recent POST requests
+import asyncio
-post_requests = list_requests(
-    httpql_filter='req.method.eq:"POST"',
-    page_size=20
-)
 
-# View a specific request
+from caido_api import list_requests, repeat_request, view_request
-request_details = view_request("req_123", part="request")
 
-# Replay with modified payload
+
-response = repeat_request("req_123", {
+async def main():
-    "body": '{"user_id": "admin"}'
+    # List recent POST requests
-})
+    post_requests = await list_requests(
-print(f"Status: {response['status_code']}")
+        httpql_filter='req.method.eq:"POST"',
+        first=20,
+    )
+
+    # View a specific request
+    request_details = await view_request("req_123", part="request")
+
+    # Replay with modified payload
+    response = await repeat_request(
+        "req_123",
+        modifications={"body": '{"user_id": "admin"}'},
+    )
+    print(response["status"], request_details is not None, len(post_requests.edges))
+
+
+asyncio.run(main())
 ```
 
 ### Available Functions
@@ -58,26 +68,34 @@ print(f"Status: {response['status_code']}")
 | `repeat_request()`     | Replay a request with modifications        |
 | `send_request()`       | Send a new HTTP request                    |
 | `scope_rules()`        | Manage proxy scope (allowlist/denylist)    |
-| `list_sitemap()`       | View discovered endpoints                  |
-| `view_sitemap_entry()` | Get details for a sitemap entry            |
 
 ### Example: Automated IDOR Testing
 
 ```python
+import asyncio
+
 # Get all requests to user endpoints
-user_requests = list_requests(
+from caido_api import list_requests, repeat_request
-    httpql_filter='req.path.cont:"/users/"'
-)
 
-for req in user_requests.get('requests', []):
-    # Try accessing with different user IDs
-    for test_id in ['1', '2', 'admin', '../admin']:
-        response = repeat_request(req['id'], {
-            'url': req['path'].replace('/users/1', f'/users/{test_id}')
-        })
 
-        if response['status_code'] == 200:
+async def main():
-            print(f"Potential IDOR: {test_id} returned 200")
+    user_requests = await list_requests(httpql_filter='req.path.cont:"/users/"')
+
+    for edge in user_requests.edges:
+        req = edge.node.request
+        scheme = "https" if req.is_tls else "http"
+        for test_id in ["1", "2", "admin", "../admin"]:
+            url = f"{scheme}://{req.host}{req.path.replace('/users/1', f'/users/{test_id}')}"
+            response = await repeat_request(
+                req.id,
+                modifications={"url": url},
+            )
+            print(req.id, test_id, response["status"])
+            if response["status"] == "DONE":
+                print(f"Replay completed for candidate {test_id}")
+
+
+asyncio.run(main())
 ```
 
 ## Human-in-the-Loop

pyproject.toml

@@ -210,9 +210,6 @@ ignore = [
 "strix/tools/thinking/tool.py" = ["TC002"]
 "strix/tools/web_search/tool.py" = ["TC002"]
 "strix/tools/proxy/tools.py" = ["TC002", "PLR0911"]
-# Driver script writes to /tmp inside the sandbox container — single-user,
-# isolated rootfs, so the multi-user race S108 warns about doesn't apply.
-"strix/tools/python/tool.py" = ["S108", "TC002"]
 "strix/tools/agents_graph/tools.py" = ["TC002"]
 "strix/agents/factory.py" = ["TC002"]
 # Entry point: ``Path`` is used at runtime by the typing of the

strix/agents/factory.py

@@ -51,7 +51,6 @@ from strix.tools.proxy.tools import (
     send_request,
     view_request,
 )
-from strix.tools.python.tool import python_action
 from strix.tools.reporting.tool import create_vulnerability_report
 from strix.tools.thinking.tool import think
 from strix.tools.todo.tools import (
@@ -272,8 +271,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
     send_request,
     repeat_request,
     scope_rules,
-    # Stateless Python execution with proxy helpers pre-bound
-    python_action,
     # Multi-agent graph tools (the coordinator is in ctx.context)
     view_agent_graph,
     send_message_to_agent,

strix/agents/prompt.py

@@ -37,8 +37,8 @@ def _resolve_skills(
     2. ``scan_modes/<mode>`` (always).
     3. ``tooling/agent_browser`` (always — every agent has shell + the
        agent-browser CLI).
-    4. ``tooling/python`` (always — every agent has the ``python_action``
+    4. ``tooling/python`` (always — Python runs through ``exec_command``;
-       tool with proxy helpers pre-bound).
+       sandbox scripts can import ``caido_api`` for Caido automation).
     5. ``coordination/root_agent`` for the root agent only — orchestration
        guidance for delegating to specialist subagents.
     6. Whitebox-specific skills if applicable.

strix/agents/prompts/system_prompt.jinja

@@ -161,14 +161,19 @@ OPERATIONAL PRINCIPLES:
 EFFICIENCY TACTICS:
 - Automate with Python scripts for complex workflows and repetitive inputs/tasks
 - Batch similar operations together
-- Use captured traffic from proxy in Python tool to automate analysis
+- Use captured traffic from the proxy tools directly, or import `caido_api`
+  from sandbox Python scripts when proxy automation is easier in code
 - Download additional tools as needed for specific tasks
 - Run multiple scans in parallel when possible
 - Load the most relevant skill before starting a specialized testing workflow if doing so will improve accuracy, speed, or tool usage
-- Prefer the python tool for Python code. Do NOT embed Python in terminal commands via heredocs, here-strings, python -c, or interactive REPL driving unless shell-only behavior is specifically required
+- Use `exec_command` for Python code: write reusable scripts under
-- The python tool exists to give you persistent interpreter state, structured code execution, cleaner debugging, and easier multi-step automation than terminal-wrapped Python
+  `/workspace/scratch/` and run them with `python3`. For one-off snippets,
+  `python3 -c` or a here-document is acceptable.
+- For Caido proxy automation inside Python, explicitly import from
+  `caido_api`:
+  `from caido_api import list_requests, view_request, send_request, repeat_request, scope_rules`
 - Prefer established fuzzers/scanners where applicable: ffuf, sqlmap, zaproxy, nuclei, wapiti, arjun, httpx, katana, semgrep, bandit, trufflehog, nmap. Use scripts mainly to coordinate or validate around them, not to replace them without reason
-- For trial-heavy vectors (SQLi, XSS, XXE, SSRF, RCE, auth/JWT, deserialization), DO NOT iterate payloads manually in the browser. Always spray payloads via the python or terminal tools
+- For trial-heavy vectors (SQLi, XSS, XXE, SSRF, RCE, auth/JWT, deserialization), DO NOT iterate payloads manually in the browser. Always spray payloads via Python scripts through `exec_command` or terminal tools.
 - When using established fuzzers/scanners, use the proxy for inspection where helpful
 - Generate/adapt large payload corpora: combine encodings (URL, unicode, base64), comment styles, wrappers, time-based/differential probes. Expand with wordlists/templates
 - Use the web_search tool to fetch and refresh payload sets (latest bypasses, WAF evasions, DB-specific syntax, browser/JS quirks) and incorporate them into sprays
@@ -404,7 +409,8 @@ SPECIALIZED TOOLS:
 - interactsh-client - OOB interaction testing
 
 PROXY & INTERCEPTION:
-- Caido CLI - Modern web proxy (already running). Used with proxy tool or with python tool (functions already imported).
+- Caido CLI - Modern web proxy (already running). Use the proxy tools
+  directly, or import `caido_api` from sandbox Python scripts.
 - NOTE: If you are seeing proxy errors when sending requests, it usually means you are not sending requests to a correct url/host/port.
 - Ignore Caido proxy-generated 50x HTML error pages; these are proxy issues (might happen when requesting a wrong host or SSL/TLS issues, etc).
 

strix/interface/assets/tui_styles.tcss

@@ -386,7 +386,6 @@ VulnerabilityDetailScreen {
 
 .browser-tool,
 .terminal-tool,
-.python-tool,
 .agents-graph-tool,
 .file-edit-tool,
 .proxy-tool,
@@ -411,8 +410,6 @@ VulnerabilityDetailScreen {
 .browser-tool.status-running,
 .terminal-tool.status-completed,
 .terminal-tool.status-running,
-.python-tool.status-completed,
-.python-tool.status-running,
 .agents-graph-tool.status-completed,
 .agents-graph-tool.status-running,
 .file-edit-tool.status-completed,

strix/skills/scan_modes/standard.md

@@ -77,7 +77,7 @@ Test each attack surface methodically. Spawn focused subagents for different are
 - Demonstrate actual impact, not theoretical risk
 - Chain vulnerabilities to show maximum severity
 - Document full attack path from entry to impact
-- Use python tool for complex exploit development
+- Use Python scripts through `exec_command` for complex exploit development
 
 ## Phase 5: Reporting
 

strix/skills/tooling/python.md

@@ -1,113 +1,74 @@
 ---
 name: python
-description: python_action — execute Python in the sandbox with Caido proxy helpers (list_requests, view_request, send_request, repeat_request, scope_rules) pre-bound as awaitables. Stateless per call; persistence via files.
+description: Run Python through exec_command in the SDK sandbox. Use the image-baked caido_api module for Caido proxy automation from Python scripts.
 ---
 
+# Python In The Sandbox
 
-# python_action — when and how
+Use `exec_command` for Python. There is no separate Strix Python executor.
 
-Use ``python_action`` for any Python-side work: payload encoding/decoding,
+Prefer writing reusable scripts to `/workspace/scratch/<name>.py` and
-parsing/transforming captured HTTP traffic, crypto operations, custom
+running them with `python3 /workspace/scratch/<name>.py`. For short
-exploit scripts, log/JSON analysis. Use ``exec_command`` for shell tools
+one-off transformations, `python3 -c` or a small here-document is fine.
-(nmap, sqlmap, ffuf, agent-browser, package managers, daemons).
 
-**Do not** wrap Python in bash heredocs, ``python3 -c`` one-liners, or
+## Proxy Automation From Python
-``echo | python3`` chains via ``exec_command`` — ``python_action`` exists
-so structured output replaces fragile stdout parsing.
 
-## What's pre-bound (no imports needed)
+The sandbox image includes an installed `caido_api` module. Import it
-
+explicitly when Python code needs Caido traffic or replay access:
-All proxy helpers are **async** — call them with ``await``:
-
-- ``list_requests(httpql_filter=, first=50, after=, sort_by=, sort_order=,
-  scope_id=)`` → cursor-paginated SDK ``Connection``. Iterate
-  ``connection.edges``; each edge has ``.cursor`` and ``.node.request`` /
-  ``.node.response``.
-- ``view_request(request_id, part="request")`` → SDK request object.
-  ``.request.raw`` and ``.response.raw`` are bytes.
-- ``send_request(method, url, headers=None, body="")`` → dict with
-  ``status``, ``error``, ``elapsed_ms``, ``response_raw`` (bytes or None),
-  ``session_id``.
-- ``repeat_request(request_id, modifications={...})`` → same shape.
-  ``modifications`` keys: ``url`` / ``params`` / ``headers`` / ``body`` /
-  ``cookies``.
-- ``scope_rules(action, allowlist=, denylist=, scope_id=, scope_name=)``
-  — same actions as the host-side tool (``list``/``get``/``create``/
-  ``update``/``delete``).
-
-Top-level ``await`` works — the body is wrapped in an async function for
-you. ``print()`` to emit visible output; the last expression is **not**
-auto-shown.
-
-## Stateless model + how to keep state
-
-Each ``python_action`` call is a **fresh process**: variables, imports,
-and definitions do not survive. To carry state across steps:
-
-- **Combine into one call** when the workflow is short — write the full
-  multi-step routine as one ``code`` block.
-- **Persist to disk** for longer-lived state. ``/workspace/scratch/`` is
-  pentester-writable and survives across calls within a scan.
-- **Build a script** with ``apply_patch`` to ``/workspace/scratch/<name>.py``
-  and run it via ``exec_command python3 ...`` when you need a file the
-  agent can iterate on.
-
-## Examples
-
-### Hunt SQLi candidates by inspecting captured traffic
 
 ```python
-# All POSTs that look interesting
+from caido_api import (
-posts = await list_requests(
+    list_requests,
-    httpql_filter='req.method.eq:"POST" AND req.path.cont:"/api/"',
+    repeat_request,
-    first=50,
+    scope_rules,
+    send_request,
+    view_request,
 )
-candidates = []
-for edge in posts.edges:
-    body = await view_request(edge.node.request.id, part="request")
-    raw = body.request.raw.decode("utf-8", errors="replace")
-    if "id=" in raw or "user=" in raw:
-        candidates.append(edge.node.request.id)
-
-print(f"{len(candidates)} candidates")
-print(candidates[:10])
 ```
 
-### Replay with a SQLi probe and a tampered cookie
+All helpers are async. Use them inside `asyncio.run(...)` or an async
+function:
 
 ```python
-result = await repeat_request(
+import asyncio
-    "req_abc123",
+
-    modifications={
+from caido_api import list_requests, view_request
-        "params": {"id": "1' OR '1'='1"},
+
-        "cookies": {"session": "ATTACKER_TOKEN"},
+
-    },
+async def main():
-)
+    posts = await list_requests(
-print(result["status"], result["elapsed_ms"], "ms")
+        httpql_filter='req.method.eq:"POST" AND req.path.cont:"/api/"',
-if result["response_raw"]:
+        first=50,
-    print(result["response_raw"].decode("utf-8", errors="replace")[:500])
+    )
+    candidates = []
+    for edge in posts.edges:
+        request_id = edge.node.request.id
+        body = await view_request(request_id, part="request")
+        raw = body.request.raw.decode("utf-8", errors="replace")
+        if "id=" in raw or "user=" in raw:
+            candidates.append(request_id)
+
+    print(f"{len(candidates)} candidates")
+    print(candidates[:10])
+
+
+asyncio.run(main())
 ```
 
-### Decode/encode payloads
+Available helpers:
 
-```python
+- `list_requests(httpql_filter=, first=50, after=, sort_by=, sort_order=, scope_id=)` returns a cursor-paginated Caido SDK `Connection`.
-import base64, urllib.parse, hashlib
+- `view_request(request_id, part="request")` returns a Caido SDK request object with raw request/response bytes.
+- `send_request(method, url, headers=None, body="")` sends an arbitrary raw request through Caido Replay.
+- `repeat_request(request_id, modifications={...})` replays a captured request after modifying `url`, `params`, `headers`, `body`, or `cookies`.
+- `scope_rules(action, allowlist=, denylist=, scope_id=, scope_name=)` manages Caido scopes.
 
-token = "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyIjoiYWxpY2UifQ.sig"
+## Workflow
-header_b64, payload_b64, _ = token.split(".")
-print(base64.urlsafe_b64decode(payload_b64 + "=="))
-```
 
-### Iterate an exploit by writing to scratch
+For iterative exploit work, put code in a file:
-
-When iterating, prefer writing the script to disk so you can edit-and-rerun
-without re-sending the whole code each call:
 
 ```text
-# 1. Use apply_patch to create /workspace/scratch/exploit.py
+1. Create or edit `/workspace/scratch/exploit.py` with `apply_patch`.
-# 2. exec_command: python3 /workspace/scratch/exploit.py
+2. Run it with `exec_command`: `python3 /workspace/scratch/exploit.py`.
-# 3. Edit + re-run; repeat until working
+3. Edit and rerun until the proof-of-concept is reliable.
 ```
-
-For one-shot crypto/encoding work or a single proxy-data analysis,
-``python_action`` is the cleaner choice.

strix/tools/proxy/caido_api.py

@@ -1,28 +1,16 @@
-"""Pure caido-sdk-client call sequences shared by ``tools.py`` and ``python_action``.
+"""Shared Caido proxy helpers and sandbox-importable ``caido_api`` module."""
-
-Functions here:
-
-- Take an explicit ``Client`` argument — no module-level state, no
-  context lookups. The caller decides what client to use.
-- Return raw caido-sdk-client objects (or dicts of primitives where the
-  composition itself is the value-add, like :func:`replay_send_raw`).
-- Live without ``@function_tool`` decorators, ``RunContextWrapper``, or
-  any host-side framework dependency. They run identically inside the
-  Strix host process (called from ``tools.py``) and inside the sandbox
-  container (called from the ``python_action`` driver), against the
-  same Caido instance — host gets there via the host-mapped port,
-  container gets there via ``localhost:48080``.
-
-Single source of truth for the Caido SDK call shapes; ``tools.py`` adds
-LLM-specific JSON serialization and error wrapping on top.
-"""
 
 from __future__ import annotations
 
+import asyncio
+import json
+import os
 import time
+import urllib.request
 from typing import TYPE_CHECKING, Any, Literal
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 
+from caido_sdk_client import Client, TokenAuthOptions
 from caido_sdk_client.types import (
     ConnectionInfoInput,
     CreateReplaySessionFromRaw,
@@ -35,7 +23,7 @@ from caido_sdk_client.types import (
 
 
 if TYPE_CHECKING:
-    from caido_sdk_client import Client
+    from caido_sdk_client import Client as CaidoClient
 
 
 RequestPart = Literal["request", "response"]
@@ -50,8 +38,10 @@ SortBy = Literal[
     "source",
 ]
 SortOrder = Literal["asc", "desc"]
+ScopeAction = Literal["get", "list", "create", "update", "delete"]
 
-
+_DEFAULT_CAIDO_URL = "http://127.0.0.1:48080"
+_CLIENT_CACHE: dict[str, Client] = {}
 _REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = {
     "timestamp": ("req", "created_at"),
     "host": ("req", "host"),
@@ -64,11 +54,56 @@ _REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = {
 }
 
 
-# ----------------------------------------------------------------------
+def caido_url() -> str:
-# Requests — list / get
+    """Return the in-sandbox Caido endpoint used by ``caido_api``."""
-# ----------------------------------------------------------------------
+    return os.environ.get("STRIX_CAIDO_URL", _DEFAULT_CAIDO_URL).rstrip("/")
-async def list_requests(
+
-    client: Client,
+
+def _graphql_url() -> str:
+    base_url = caido_url()
+    parsed = urlparse(base_url)
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        raise ValueError(f"Invalid Caido URL: {base_url}")
+    return f"{base_url}/graphql"
+
+
+def _login_as_guest() -> str:
+    body = json.dumps({"query": "mutation { loginAsGuest { token { accessToken } } }"}).encode(
+        "utf-8"
+    )
+    req = urllib.request.Request(  # noqa: S310
+        _graphql_url(),
+        data=body,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310  # nosec B310
+        payload = json.loads(resp.read())
+    return str(payload["data"]["loginAsGuest"]["token"]["accessToken"])
+
+
+async def get_client() -> Client:
+    """Return a connected Caido SDK client for the local sandbox sidecar."""
+    if client := _CLIENT_CACHE.get("default"):
+        return client
+
+    token = await asyncio.to_thread(_login_as_guest)
+    client = Client(caido_url(), auth=TokenAuthOptions(token=token))
+    await client.connect()
+    _CLIENT_CACHE["default"] = client
+    return client
+
+
+async def close_client() -> None:
+    """Close the cached sandbox Caido client, if one was opened."""
+    client = _CLIENT_CACHE.pop("default", None)
+    if client is None:
+        return
+    await client.aclose()
+
+
+async def list_requests_with_client(
+    client: CaidoClient,
     *,
     httpql_filter: str | None = None,
     first: int = 50,
@@ -89,8 +124,8 @@ async def list_requests(
     return await builder.execute()
 
 
-async def get_request(
+async def get_request_with_client(
-    client: Client,
+    client: CaidoClient,
     request_id: str,
     *,
     part: RequestPart = "request",
@@ -102,9 +137,6 @@ async def get_request(
     return await client.request.get(request_id, opts)
 
 
-# ----------------------------------------------------------------------
-# Raw HTTP request build / parse / mutate
-# ----------------------------------------------------------------------
 def build_raw_request(
     *,
     method: str,
@@ -131,7 +163,6 @@ def build_raw_request(
     lines = [f"{method.upper()} {path} HTTP/1.1"]
     lines.extend(f"{k}: {v}" for k, v in final_headers.items())
     raw = ("\r\n".join(lines) + "\r\n\r\n" + body).encode("utf-8")
-
     return ConnectionInfoInput(host=host, port=port, is_tls=is_tls), raw
 
 
@@ -183,13 +214,10 @@ def apply_modifications(
         existing = {k: v[0] if v else "" for k, v in parse_qs(parsed.query).items()}
         existing.update(modifications["params"])
         final_url = urlunparse(parsed._replace(query=urlencode(existing)))
-
     if "headers" in modifications:
         headers.update(modifications["headers"])
-
     if "body" in modifications:
         body = modifications["body"]
-
     if "cookies" in modifications:
         cookies: dict[str, str] = {}
         if headers.get("Cookie"):
@@ -208,11 +236,8 @@ def apply_modifications(
     }
 
 
-# ----------------------------------------------------------------------
-# Replay — send raw bytes, get a result
-# ----------------------------------------------------------------------
 async def replay_send_raw(
-    client: Client,
+    client: CaidoClient,
     *,
     raw: bytes,
     connection: ConnectionInfoInput,
@@ -238,19 +263,16 @@ async def replay_send_raw(
     }
 
 
-# ----------------------------------------------------------------------
+async def scope_list(client: CaidoClient) -> Any:
-# Scope CRUD
-# ----------------------------------------------------------------------
-async def scope_list(client: Client) -> Any:
     return await client.scope.list()
 
 
-async def scope_get(client: Client, scope_id: str) -> Any:
+async def scope_get(client: CaidoClient, scope_id: str) -> Any:
     return await client.scope.get(scope_id)
 
 
 async def scope_create(
-    client: Client,
+    client: CaidoClient,
     *,
     name: str,
     allowlist: list[str] | None = None,
@@ -266,7 +288,7 @@ async def scope_create(
 
 
 async def scope_update(
-    client: Client,
+    client: CaidoClient,
     scope_id: str,
     *,
     name: str,
@@ -283,5 +305,133 @@ async def scope_update(
     )
 
 
-async def scope_delete(client: Client, scope_id: str) -> None:
+async def scope_delete(client: CaidoClient, scope_id: str) -> None:
     await client.scope.delete(scope_id)
+
+
+async def list_requests(
+    *,
+    httpql_filter: str | None = None,
+    first: int = 50,
+    after: str | None = None,
+    sort_by: SortBy = "timestamp",
+    sort_order: SortOrder = "desc",
+    scope_id: str | None = None,
+) -> Any:
+    """List captured HTTP requests from sandbox Python."""
+    return await list_requests_with_client(
+        await get_client(),
+        httpql_filter=httpql_filter,
+        first=first,
+        after=after,
+        sort_by=sort_by,
+        sort_order=sort_order,
+        scope_id=scope_id,
+    )
+
+
+async def view_request(request_id: str, *, part: RequestPart = "request") -> Any:
+    """Return one captured request/response from sandbox Python."""
+    return await get_request_with_client(await get_client(), request_id, part=part)
+
+
+async def send_request(
+    method: str,
+    url: str,
+    *,
+    headers: dict[str, str] | None = None,
+    body: str = "",
+) -> dict[str, Any]:
+    """Send an arbitrary raw HTTP request through Caido Replay."""
+    connection, raw = build_raw_request(
+        method=method,
+        url=url,
+        headers=headers or {},
+        body=body,
+    )
+    return await replay_send_raw(await get_client(), raw=raw, connection=connection)
+
+
+async def repeat_request(
+    request_id: str,
+    *,
+    modifications: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Replay a captured request after applying request modifications."""
+    mods = modifications or {}
+    result = await get_request_with_client(await get_client(), request_id, part="request")
+    if result is None or result.request.raw is None:
+        raise ValueError(f"Request {request_id} not found")
+
+    original = result.request
+    raw_str = result.request.raw.decode("utf-8", errors="replace")
+    components = parse_raw_request(raw_str)
+    full_url = full_url_from_components(original, components, mods)
+    modified = apply_modifications(components, mods, full_url)
+    connection, raw = build_raw_request(
+        method=modified["method"],
+        url=modified["url"],
+        headers=modified["headers"],
+        body=modified["body"],
+    )
+    return await replay_send_raw(await get_client(), raw=raw, connection=connection)
+
+
+async def scope_rules(
+    action: ScopeAction,
+    *,
+    allowlist: list[str] | None = None,
+    denylist: list[str] | None = None,
+    scope_id: str | None = None,
+    scope_name: str | None = None,
+) -> Any:
+    """Manage Caido scope rules from sandbox Python."""
+    client = await get_client()
+    if action == "list":
+        result = await scope_list(client)
+    elif action == "get":
+        if not scope_id:
+            raise ValueError("scope_id required for get")
+        result = await scope_get(client, scope_id)
+    elif action == "create":
+        if not scope_name:
+            raise ValueError("scope_name required for create")
+        result = await scope_create(
+            client,
+            name=scope_name,
+            allowlist=allowlist,
+            denylist=denylist,
+        )
+    elif action == "update":
+        if not scope_id or not scope_name:
+            raise ValueError("scope_id and scope_name required for update")
+        result = await scope_update(
+            client,
+            scope_id,
+            name=scope_name,
+            allowlist=allowlist,
+            denylist=denylist,
+        )
+    elif action == "delete":
+        if not scope_id:
+            raise ValueError("scope_id required for delete")
+        await scope_delete(client, scope_id)
+        result = {"deleted": scope_id}
+    else:
+        raise ValueError(f"Unknown action: {action}")
+    return result
+
+
+__all__ = [
+    "RequestPart",
+    "ScopeAction",
+    "SortBy",
+    "SortOrder",
+    "close_client",
+    "get_client",
+    "list_requests",
+    "repeat_request",
+    "scope_rules",
+    "send_request",
+    "view_request",
+]

strix/tools/proxy/tools.py

@@ -1,10 +1,9 @@
-"""Caido proxy tools — host-side ``@function_tool`` wrappers around ``_calls``.
+"""Caido proxy tools — host-side ``@function_tool`` wrappers.
 
-The five tools delegate to :mod:`strix.tools.proxy._calls` for the actual
+The five tools delegate to :mod:`strix.tools.proxy.caido_api` for the actual
 caido-sdk-client work and add LLM-friendly JSON serialization + error
-wrapping on top. The shared call layer is also reused by the
+wrapping on top. The delegated ``caido_api.py`` module is also copied into
-``python_action`` tool to expose the same proxy surface inside the
+the sandbox image as the importable ``caido_api`` Python module.
-sandbox's Python kernel — single source of truth for the SDK shapes.
 
 Tools: ``list_requests``, ``view_request``, ``send_request``,
 ``repeat_request``, ``scope_rules``.
@@ -22,7 +21,7 @@ from typing import TYPE_CHECKING, Any, Literal
 
 from agents import RunContextWrapper, function_tool
 
-from strix.tools.proxy import _calls
+from strix.tools.proxy import caido_api
 
 
 logger = logging.getLogger(__name__)
@@ -31,13 +30,13 @@ logger = logging.getLogger(__name__)
 if TYPE_CHECKING:
     from caido_sdk_client import Client
 
-    from strix.tools.proxy._calls import RequestPart, SortBy, SortOrder
+    from strix.tools.proxy.caido_api import RequestPart, SortBy, SortOrder
 else:
     # Runtime import: ``function_tool`` resolves the annotations via
     # ``typing.get_type_hints`` so the Literal aliases must be reachable
     # in module globals at decoration time even though they're "only"
     # used in annotations.
-    from strix.tools.proxy._calls import (  # noqa: TC001
+    from strix.tools.proxy.caido_api import (  # noqa: TC001
         RequestPart,
         SortBy,
         SortOrder,
@@ -52,8 +51,10 @@ def _ctx_client(ctx: RunContextWrapper) -> Client | None:
     return inner.get("caido_client")
 
 
-def _serialize(value: Any) -> Any:
+# Tool-output formatting. Caido SDK returns typed Python objects; function
-    """Recursively convert SDK dataclasses/Pydantic objects to JSON-safe primitives."""
+# tools need compact JSON-safe values for the model and TUI.
+def _to_tool_json(value: Any) -> Any:
+    """Recursively convert SDK dataclasses/Pydantic objects to tool JSON values."""
     if value is None or isinstance(value, str | int | float | bool):
         return value
     if isinstance(value, bytes):
@@ -64,13 +65,13 @@ def _serialize(value: Any) -> Any:
     if isinstance(value, datetime):
         return value.isoformat()
     if is_dataclass(value) and not isinstance(value, type):
-        return {k: _serialize(v) for k, v in dataclasses.asdict(value).items()}
+        return {k: _to_tool_json(v) for k, v in dataclasses.asdict(value).items()}
     if hasattr(value, "model_dump"):
-        return _serialize(value.model_dump())
+        return _to_tool_json(value.model_dump())
     if isinstance(value, dict):
-        return {str(k): _serialize(v) for k, v in value.items()}
+        return {str(k): _to_tool_json(v) for k, v in value.items()}
     if isinstance(value, list | tuple | set):
-        return [_serialize(v) for v in value]
+        return [_to_tool_json(v) for v in value]
     return str(value)
 
 
@@ -145,7 +146,7 @@ async def list_requests(
         return _no_client()
 
     try:
-        connection = await _calls.list_requests(
+        connection = await caido_api.list_requests_with_client(
             client,
             httpql_filter=httpql_filter,
             first=first,
@@ -246,7 +247,7 @@ async def view_request(
         return _no_client()
 
     try:
-        result = await _calls.get_request(client, request_id, part=part)
+        result = await caido_api.get_request_with_client(client, request_id, part=part)
         if result is None:
             return json.dumps(
                 {"success": False, "error": f"Request {request_id} not found"},
@@ -271,10 +272,14 @@ async def view_request(
         content = raw_bytes.decode("utf-8", errors="replace")
 
         if search_pattern:
-            return json.dumps(_regex_hits(content, search_pattern), ensure_ascii=False, default=str)
+            return json.dumps(
+                _format_search_hits(content, search_pattern),
+                ensure_ascii=False,
+                default=str,
+            )
 
         return json.dumps(
-            _paginate_lines(content, page=page, page_size=page_size),
+            _format_text_page(content, page=page, page_size=page_size),
             ensure_ascii=False,
             default=str,
         )
@@ -282,7 +287,7 @@ async def view_request(
         return _err("view_request", exc)
 
 
-def _regex_hits(content: str, pattern: str) -> dict[str, Any]:
+def _format_search_hits(content: str, pattern: str) -> dict[str, Any]:
     try:
         regex = re.compile(pattern)
     except re.error as exc:
@@ -307,7 +312,7 @@ def _regex_hits(content: str, pattern: str) -> dict[str, Any]:
     return {"success": True, "hits": hits, "total_hits": len(hits)}
 
 
-def _paginate_lines(content: str, *, page: int, page_size: int) -> dict[str, Any]:
+def _format_text_page(content: str, *, page: int, page_size: int) -> dict[str, Any]:
     lines = content.splitlines()
     start = max(0, (page - 1) * page_size)
     end = start + page_size
@@ -353,11 +358,11 @@ async def send_request(
         return _no_client()
 
     try:
-        connection, raw = _calls.build_raw_request(
+        connection, raw = caido_api.build_raw_request(
             method=method, url=url, headers=headers or {}, body=body
         )
-        result = await _calls.replay_send_raw(client, raw=raw, connection=connection)
+        result = await caido_api.replay_send_raw(client, raw=raw, connection=connection)
-        return _format_replay_result(result)
+        return _format_replay_tool_result(result)
     except Exception as exc:  # noqa: BLE001
         return _err("send_request", exc)
 
@@ -402,7 +407,7 @@ async def repeat_request(
     mods = modifications or {}
 
     try:
-        result = await _calls.get_request(client, request_id, part="request")
+        result = await caido_api.get_request_with_client(client, request_id, part="request")
         if result is None or result.request.raw is None:
             return json.dumps(
                 {"success": False, "error": f"Request {request_id} not found"},
@@ -412,22 +417,22 @@ async def repeat_request(
 
         original = result.request
         raw_str = result.request.raw.decode("utf-8", errors="replace")
-        components = _calls.parse_raw_request(raw_str)
+        components = caido_api.parse_raw_request(raw_str)
-        full_url = _calls.full_url_from_components(original, components, mods)
+        full_url = caido_api.full_url_from_components(original, components, mods)
-        modified = _calls.apply_modifications(components, mods, full_url)
+        modified = caido_api.apply_modifications(components, mods, full_url)
-        connection, raw = _calls.build_raw_request(
+        connection, raw = caido_api.build_raw_request(
             method=modified["method"],
             url=modified["url"],
             headers=modified["headers"],
             body=modified["body"],
         )
-        replay = await _calls.replay_send_raw(client, raw=raw, connection=connection)
+        replay = await caido_api.replay_send_raw(client, raw=raw, connection=connection)
-        return _format_replay_result(replay)
+        return _format_replay_tool_result(replay)
     except Exception as exc:  # noqa: BLE001
         return _err("repeat_request", exc)
 
 
-def _format_replay_result(replay: dict[str, Any]) -> str:
+def _format_replay_tool_result(replay: dict[str, Any]) -> str:
     response_raw = replay.get("response_raw")
     response: dict[str, Any] | None = None
     if response_raw is not None:
@@ -501,9 +506,9 @@ async def scope_rules(
 
     try:
         if action == "list":
-            scopes = await _calls.scope_list(client)
+            scopes = await caido_api.scope_list(client)
             return json.dumps(
-                {"success": True, "scopes": [_serialize(s) for s in scopes]},
+                {"success": True, "scopes": [_to_tool_json(s) for s in scopes]},
                 ensure_ascii=False,
                 default=str,
             )
@@ -514,9 +519,9 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await _calls.scope_get(client, scope_id)
+            scope = await caido_api.scope_get(client, scope_id)
             return json.dumps(
-                {"success": True, "scope": _serialize(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
             )
         if action == "create":
             if not scope_name:
@@ -525,11 +530,11 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await _calls.scope_create(
+            scope = await caido_api.scope_create(
                 client, name=scope_name, allowlist=allowlist, denylist=denylist
             )
             return json.dumps(
-                {"success": True, "scope": _serialize(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
             )
         if action == "update":
             if not scope_id or not scope_name:
@@ -541,11 +546,11 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await _calls.scope_update(
+            scope = await caido_api.scope_update(
                 client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
             )
             return json.dumps(
-                {"success": True, "scope": _serialize(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
             )
         # action == "delete" — exhaustive Literal
         if not scope_id:
@@ -554,7 +559,7 @@ async def scope_rules(
                 ensure_ascii=False,
                 default=str,
             )
-        await _calls.scope_delete(client, scope_id)
+        await caido_api.scope_delete(client, scope_id)
         return json.dumps({"success": True, "deleted": scope_id}, ensure_ascii=False, default=str)
     except Exception as exc:  # noqa: BLE001
         return _err("scope_rules", exc)

strix/tools/python/tool.py

@@ -1,308 +0,0 @@
-"""``python_action`` — execute Python code in the sandbox with proxy helpers.
-
-Stateless per-call. Each invocation:
-
-1. Ships the shared :mod:`strix.tools.proxy._calls` module source into
-   ``/tmp/`` inside the sandbox (single source of truth: the host-side
-   ``proxy/tools.py`` and this kernel both import the same file). The
-   logic is host-managed; updating the proxy helpers does not require
-   an image rebuild.
-2. Writes a per-call driver that connects a fresh ``caido_sdk_client``
-   to the in-container Caido at ``localhost:48080``, defines user-facing
-   wrappers (``list_requests``, ``view_request``, ``send_request``,
-   ``repeat_request``, ``scope_rules``) bound to that client, then
-   ``exec``s the user's code wrapped in an ``async def`` so top-level
-   ``await`` works.
-3. Captures ``stdout`` / ``stderr``, parses a sentinel-delimited JSON
-   payload from the driver's output, and returns it as the tool result.
-
-State is **not** preserved across calls. For multi-step workflows, write
-a single combined script via ``apply_patch`` and run it with
-``exec_command``, or pass intermediate results through files in
-``/workspace/scratch/``.
-"""
-
-from __future__ import annotations
-
-import base64
-import importlib.resources
-import io
-import json
-import logging
-import uuid
-from pathlib import Path
-from typing import TYPE_CHECKING
-
-from agents import RunContextWrapper, function_tool
-
-
-if TYPE_CHECKING:
-    from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
-
-
-logger = logging.getLogger(__name__)
-
-
-_CALLS_SOURCE = (importlib.resources.files("strix.tools.proxy") / "_calls.py").read_text(
-    encoding="utf-8"
-)
-
-
-_DRIVER_TEMPLATE = '''\
-"""Auto-generated python_action driver. Do not edit."""
-
-from __future__ import annotations
-
-import asyncio
-import base64
-import io
-import json
-import sys
-import textwrap
-import traceback
-import urllib.request
-
-sys.path.insert(0, "/tmp")
-import strix_calls  # noqa: E402  shipped alongside this driver
-
-from caido_sdk_client import Client  # noqa: E402
-from caido_sdk_client.types import TokenAuthOptions  # noqa: E402
-
-
-_SENTINEL = "{sentinel}"
-_USER_CODE_B64 = "{user_code_b64}"
-
-
-def _login_as_guest() -> str:
-    body = json.dumps(
-        {{"query": "mutation {{ loginAsGuest {{ token {{ accessToken }} }} }}"}}
-    ).encode("utf-8")
-    req = urllib.request.Request(
-        "http://127.0.0.1:48080/graphql",
-        data=body,
-        headers={{"Content-Type": "application/json"}},
-        method="POST",
-    )
-    with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
-        payload = json.loads(resp.read())
-    return str(payload["data"]["loginAsGuest"]["token"]["accessToken"])
-
-
-async def _strix_main() -> None:
-    client = Client("http://127.0.0.1:48080", auth=TokenAuthOptions(token=_login_as_guest()))
-    await client.connect()
-
-    async def list_requests(**kwargs):
-        return await strix_calls.list_requests(client, **kwargs)
-
-    async def view_request(request_id, *, part="request"):
-        return await strix_calls.get_request(client, request_id, part=part)
-
-    async def send_request(method, url, *, headers=None, body=""):
-        connection, raw = strix_calls.build_raw_request(
-            method=method, url=url, headers=headers or {{}}, body=body
-        )
-        return await strix_calls.replay_send_raw(client, raw=raw, connection=connection)
-
-    async def repeat_request(request_id, *, modifications=None):
-        mods = modifications or {{}}
-        result = await strix_calls.get_request(client, request_id, part="request")
-        if result is None or result.request.raw is None:
-            raise ValueError(f"Request {{request_id}} not found")
-        original = result.request
-        raw_str = result.request.raw.decode("utf-8", errors="replace")
-        components = strix_calls.parse_raw_request(raw_str)
-        full_url = strix_calls.full_url_from_components(original, components, mods)
-        modified = strix_calls.apply_modifications(components, mods, full_url)
-        connection, raw = strix_calls.build_raw_request(
-            method=modified["method"],
-            url=modified["url"],
-            headers=modified["headers"],
-            body=modified["body"],
-        )
-        return await strix_calls.replay_send_raw(client, raw=raw, connection=connection)
-
-    async def scope_rules(action, *, allowlist=None, denylist=None, scope_id=None, scope_name=None):
-        if action == "list":
-            return await strix_calls.scope_list(client)
-        if action == "get":
-            if not scope_id:
-                raise ValueError("scope_id required for get")
-            return await strix_calls.scope_get(client, scope_id)
-        if action == "create":
-            if not scope_name:
-                raise ValueError("scope_name required for create")
-            return await strix_calls.scope_create(
-                client, name=scope_name, allowlist=allowlist, denylist=denylist
-            )
-        if action == "update":
-            if not scope_id or not scope_name:
-                raise ValueError("scope_id and scope_name required for update")
-            return await strix_calls.scope_update(
-                client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
-            )
-        if action == "delete":
-            if not scope_id:
-                raise ValueError("scope_id required for delete")
-            await strix_calls.scope_delete(client, scope_id)
-            return {{"deleted": scope_id}}
-        raise ValueError(f"Unknown action: {{action}}")
-
-    user_code = base64.b64decode(_USER_CODE_B64).decode("utf-8")
-    wrapped = "async def __strix_user():\\n    pass\\n" + textwrap.indent(user_code, "    ")
-
-    namespace: dict = {{
-        "list_requests": list_requests,
-        "view_request": view_request,
-        "send_request": send_request,
-        "repeat_request": repeat_request,
-        "scope_rules": scope_rules,
-        "client": client,
-    }}
-
-    out_buf = io.StringIO()
-    err_buf = io.StringIO()
-    error: str | None = None
-
-    real_stdout = sys.stdout
-    real_stderr = sys.stderr
-    sys.stdout = out_buf
-    sys.stderr = err_buf
-    try:
-        try:
-            exec(compile(wrapped, "<python_action>", "exec"), namespace)  # noqa: S102
-            await namespace["__strix_user"]()
-        except BaseException:
-            error = traceback.format_exc()
-    finally:
-        sys.stdout = real_stdout
-        sys.stderr = real_stderr
-
-    payload = json.dumps(
-        {{
-            "stdout": out_buf.getvalue(),
-            "stderr": err_buf.getvalue(),
-            "error": error,
-        }},
-        ensure_ascii=False,
-    )
-    sys.stdout.write("\\n" + _SENTINEL + payload + "\\n")
-
-
-asyncio.run(_strix_main())
-'''
-
-
-@function_tool(timeout=180)
-async def python_action(
-    ctx: RunContextWrapper,
-    code: str,
-    timeout: int = 60,
-) -> str:
-    """Execute Python code in the sandbox with Caido proxy helpers pre-bound.
-
-    Each call is a fresh process — variables, imports, and function
-    definitions do **not** persist between calls. For multi-step
-    workflows, combine into a single ``code`` block, or write a script
-    to ``/workspace/scratch/<name>.py`` via ``apply_patch`` and run with
-    ``exec_command``.
-
-    **Pre-bound proxy helpers** (no imports needed; all are async — use
-    ``await``):
-
-    - ``list_requests(httpql_filter=, first=, after=, sort_by=,
-      sort_order=, scope_id=)`` → cursor-paginated SDK ``Connection``.
-      Iterate ``connection.edges`` for entries.
-    - ``view_request(request_id, part="request")`` → SDK request object
-      with ``.request.raw`` / ``.response.raw`` bytes.
-    - ``send_request(method, url, headers=None, body="")`` → dict with
-      ``status``, ``response_raw``, ``elapsed_ms``.
-    - ``repeat_request(request_id, modifications={"url"|"headers"|
-      "body"|"params"|"cookies": ...})`` → same shape as
-      ``send_request``.
-    - ``scope_rules(action, allowlist=, denylist=, scope_id=,
-      scope_name=)`` → SDK scope objects.
-
-    Top-level ``await`` is supported — the body is wrapped in an async
-    function. Use ``print()`` to emit visible output; the last
-    expression is **not** auto-shown.
-
-    Args:
-        code: Python source to execute. Multi-line is fine.
-        timeout: Hard timeout in seconds (default 60). The container
-            is killed if the driver exceeds this.
-
-    Returns:
-        JSON dict with ``success``, ``stdout``, ``stderr``, ``error``
-        (a formatted traceback if the user code raised).
-    """
-    inner = ctx.context if isinstance(ctx.context, dict) else {}
-    session: BaseSandboxSession | None = inner.get("sandbox_session")
-    if session is None:
-        return json.dumps(
-            {"success": False, "error": "No sandbox session in run context"},
-            ensure_ascii=False,
-        )
-
-    sentinel = "__STRIX_PY_RESULT_" + uuid.uuid4().hex + "__"
-    user_code_b64 = base64.b64encode(code.encode("utf-8")).decode("ascii")
-    driver = _DRIVER_TEMPLATE.format(sentinel=sentinel, user_code_b64=user_code_b64)
-    logger.info("python_action: invoking driver (code_len=%d, timeout=%ds)", len(code), timeout)
-    # /tmp inside the sandbox container is single-user (pentester) and
-    # disposable per scan; the multi-user race B108/S108 warns about
-    # doesn't apply.
-    driver_path = Path(f"/tmp/strix_driver_{uuid.uuid4().hex}.py")  # nosec B108
-    calls_path = Path("/tmp/strix_calls.py")  # nosec B108
-
-    try:
-        await session.write(calls_path, io.BytesIO(_CALLS_SOURCE.encode("utf-8")))
-        await session.write(driver_path, io.BytesIO(driver.encode("utf-8")))
-        result = await session.exec("python3", "-u", str(driver_path), timeout=timeout)
-    except Exception as exc:  # noqa: BLE001
-        return json.dumps(
-            {"success": False, "error": f"python_action launch failed: {exc}"},
-            ensure_ascii=False,
-        )
-    finally:
-        # Best-effort cleanup; ignore failure (the file is in /tmp anyway).
-        try:
-            await session.exec("rm", "-f", str(driver_path), timeout=5)
-        except Exception:  # noqa: BLE001
-            logger.debug("cleanup failed for %s", driver_path)
-
-    raw_stdout = result.stdout.decode("utf-8", errors="replace")
-    raw_stderr = result.stderr.decode("utf-8", errors="replace")
-
-    if sentinel in raw_stdout:
-        head, _, tail = raw_stdout.rpartition(sentinel)
-        try:
-            payload = json.loads(tail.strip())
-        except json.JSONDecodeError as exc:
-            return json.dumps(
-                {
-                    "success": False,
-                    "error": f"Driver output not parseable: {exc}",
-                    "stdout": raw_stdout,
-                    "stderr": raw_stderr,
-                },
-                ensure_ascii=False,
-            )
-        return json.dumps(
-            {
-                "success": payload.get("error") is None,
-                "stdout": (head.rstrip() + payload.get("stdout", "")) or "",
-                "stderr": payload.get("stderr", "") or raw_stderr,
-                "error": payload.get("error"),
-            },
-            ensure_ascii=False,
-        )
-
-    return json.dumps(
-        {
-            "success": False,
-            "error": "Driver exited without producing a result sentinel",
-            "stdout": raw_stdout,
-            "stderr": raw_stderr,
-        },
-        ensure_ascii=False,
-    )