feat: agent tool definitions (browser, flows, doc) and registry

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-06-01 00:14:37 +08:00
parent 9161f623a7
commit 6b1ed67d4a
5 changed files with 228 additions and 0 deletions
+27
View File
@@ -0,0 +1,27 @@
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from auto_reverse.tools.browser_tools import browser_tools
from auto_reverse.tools.doc_tools import doc_tools
from auto_reverse.tools.flows_tools import flows_tools
if TYPE_CHECKING:
from auto_reverse.doc.engine import DocEngine
from auto_reverse.store import FlowStore
Handler = Callable[[dict[str, Any]], dict[str, Any]]
Registry = dict[str, tuple[dict[str, Any], Handler]]
def build_registry(browser: Any, store: FlowStore, engine: DocEngine) -> Registry:
registry: Registry = {}
registry.update(browser_tools(browser))
registry.update(flows_tools(store))
registry.update(doc_tools(store, engine))
return registry
def tool_schemas(registry: Registry) -> list[dict[str, Any]]:
return [schema for schema, _ in registry.values()]
+58
View File
@@ -0,0 +1,58 @@
from __future__ import annotations
from collections.abc import Callable
from typing import Any
Handler = Callable[[dict[str, Any]], dict[str, Any]]
def browser_tools(browser: Any) -> dict[str, tuple[dict[str, Any], Handler]]:
return {
"browser_navigate": (
{
"name": "browser_navigate",
"description": "Navigate the browser to a URL and return a page snapshot.",
"input_schema": {
"type": "object",
"properties": {"url": {"type": "string"}},
"required": ["url"],
},
},
lambda inp: browser.navigate(inp["url"]),
),
"browser_click": (
{
"name": "browser_click",
"description": "Click an element by CSS selector; returns a new snapshot.",
"input_schema": {
"type": "object",
"properties": {"selector": {"type": "string"}},
"required": ["selector"],
},
},
lambda inp: browser.click(inp["selector"]),
),
"browser_type": (
{
"name": "browser_type",
"description": "Fill a form field (CSS selector) with text.",
"input_schema": {
"type": "object",
"properties": {
"selector": {"type": "string"},
"text": {"type": "string"},
},
"required": ["selector", "text"],
},
},
lambda inp: browser.type_text(inp["selector"], inp["text"]),
),
"browser_snapshot": (
{
"name": "browser_snapshot",
"description": "Return the current page snapshot without acting.",
"input_schema": {"type": "object", "properties": {}},
},
lambda inp: browser.snapshot(),
),
}
+49
View File
@@ -0,0 +1,49 @@
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from auto_reverse.doc.engine import DocEngine
from auto_reverse.store import FlowStore
Handler = Callable[[dict[str, Any]], dict[str, Any]]
def doc_tools(store: FlowStore, engine: DocEngine) -> dict[str, tuple[dict[str, Any], Handler]]:
def document(inp: dict[str, Any]) -> dict[str, Any]:
path = inp["path_template"]
for rec in store.endpoints():
if rec.signature.path_template == path:
if inp.get("summary"):
rec.summary = inp["summary"]
if inp.get("description"):
rec.description = inp["description"]
if inp.get("tag"):
rec.tag = inp["tag"]
engine.document(rec.signature)
return {"documented": path}
return {"error": f"no endpoint matching {path}"}
return {
"doc_document": (
{
"name": "doc_document",
"description": (
"Enrich and (re)write docs for an endpoint by path template, "
"optionally setting a human summary/description/tag."
),
"input_schema": {
"type": "object",
"properties": {
"path_template": {"type": "string"},
"summary": {"type": "string"},
"description": {"type": "string"},
"tag": {"type": "string"},
},
"required": ["path_template"],
},
},
document,
),
}
+41
View File
@@ -0,0 +1,41 @@
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from auto_reverse.models import EndpointRecord
from auto_reverse.store import FlowStore
Handler = Callable[[dict[str, Any]], dict[str, Any]]
def _record_view(rec: EndpointRecord) -> dict[str, Any]:
return {
"method": rec.signature.method,
"path": rec.signature.path_template,
"status": rec.signature.status_class,
"samples": rec.sample_count,
"documented": rec.documented,
}
def flows_tools(store: FlowStore) -> dict[str, tuple[dict[str, Any], Handler]]:
def search(inp: dict[str, Any]) -> dict[str, Any]:
query = inp.get("query", "")
records = store.search(query) if query else store.endpoints()
return {"endpoints": [_record_view(r) for r in records]}
return {
"flows_search": (
{
"name": "flows_search",
"description": "List/search discovered API endpoints captured so far.",
"input_schema": {
"type": "object",
"properties": {"query": {"type": "string"}},
},
},
search,
),
}
+53
View File
@@ -0,0 +1,53 @@
from auto_reverse.doc.engine import DocEngine
from auto_reverse.models import CapturedFlow
from auto_reverse.store import FlowStore, ScopeFilter
from auto_reverse.tools import build_registry
class FakeBrowser:
def navigate(self, url):
return {"url": url, "title": "T", "elements": []}
def click(self, selector):
return {"url": "u", "title": "T", "elements": []}
def type_text(self, selector, text):
return {"url": "u", "title": "T", "elements": []}
def snapshot(self):
return {"url": "u", "title": "T", "elements": []}
def _store_with_endpoint(tmp_path):
store = FlowStore(ScopeFilter(target_hosts={"ex.com"}))
store.ingest(CapturedFlow(
method="GET", host="ex.com", path="/api/users", query={}, req_headers={},
req_body=None, status=200,
resp_headers={"content-type": "application/json"}, resp_body=b"[]",
timestamp=0.0,
))
engine = DocEngine(store, out_dir=tmp_path, title="x", use_llm=False)
return store, engine
def test_registry_has_expected_tools(tmp_path):
store, engine = _store_with_endpoint(tmp_path)
reg = build_registry(FakeBrowser(), store, engine)
names = {schema["name"] for schema, _ in reg.values()}
assert {"browser_navigate", "browser_click", "flows_search", "doc_document"} <= names
def test_flows_search_handler_returns_matches(tmp_path):
store, engine = _store_with_endpoint(tmp_path)
reg = build_registry(FakeBrowser(), store, engine)
_, handler = reg["flows_search"]
result = handler({"query": "users"})
assert any("/api/users" in ep["path"] for ep in result["endpoints"])
def test_browser_navigate_handler(tmp_path):
store, engine = _store_with_endpoint(tmp_path)
reg = build_registry(FakeBrowser(), store, engine)
_, handler = reg["browser_navigate"]
result = handler({"url": "http://x"})
assert result["url"] == "http://x"