Restore sitemap tools + unify proxy I/O contract

Re-add list_sitemap and view_sitemap_entry from main, ported to the
new caido-sdk-client layout via raw GraphQL queries (the typed SDK
doesn't expose sitemap operations, but the Caido server still
supports sitemapRootEntries / sitemapDescendantEntries / sitemapEntry).
Wired through caido_api (sandbox-importable helpers), the host-side
@function_tool wrappers, factory _BASE_TOOLS, the system prompt, the
python skill doc, and the public proxy docs.

While threading these through, lock down the output contract across
every proxy tool so the model sees one consistent shape:

- All tools wrap success/failure in {"success": bool, "error"?: str}
- Canonical field names: status_code, length, roundtrip_ms (omitted
  when 0), is_tls, has_descendants. snake_case everywhere on output;
  camelCase stays only on the input side where it's the GraphQL
  schema.
- repeat_request now returns a structured response that matches
  list_requests' response_summary shape (parse_raw_response parses
  the raw bytes into status_code / length / headers / body), with
  body capped at 8KB and a body_truncated flag so the model knows
  when to fetch the full body via view_request.
- RepeatRequestRenderer was reading non-existent top-level keys
  (status_code, response_time_ms, body) and silently displaying
  nothing useful — now reads the structured response shape.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

docs/tools/proxy.mdx

@@ -66,6 +66,8 @@ asyncio.run(main())
 | `list_requests()`      | Query captured traffic with HTTPQL filters |
 | `view_request()`       | Get full request/response details          |
 | `repeat_request()`     | Replay a request with modifications        |
+| `list_sitemap()`       | Browse the request-tree view of discovered surface |
+| `view_sitemap_entry()` | Inspect one sitemap entry + its related requests |
 | `scope_rules()`        | Manage proxy scope (allowlist/denylist)    |
 
 For one-off arbitrary requests, use shell tooling like `curl` — the

strix/agents/factory.py

@@ -46,9 +46,11 @@ from strix.tools.notes.tools import (
 )
 from strix.tools.proxy.tools import (
     list_requests,
+    list_sitemap,
     repeat_request,
     scope_rules,
     view_request,
+    view_sitemap_entry,
 )
 from strix.tools.reporting.tool import create_vulnerability_report
 from strix.tools.thinking.tool import think
