feat(migration): phase 2.4 — wrap remaining local SDK tools

Five tool families ported to SDK function tools using the proven
delegation pattern from Phase 2.3:

- web_search (1 tool): asyncio.to_thread around the synchronous
  Perplexity request so the 300s API call doesn't block the SDK
  event loop.

- file_edit (3 tools — str_replace_editor, list_files, search_files):
  these run *inside* the sandbox container in the legacy harness
  (sandbox_execution=True), so the SDK wrappers route through
  post_to_sandbox rather than importing the legacy module on the
  host (which pulls in openhands_aci, a sandbox-only dependency).

- reporting (1 tool — create_vulnerability_report): asyncio.to_thread
  around the legacy function, which itself runs CVSS XML parsing,
  LLM-based dedup against existing findings, and tracer persistence.

- load_skill (1 tool): legacy adapter passes ctx.context['agent_id']
  through. The legacy implementation reaches into _agent_instances,
  a global Phase 3 will replace; until then the call degrades to a
  structured error rather than crashing.

- finish_scan (1 tool): legacy adapter pattern. Validates non-empty
  fields, checks no other agents are still active (via legacy
  _agent_graph), persists the four executive sections through the
  global tracer.

Tests: 12 new tests in test_sdk_remaining_local_tools.py — registration
checks, web_search delegation + missing-key path, file_edit dispatch
shape verification, vuln-report validation + delegation, load_skill
adapter passthrough, finish_scan validation + delegation. The two
finish_scan tests use a fixture that snapshots/clears the legacy
_agent_graph['nodes'] dict so cross-test pollution from legacy
multi-agent tests doesn't mask the validation path.

Per-file ruff TC002 ignores added for the five new wrapper modules
(same reason as Phase 2.3 — RunContextWrapper must be runtime-importable
for SDK function_schema().get_type_hints()).

