diff --git a/src/auto_reverse/tools/__init__.py b/src/auto_reverse/tools/__init__.py new file mode 100644 index 0000000..b33f652 --- /dev/null +++ b/src/auto_reverse/tools/__init__.py @@ -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()] diff --git a/src/auto_reverse/tools/browser_tools.py b/src/auto_reverse/tools/browser_tools.py new file mode 100644 index 0000000..73f302f --- /dev/null +++ b/src/auto_reverse/tools/browser_tools.py @@ -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(), + ), + } diff --git a/src/auto_reverse/tools/doc_tools.py b/src/auto_reverse/tools/doc_tools.py new file mode 100644 index 0000000..793df9f --- /dev/null +++ b/src/auto_reverse/tools/doc_tools.py @@ -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, + ), + } diff --git a/src/auto_reverse/tools/flows_tools.py b/src/auto_reverse/tools/flows_tools.py new file mode 100644 index 0000000..8b0c7a6 --- /dev/null +++ b/src/auto_reverse/tools/flows_tools.py @@ -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, + ), + } diff --git a/tests/test_tools.py b/tests/test_tools.py new file mode 100644 index 0000000..29163bf --- /dev/null +++ b/tests/test_tools.py @@ -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"