1+ """Dynamic tool registry.
2+
3+ Tools are defined by YAML spec files and backed by Python callables.
4+ Each tool spec looks like:
5+
6+ ```yaml
7+ name: echo
8+ description: Echo the user input
9+ python: mini_agent_harness.tools.echo:echo_tool
10+ ```
11+
12+ At runtime we import the callable and expose it via `load_tools`.
13+ """
14+ from __future__ import annotations
15+
16+ import importlib
17+ import inspect
18+ from dataclasses import dataclass
19+ from pathlib import Path
20+ from typing import Callable , Dict , Iterable , List
21+
22+ import yaml
23+
24+
25+ @dataclass
26+ class ToolSpec :
27+ name : str
28+ description : str
29+ python : str # module:path e.g. path.to.mod:function_name
30+ func : Callable | None = None # populated after import
31+
32+ def import_func (self ) -> Callable :
33+ if self .func is not None :
34+ return self .func
35+ module_path , _ , attr = self .python .partition (":" )
36+ if not module_path or not attr :
37+ raise ValueError (f"Invalid python path in tool spec: { self .python } " )
38+ module = importlib .import_module (module_path )
39+ func : Callable = getattr (module , attr )
40+ if not callable (func ):
41+ raise TypeError (f"Tool target { self .python } is not callable" )
42+ # Basic signature check: first param should be str or none
43+ sig = inspect .signature (func )
44+ if len (sig .parameters ) == 0 :
45+ raise TypeError ("Tool callable must accept at least one argument (input string)" )
46+ self .func = func
47+ return func
48+
49+
50+ def _load_tool_yaml (path : Path ) -> ToolSpec :
51+ data = yaml .safe_load (path .read_text ())
52+ return ToolSpec (** data )
53+
54+
55+ def discover_tools (tool_paths : Iterable [str | Path ]) -> Dict [str , ToolSpec ]:
56+ """Load multiple tool YAMLs into a registry keyed by tool name."""
57+ registry : Dict [str , ToolSpec ] = {}
58+ for p in tool_paths :
59+ spec = _load_tool_yaml (Path (p ))
60+ if spec .name in registry :
61+ raise ValueError (f"Duplicate tool name { spec .name } " )
62+ registry [spec .name ] = spec
63+ return registry
64+
65+
66+ def load_tools_from_manifest (manifest : dict ) -> Dict [str , Callable ]:
67+ """Given the agent manifest dict, import and return callable tools."""
68+ tools_field : List [str ] = manifest .get ("tools" , [])
69+ registry = discover_tools (tools_field )
70+ return {name : spec .import_func () for name , spec in registry .items ()}
71+
72+
73+ __all__ = [
74+ "ToolSpec" ,
75+ "discover_tools" ,
76+ "load_tools_from_manifest" ,
77+ ]
0 commit comments