1from dataclasses import dataclass, field
2from typing import Callable, Any
3from functools import wraps
4import inspect, json
5
6@dataclass
7class ToolSpec:
8 name: str
9 description: str
10 input_schema: dict
11 fn: Callable
12 version: str = "1.0.0"
13 tags: list[str] = field(default_factory=list)
14 requires_auth: bool = False
15
16class ToolRegistry:
17 def __init__(self):
18 self._tools: dict[str, ToolSpec] = {}
19
20 def register(self, description: str, tags: list[str] = None, requires_auth: bool = False):
21 def decorator(fn: Callable) -> Callable:
22 schema = self._infer_schema(fn)
23 spec = ToolSpec(
24 name=fn.__name__, description=description,
25 input_schema=schema, fn=fn,
26 tags=tags or [], requires_auth=requires_auth
27 )
28 self._tools[fn.__name__] = spec
29 @wraps(fn)
30 def wrapper(*args, **kwargs):
31 return fn(*args, **kwargs)
32 return wrapper
33 return decorator
34
35 def _infer_schema(self, fn: Callable) -> dict:
36 hints = fn.__annotations__
37 props = {}
38 for name, hint in hints.items():
39 if name == "return": continue
40 props[name] = {"type": "string" if hint == str else "number" if hint in (int, float) else "object"}
41 return {"type": "object", "properties": props, "required": list(props.keys())}
42
43 def get(self, name: str) -> ToolSpec | None:
44 return self._tools.get(name)
45
46 def list_all(self) -> list[dict]:
47 return [{"type": "function", "function": {"name": s.name, "description": s.description, "parameters": s.input_schema}} for s in self._tools.values()]
48
49 def filter_by_tag(self, tag: str) -> list[ToolSpec]:
50 return [s for s in self._tools.values() if tag in s.tags]
51
52registry = ToolRegistry()
53
54@registry.register("Search the web for real-time information", tags=["search", "web"])
55def web_search(query: str) -> str:
56 return f"Top results for: {query}"
57
58@registry.register("Read a local file's contents", tags=["file", "io"], requires_auth=True)
59def read_file(path: str) -> str:
60 with open(path) as f:
61 return f.read()