Refs: PLAYBOOK.md §3.5.
This commit is contained in:
0xallam
2026-04-25 00:21:37 -07:00
parent b7ac7cc1a5
commit d25980bd8d
7 changed files with 700 additions and 0 deletions
@@ -0,0 +1,109 @@
"""SDK function-tool wrappers for the legacy ``file_edit`` tools.
These three tools (``str_replace_editor``, ``list_files``, ``search_files``)
operate on files inside the sandbox container's ``/workspace`` filesystem.
The legacy harness marks them ``sandbox_execution=True`` (default) so the
executor POSTs them to the in-container tool server.
The host-side SDK wrappers therefore delegate to ``post_to_sandbox`` —
the legacy implementations live in the container image and we don't
import them on the host (they pull in ``openhands_aci``, which is a
sandbox-only dependency).
"""
from __future__ import annotations
import json
from typing import Any
from agents import RunContextWrapper
from strix.tools._decorator import strix_tool
from strix.tools._sandbox_dispatch import post_to_sandbox
def _dump(result: dict[str, Any]) -> str:
return json.dumps(result, ensure_ascii=False, default=str)
@strix_tool(timeout=180)
async def str_replace_editor(
ctx: RunContextWrapper,
command: str,
path: str,
file_text: str | None = None,
view_range: list[int] | None = None,
old_str: str | None = None,
new_str: str | None = None,
insert_line: int | None = None,
) -> str:
"""View, create, or edit a file in the sandbox.
Args:
command: One of ``"view" | "create" | "str_replace" | "insert" |
"undo_edit"``.
path: File path. Relative paths are anchored at ``/workspace``.
file_text: Required for ``create``.
view_range: Optional ``[start, end]`` line range for ``view``.
old_str / new_str: Required for ``str_replace``.
insert_line: Required for ``insert``.
"""
return _dump(
await post_to_sandbox(
ctx,
"str_replace_editor",
{
"command": command,
"path": path,
"file_text": file_text,
"view_range": view_range,
"old_str": old_str,
"new_str": new_str,
"insert_line": insert_line,
},
),
)
@strix_tool(timeout=120)
async def list_files(
ctx: RunContextWrapper,
path: str,
recursive: bool = False,
) -> str:
"""List files and directories under a sandbox path.
Args:
path: Directory path, relative paths anchored at ``/workspace``.
recursive: When True, walks subdirectories (capped at 500 entries).
"""
return _dump(
await post_to_sandbox(
ctx,
"list_files",
{"path": path, "recursive": recursive},
),
)
@strix_tool(timeout=120)
async def search_files(
ctx: RunContextWrapper,
path: str,
regex: str,
file_pattern: str = "*",
) -> str:
"""Recursively grep files in the sandbox using ripgrep.
Args:
path: Root path to search; relative paths anchored at ``/workspace``.
regex: Pattern to match (passed straight to ``rg``).
file_pattern: Glob filter (e.g. ``"*.py"``). Defaults to all files.
"""
return _dump(
await post_to_sandbox(
ctx,
"search_files",
{"path": path, "regex": regex, "file_pattern": file_pattern},
),
)
+67
View File
@@ -0,0 +1,67 @@
"""SDK function-tool wrapper for the legacy ``finish_scan`` tool.
The legacy function:
- Validates the caller is the root agent (``parent_id is None``).
- Checks no other agents are still running (via the legacy
``_agent_graph`` global).
- Persists the four executive-summary fields via
``get_global_tracer().update_scan_final_fields(...)``.
- Reports the final vulnerability count.
Both the parent-id check and the agent-graph check rely on legacy
multi-agent state that Phase 3 will reimplement on top of the SDK
``RunContextWrapper`` + a per-run registry. Until Phase 3 lands, the
legacy adapter returns an object with no ``parent_id`` attribute —
``hasattr`` returns False, the validation skips, and the call proceeds
as if invoked by a root agent. That's the correct degenerate behavior
in single-agent mode, which is all Phase 2 ships.
"""
from __future__ import annotations
import asyncio
import json
from typing import Any
from agents import RunContextWrapper
from strix.tools._decorator import strix_tool
from strix.tools._legacy_adapter import adapter_from_ctx
from strix.tools.finish import finish_actions as _legacy
def _dump(result: dict[str, Any]) -> str:
return json.dumps(result, ensure_ascii=False, default=str)
@strix_tool(timeout=60)
async def finish_scan(
ctx: RunContextWrapper,
executive_summary: str,
methodology: str,
technical_analysis: str,
recommendations: str,
) -> str:
"""Finalize the scan and persist the four executive summary sections.
Only the root agent should call this. Subagents should use
``agent_finish`` from the agents_graph tool family instead.
Args:
executive_summary: High-level scan outcome.
methodology: Approach taken.
technical_analysis: Findings detail across the engagement.
recommendations: Prioritized fix list.
"""
state = adapter_from_ctx(ctx)
return _dump(
await asyncio.to_thread(
_legacy.finish_scan,
executive_summary=executive_summary,
methodology=methodology,
technical_analysis=technical_analysis,
recommendations=recommendations,
agent_state=state,
),
)
@@ -0,0 +1,46 @@
"""SDK function-tool wrapper for the legacy ``load_skill`` tool.
The legacy implementation reaches into ``_agent_instances`` (a global
dict the legacy multi-agent orchestrator maintains) to find the running
``Agent`` instance and call ``agent.llm.add_skills(...)``. That global
goes away under the SDK migration — Phase 3 will replace it with a
context-keyed registry, and this wrapper will be updated to read from
that registry.
For Phase 2 we ship the wrapper as-is. The legacy function falls back
to a clean error path when the agent instance lookup fails, so the
tool degrades gracefully ("Could not find running agent instance...")
until Phase 3 lands. That's better than crashing or stubbing out the
tool entirely — the model still gets a structured error it can react to.
"""
from __future__ import annotations
import asyncio
import json
from typing import Any
from agents import RunContextWrapper
from strix.tools._decorator import strix_tool
from strix.tools._legacy_adapter import adapter_from_ctx
from strix.tools.load_skill import load_skill_actions as _legacy
def _dump(result: dict[str, Any]) -> str:
return json.dumps(result, ensure_ascii=False, default=str)
@strix_tool(timeout=60)
async def load_skill(ctx: RunContextWrapper, skills: str) -> str:
"""Load one or more named skills into this agent's prompt context.
Args:
skills: Comma-separated skill names (max 5). E.g.
``"recon,xss,sqli"``. Skill discovery uses
``strix.skills.parse_skill_list``.
"""
state = adapter_from_ctx(ctx)
return _dump(
await asyncio.to_thread(_legacy.load_skill, agent_state=state, skills=skills),
)
@@ -0,0 +1,90 @@
"""SDK function-tool wrapper for the legacy ``create_vulnerability_report``.
One tool. Local execution (``sandbox_execution=False`` in the legacy
registration). The legacy implementation handles XML parsing for the
CVSS breakdown and code locations, runs LLM-based dedup against
existing reports through ``strix.llm.dedupe.check_duplicate``, and
persists via ``get_global_tracer().add_vulnerability_report``.
We wrap the synchronous legacy function in ``asyncio.to_thread`` because
the dedup check makes a network call and we don't want to block the
event loop while it waits.
"""
from __future__ import annotations
import asyncio
import json
from typing import Any
from agents import RunContextWrapper
from strix.tools._decorator import strix_tool
from strix.tools.reporting import reporting_actions as _legacy
def _dump(result: dict[str, Any]) -> str:
return json.dumps(result, ensure_ascii=False, default=str)
# Generous timeout: the dedup check makes a separate LLM call, and large
# scans can have many existing reports to compare against.
@strix_tool(timeout=180)
async def create_vulnerability_report(
ctx: RunContextWrapper,
title: str,
description: str,
impact: str,
target: str,
technical_analysis: str,
poc_description: str,
poc_script_code: str,
remediation_steps: str,
cvss_breakdown: str,
endpoint: str | None = None,
method: str | None = None,
cve: str | None = None,
cwe: str | None = None,
code_locations: str | None = None,
) -> str:
"""File a vulnerability report against the active scan.
The report is dedup-checked against existing reports (LLM-based
similarity); if it's a near-duplicate, the call returns a
``duplicate_of`` pointer instead of creating a new entry.
Args:
title: Short headline (e.g. ``"Reflected XSS in /search?q="``).
description: What the vuln is.
impact: Concrete impact statement.
target: Affected URL / host / service.
technical_analysis: How it works.
poc_description: Reproduction summary.
poc_script_code: Working PoC (curl, python, etc.).
remediation_steps: Recommended fix.
cvss_breakdown: CVSS 3.1 vector parameters as XML (legacy schema).
endpoint: Optional endpoint path.
method: Optional HTTP method.
cve: Optional CVE identifier.
cwe: Optional CWE identifier.
code_locations: Optional XML list of file/line references.
"""
return _dump(
await asyncio.to_thread(
_legacy.create_vulnerability_report,
title=title,
description=description,
impact=impact,
target=target,
technical_analysis=technical_analysis,
poc_description=poc_description,
poc_script_code=poc_script_code,
remediation_steps=remediation_steps,
cvss_breakdown=cvss_breakdown,
endpoint=endpoint,
method=method,
cve=cve,
cwe=cwe,
code_locations=code_locations,
),
)
@@ -0,0 +1,42 @@
"""SDK function-tool wrapper for the legacy ``web_search`` tool.
The legacy ``web_search_actions.web_search`` is a synchronous Perplexity
API call (300s timeout, ``requests``). We wrap it with
``asyncio.to_thread`` so the call doesn't block the SDK event loop while
the API responds — same parity for the model, no surprises.
Pattern matches notes/todo/think wrappers from Phase 2.3.
"""
from __future__ import annotations
import asyncio
import json
from typing import Any
from agents import RunContextWrapper
from strix.tools._decorator import strix_tool
from strix.tools.web_search import web_search_actions as _legacy
def _dump(result: dict[str, Any]) -> str:
return json.dumps(result, ensure_ascii=False, default=str)
# Perplexity request timeout in the legacy code is 300s; give the SDK
# tool a slightly larger budget so the network round-trip + JSON decode
# doesn't push us over the edge under load.
@strix_tool(timeout=330)
async def web_search(ctx: RunContextWrapper, query: str) -> str:
"""Search the web with Perplexity, scoped to security-relevant content.
Returns a JSON-encoded ``{"success": bool, "content": str, ...}``
dict matching the legacy shape exactly.
Args:
query: The search query. The legacy tool prepends a security-focused
system prompt to bias results toward CVEs, exploits, and Kali-
compatible commands.
"""
return _dump(await asyncio.to_thread(_legacy.web_search, query=query))