@@ -268,6 +270,8 @@ _BASE_TOOLS: tuple[Tool, ...] = (
     list_requests,
     view_request,
     repeat_request,
+    list_sitemap,
+    view_sitemap_entry,
     scope_rules,
     # Multi-agent graph tools (the coordinator is in ctx.context)
     view_agent_graph,

strix/agents/prompts/system_prompt.jinja

@@ -171,7 +171,7 @@ EFFICIENCY TACTICS:
   `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, repeat_request, scope_rules`
+  `from caido_api import list_requests, view_request, repeat_request, list_sitemap, view_sitemap_entry, 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 Python scripts through `exec_command` or terminal tools.
 - When using established fuzzers/scanners, use the proxy for inspection where helpful

strix/interface/tui/renderers/proxy_renderer.py

@@ -258,30 +258,26 @@ class RepeatRequestRenderer(BaseToolRenderer):
             text.append(f"\n  {_truncate(modifications, 200)}", style="dim italic")
 
         if status == "completed" and isinstance(result, dict):
-            if "error" in result:
+            if not result.get("success", True) and result.get("error"):
                 text.append(f"\n  error: {_sanitize(str(result['error']), 150)}", style="#ef4444")
             else:
-                req = result.get("request", {})
-                method = req.get("method", "")
-                url = req.get("url", "")
-                code = result.get("status_code")
-                time_ms = result.get("response_time_ms")
-
-                text.append("\n")
-                text.append("  >> ", style="#3b82f6")
-                if method:
-                    text.append(f"{method} ", style="#a78bfa")
-                if url:
-                    text.append(_truncate(url, 180), style="dim")
+                elapsed_ms = result.get("elapsed_ms")
+                response = result.get("response") or {}
+                code = response.get("status_code") if isinstance(response, dict) else None
+                body = response.get("body", "") if isinstance(response, dict) else ""
+                body_truncated = (
+                    bool(response.get("body_truncated")) if isinstance(response, dict) else False
+                )
 
                 text.append("\n")
                 text.append("  << ", style="#22c55e")
                 if code:
                     text.append(f"{code}", style=_status_style(code))
-                if time_ms:
-                    text.append(f" ({time_ms}ms)", style="dim")
+                else:
+                    text.append("(no response)", style="dim")
+                if elapsed_ms:
+                    text.append(f" ({elapsed_ms}ms)", style="dim")
 
-                body = result.get("body", "")
                 if body and isinstance(body, str):
                     lines = body.split("\n")[:5]
                     for line in lines:
@@ -289,7 +285,7 @@ class RepeatRequestRenderer(BaseToolRenderer):
                         text.append("  << ", style="#22c55e")
                         text.append(_truncate(line, MAX_LINE_LENGTH - 5), style="dim")
 
-                    if len(body.split("\n")) > 5:
+                    if body_truncated or len(body.split("\n")) > 5:
                         text.append("\n")
                         text.append("  ...", style="dim italic")
 

strix/skills/tooling/python.md

@@ -19,9 +19,11 @@ explicitly when Python code needs Caido traffic or replay access:
 ```python
 from caido_api import (
     list_requests,
+    list_sitemap,
     repeat_request,
     scope_rules,
     view_request,
+    view_sitemap_entry,
 )
 ```
 
@@ -59,6 +61,8 @@ Available helpers:
 - `list_requests(httpql_filter=, first=50, after=, sort_by=, sort_order=, scope_id=)` returns a cursor-paginated Caido SDK `Connection`.
 - `view_request(request_id, part="request")` returns a Caido SDK request object with raw request/response bytes.
 - `repeat_request(request_id, modifications={...})` replays a captured request after modifying `url`, `params`, `headers`, `body`, or `cookies`.
+- `list_sitemap(scope_id=, parent_id=, depth="DIRECT", page=1)` walks Caido's request-tree view of the discovered surface. Omit `parent_id` for root domains; pass an entry id with `depth="DIRECT"` or `"ALL"` to drill in.
+- `view_sitemap_entry(entry_id)` returns one entry plus its 30 most recent related requests.
 - `scope_rules(action, allowlist=, denylist=, scope_id=, scope_name=)` manages Caido scopes.
 
 For one-off arbitrary requests (e.g. probing a fresh endpoint, hitting an

strix/tools/proxy/caido_api.py

@@ -37,6 +37,8 @@ SortBy = Literal[
 ]
 SortOrder = Literal["asc", "desc"]
 ScopeAction = Literal["get", "list", "create", "update", "delete"]
+SitemapDepth = Literal["DIRECT", "ALL"]
+_SITEMAP_PAGE_SIZE = 30
 
 _DEFAULT_CAIDO_URL = "http://127.0.0.1:48080"
 _CLIENT_CACHE: dict[str, Client] = {}
@@ -167,6 +169,53 @@ def build_raw_request(
     return ConnectionInfoInput(host=host, port=port, is_tls=is_tls), raw
 
 
+# Cap inline response bodies returned through tool results so a single
+# large response (HTML pages, JSON dumps) can't blow out the model's
+# context. The model can re-fetch the full body via ``view_request``
+# using the captured request id from ``list_requests`` if it needs more.
+_RESPONSE_BODY_MAX_CHARS = 8192
+
+
+def parse_raw_response(raw_bytes: bytes | None) -> dict[str, Any] | None:
+    """Parse a raw HTTP response into the same shape ``list_requests`` emits.
+
+    Returns ``None`` when ``raw_bytes`` is missing or unparseable. On
+    success returns ``{status_code, length, headers, body, body_truncated}``
+    where ``body`` is decoded as UTF-8 (replacement chars on invalid
+    bytes) and clipped at :data:`_RESPONSE_BODY_MAX_CHARS`.
+    """
+    if not raw_bytes:
+        return None
+    try:
+        head, _, body_bytes = raw_bytes.partition(b"\r\n\r\n")
+        lines = head.decode("iso-8859-1", errors="replace").split("\r\n")
+        if not lines:
+            return None
+        status_parts = lines[0].split(" ", 2)
+        if len(status_parts) < 2 or not status_parts[1].isdigit():
+            return None
+        status_code = int(status_parts[1])
+        headers: dict[str, str] = {}
+        for line in lines[1:]:
+            if ":" not in line:
+                continue
+            k, v = line.split(":", 1)
+            headers[k.strip()] = v.strip()
+        body_text = body_bytes.decode("utf-8", errors="replace")
+        body_truncated = len(body_text) > _RESPONSE_BODY_MAX_CHARS
+        if body_truncated:
+            body_text = body_text[:_RESPONSE_BODY_MAX_CHARS]
+        return {
+            "status_code": status_code,
+            "length": len(body_bytes),
+            "headers": headers,
+            "body": body_text,
+            "body_truncated": body_truncated,
+        }
+    except Exception:  # noqa: BLE001 - tolerate any malformed raw bytes; None signals "unparseable" to the caller.
+        return None
+
+
 def parse_raw_request(raw_content: str) -> dict[str, Any]:
     lines = raw_content.split("\n")
     request_line = lines[0].strip().split(" ")
@@ -440,15 +489,223 @@ async def scope_rules(
     return result
 
 
+_SITEMAP_ROOTS_QUERY = """
+query GetSitemapRoots($scopeId: ID) {
+    sitemapRootEntries(scopeId: $scopeId) {
+        edges { node {
+            id kind label hasDescendants
+            metadata { ... on SitemapEntryMetadataDomain { isTls port } }
+            request { method path response { statusCode } }
+        } }
+        count { value }
+    }
+}
+"""
+
+_SITEMAP_DESCENDANTS_QUERY = """
+query GetSitemapDescendants($parentId: ID!, $depth: SitemapDescendantsDepth!) {
+    sitemapDescendantEntries(parentId: $parentId, depth: $depth) {
+        edges { node {
+            id kind label hasDescendants
+            request { method path response { statusCode } }
+        } }
+        count { value }
+    }
+}
+"""
+
+_SITEMAP_ENTRY_QUERY = """
+query GetSitemapEntry($id: ID!) {
+    sitemapEntry(id: $id) {
+        id kind label hasDescendants
+        metadata { ... on SitemapEntryMetadataDomain { isTls port } }
+        request { method path response { statusCode length roundtripTime } }
+        requests(first: 30, order: {by: CREATED_AT, ordering: DESC}) {
+            edges { node { method path response { statusCode length } } }
+            count { value }
+        }
+    }
+}
+"""
+
+
+def _clean_sitemap_metadata(node: dict[str, Any]) -> dict[str, Any]:
+    cleaned: dict[str, Any] = {
+        "id": node["id"],
+        "kind": node["kind"],
+        "label": node["label"],
+        "has_descendants": node["hasDescendants"],
+    }
+    meta = node.get("metadata")
+    if isinstance(meta, dict) and (meta.get("isTls") is not None or meta.get("port")):
+        meta_out: dict[str, Any] = {}
+        if meta.get("isTls") is not None:
+            meta_out["is_tls"] = meta["isTls"]
+        if meta.get("port"):
+            meta_out["port"] = meta["port"]
+        cleaned["metadata"] = meta_out
+    return cleaned
+
+
+def _clean_sitemap_request_summary(req: dict[str, Any] | None) -> dict[str, Any] | None:
+    """Same field names as ``list_requests`` emits for a request_summary."""
+    if not req:
+        return None
+    out: dict[str, Any] = {}
+    if req.get("method"):
+        out["method"] = req["method"]
+    if req.get("path"):
+        out["path"] = req["path"]
+    resp = req.get("response") or {}
+    if resp.get("statusCode"):
+        out["status_code"] = resp["statusCode"]
+    return out or None
+
+
+def _clean_sitemap_response(resp: dict[str, Any]) -> dict[str, Any]:
+    """Same field names as ``list_requests`` emits for a response_summary."""
+    out: dict[str, Any] = {}
+    if resp.get("statusCode"):
+        out["status_code"] = resp["statusCode"]
+    if resp.get("length"):
+        out["length"] = resp["length"]
+    # Suppress 0 the same way list_requests does — Caido leaves it unset
+    # on a lot of proxy-captured traffic and a misleading "0ms" is worse
+    # than the field simply being absent.
+    if resp.get("roundtripTime"):
+        out["roundtrip_ms"] = resp["roundtripTime"]
+    return out
+
+
+async def list_sitemap_with_client(
+    client: CaidoClient,
+    *,
+    scope_id: str | None = None,
+    parent_id: str | None = None,
+    depth: SitemapDepth = "DIRECT",
+    page: int = 1,
+    page_size: int = _SITEMAP_PAGE_SIZE,
+) -> dict[str, Any]:
+    """Browse Caido's discovered sitemap. Mirrors main-branch shape.
+
+    The Caido GraphQL ``sitemap*Entries`` operations don't support native
+    pagination, so we fetch all edges for the requested level and slice
+    client-side. That's fine for typical surface sizes; for very large
+    sitemaps the caller can drill into ``parent_id`` instead of paging
+    the root list.
+    """
+    if parent_id:
+        raw = await client.graphql.query(
+            _SITEMAP_DESCENDANTS_QUERY,
+            variables={"parentId": parent_id, "depth": depth},
+        )
+        data = raw.get("sitemapDescendantEntries") or {}
+    else:
+        raw = await client.graphql.query(
+            _SITEMAP_ROOTS_QUERY,
+            variables={"scopeId": scope_id},
+        )
+        data = raw.get("sitemapRootEntries") or {}
+
+    edges = data.get("edges") or []
+    total = (data.get("count") or {}).get("value", 0)
+    skip = max(0, (page - 1) * page_size)
+    sliced = [edge["node"] for edge in edges[skip : skip + page_size]]
+
+    cleaned: list[dict[str, Any]] = []
+    for node in sliced:
+        entry = _clean_sitemap_metadata(node)
+        summary = _clean_sitemap_request_summary(node.get("request"))
+        if summary:
+            entry["request"] = summary
+        cleaned.append(entry)
+
+    total_pages = (total + page_size - 1) // page_size if total else 0
+    return {
+        "success": True,
+        "entries": cleaned,
+        "page": page,
+        "page_size": page_size,
+        "total_pages": total_pages,
+        "total_count": total,
+        "has_more": page < total_pages,
+    }
+
+
+async def view_sitemap_entry_with_client(
+    client: CaidoClient,
+    entry_id: str,
+) -> dict[str, Any]:
+    """Fetch one sitemap entry plus its recent related requests."""
+    raw = await client.graphql.query(_SITEMAP_ENTRY_QUERY, variables={"id": entry_id})
+    entry = raw.get("sitemapEntry")
+    if not entry:
+        return {"success": False, "error": f"Sitemap entry {entry_id} not found"}
+
+    cleaned = _clean_sitemap_metadata(entry)
+    primary = entry.get("request") or {}
+    if primary:
+        primary_clean: dict[str, Any] = {}
+        if primary.get("method"):
+            primary_clean["method"] = primary["method"]
+        if primary.get("path"):
+            primary_clean["path"] = primary["path"]
+        if primary.get("response"):
+            primary_clean["response"] = _clean_sitemap_response(primary["response"])
+        if primary_clean:
+            cleaned["request"] = primary_clean
+
+    related = entry.get("requests") or {}
+    related_edges = related.get("edges") or []
+    related_nodes = [edge["node"] for edge in related_edges]
+    related_clean = [
+        summary
+        for summary in (_clean_sitemap_request_summary(n) for n in related_nodes)
+        if summary is not None
+    ]
+    cleaned["related_requests"] = {
+        "requests": related_clean,
+        "total_count": (related.get("count") or {}).get("value", 0),
+    }
+    return {"success": True, "entry": cleaned}
+
+
+async def list_sitemap(
+    *,
+    scope_id: str | None = None,
+    parent_id: str | None = None,
+    depth: SitemapDepth = "DIRECT",
+    page: int = 1,
+    page_size: int = _SITEMAP_PAGE_SIZE,
+) -> dict[str, Any]:
+    """Sandbox-Python entry point for sitemap browsing."""
+    return await list_sitemap_with_client(
+        await get_client(),
+        scope_id=scope_id,
+        parent_id=parent_id,
+        depth=depth,
+        page=page,
+        page_size=page_size,
+    )
+
+
+async def view_sitemap_entry(entry_id: str) -> dict[str, Any]:
+    """Sandbox-Python entry point for sitemap entry detail."""
+    return await view_sitemap_entry_with_client(await get_client(), entry_id)
+
+
 __all__ = [
     "RequestPart",
     "ScopeAction",
+    "SitemapDepth",
     "SortBy",
     "SortOrder",
     "close_client",
     "get_client",
     "list_requests",
+    "list_sitemap",
     "repeat_request",
     "scope_rules",
     "view_request",
+    "view_sitemap_entry",
 ]

strix/tools/proxy/tools.py

@@ -33,7 +33,12 @@ logger = logging.getLogger(__name__)
 if TYPE_CHECKING:
     from caido_sdk_client import Client
 
-    from strix.tools.proxy.caido_api import RequestPart, SortBy, SortOrder
+    from strix.tools.proxy.caido_api import (
+        RequestPart,
+        SitemapDepth,
+        SortBy,
+        SortOrder,
+    )
 else:
     # Runtime import: ``function_tool`` resolves the annotations via
     # ``typing.get_type_hints`` so the Literal aliases must be reachable
@@ -41,6 +46,7 @@ else:
     # used in annotations.
     from strix.tools.proxy.caido_api import (  # noqa: TC001
         RequestPart,
+        SitemapDepth,
         SortBy,
         SortOrder,
     )
@@ -400,22 +406,97 @@ async def repeat_request(
 
 
 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:
-        response = {"raw": response_raw.decode("utf-8", errors="replace")}
-    return json.dumps(
-        {
-            "success": replay["status"] == "DONE",
-            "status": replay["status"],
-            "error": replay["error"],
-            "session_id": replay["session_id"],
-            "elapsed_ms": replay["elapsed_ms"],
-            "response": response,
-        },
-        ensure_ascii=False,
-        default=str,
-    )
+    response = caido_api.parse_raw_response(replay.get("response_raw"))
+    payload: dict[str, Any] = {
+        "success": replay["status"] == "DONE",
+        "status": replay["status"],
+        "session_id": replay["session_id"],
+        "elapsed_ms": replay["elapsed_ms"],
+        "response": response,
+    }
+    if replay.get("error"):
+        payload["error"] = replay["error"]
+    return json.dumps(payload, ensure_ascii=False, default=str)
+
+
+# ----------------------------------------------------------------------
+# list_sitemap
+# ----------------------------------------------------------------------
+@function_tool(timeout=60)
+async def list_sitemap(
+    ctx: RunContextWrapper,
+    scope_id: str | None = None,
+    parent_id: str | None = None,
+    depth: SitemapDepth = "DIRECT",
+    page: int = 1,
+) -> str:
+    """Browse Caido's hierarchical sitemap of proxied traffic.
+
+    Caido aggregates every captured request into a tree:
+    ``DOMAIN`` → ``DIRECTORY`` (path segments) → ``REQUEST`` →
+    ``REQUEST_BODY`` / ``REQUEST_QUERY`` (variant per body/query shape).
+    Use this to understand the discovered attack surface, locate
+    promising directories, and pick endpoints worth deeper testing.
+
+    Workflow:
+    - Start with no ``parent_id`` to list root domains (scoped by
+      ``scope_id`` if you only care about in-scope hosts).
+    - Pick an entry where ``hasDescendants=true`` and pass its ``id``
+      as ``parent_id`` to drill in. ``depth="DIRECT"`` returns only
+      immediate children; ``"ALL"`` flattens the full subtree.
+    - Hand any ``id`` to ``view_sitemap_entry`` for the full record
+      and recent matching requests.
+
+    Args:
+        scope_id: Limit roots to a Caido scope (only used when
+            ``parent_id`` is omitted). Manage scopes via ``scope_rules``.
+        parent_id: Entry ID to expand; omit for root domains.
+        depth: ``"DIRECT"`` (immediate children) or ``"ALL"``
+            (recursive subtree). Only meaningful with ``parent_id``.
+        page: 1-indexed page (30 entries per page).
+    """
+    client = _ctx_client(ctx)
+    if client is None:
+        return _no_client()
+    try:
+        payload = await caido_api.list_sitemap_with_client(
+            client,
+            scope_id=scope_id,
+            parent_id=parent_id,
+            depth=depth,
+            page=page,
+        )
+        return json.dumps(payload, ensure_ascii=False, default=str)
+    except Exception as exc:  # noqa: BLE001
+        return _err("list_sitemap", exc)
+
+
+# ----------------------------------------------------------------------
+# view_sitemap_entry
+# ----------------------------------------------------------------------
+@function_tool(timeout=60)
+async def view_sitemap_entry(
+    ctx: RunContextWrapper,
+    entry_id: str,
+) -> str:
+    """Get full detail for a sitemap entry plus its recent requests.
+
+    Returns the entry's metadata, the primary request shape
+    (method/path/response if any), and the most recent 30 related
+    requests that fall under this entry. Pair with ``list_sitemap`` to
+    pick the ``entry_id``.
+
+    Args:
+        entry_id: ID from ``list_sitemap`` (or any nested entry).
+    """
+    client = _ctx_client(ctx)
+    if client is None:
+        return _no_client()
+    try:
+        payload = await caido_api.view_sitemap_entry_with_client(client, entry_id)
+        return json.dumps(payload, ensure_ascii=False, default=str)
+    except Exception as exc:  # noqa: BLE001
+        return _err("view_sitemap_entry", exc)
 
 
 # ----------------------------------------------------------------------