Inspect and cache callable metadata for tool schemas.
| 31 | |
| 32 | |
| 33 | class FunctionCatalog: |
| 34 | """Inspect and cache callable metadata for tool schemas.""" |
| 35 | |
| 36 | def __init__(self, functions_dir: str | Path = FUNCTION_CALLING_DIR) -> None: |
| 37 | self._functions_dir = Path(functions_dir).resolve() |
| 38 | self._metadata: Dict[str, FunctionMetadata] = {} |
| 39 | self._loaded = False |
| 40 | self._load_error: Exception | None = None |
| 41 | self._module_index: Dict[str, List[str]] = {} |
| 42 | |
| 43 | def refresh(self) -> None: |
| 44 | """Reload metadata from the function directory.""" |
| 45 | self._metadata.clear() |
| 46 | self._module_index = {} |
| 47 | self._load_error = None |
| 48 | manager = get_function_manager(self._functions_dir) |
| 49 | try: |
| 50 | manager.load_functions() |
| 51 | except Exception as exc: # pragma: no cover - propagated via catalog usage |
| 52 | self._loaded = True |
| 53 | self._load_error = exc |
| 54 | return |
| 55 | |
| 56 | module_index: Dict[str, List[str]] = {} |
| 57 | for name, fn in manager.list_functions().items(): |
| 58 | try: |
| 59 | metadata = _build_function_metadata(name, fn, self._functions_dir) |
| 60 | self._metadata[name] = metadata |
| 61 | module_bucket = module_index.setdefault(metadata.module_name, []) |
| 62 | module_bucket.append(name) |
| 63 | except Exception as exc: # pragma: no cover - guarded to avoid cascading failures |
| 64 | print(f"[FunctionCatalog] Failed to load metadata for {name}: {exc}") |
| 65 | for module_name, names in module_index.items(): |
| 66 | names.sort() |
| 67 | self._module_index = module_index |
| 68 | self._loaded = True |
| 69 | |
| 70 | def _ensure_loaded(self) -> None: |
| 71 | if not self._loaded: |
| 72 | self.refresh() |
| 73 | |
| 74 | def get(self, name: str) -> FunctionMetadata | None: |
| 75 | self._ensure_loaded() |
| 76 | return self._metadata.get(name) |
| 77 | |
| 78 | def list_function_names(self) -> List[str]: |
| 79 | self._ensure_loaded() |
| 80 | return sorted(self._metadata.keys()) |
| 81 | |
| 82 | def list_metadata(self) -> Dict[str, FunctionMetadata]: |
| 83 | self._ensure_loaded() |
| 84 | return self._metadata.copy() |
| 85 | |
| 86 | def iter_modules(self) -> List[Tuple[str, List[FunctionMetadata]]]: |
| 87 | """Return functions grouped by Python file (module_name).""" |
| 88 | |
| 89 | self._ensure_loaded() |
| 90 | modules: List[Tuple[str, List[FunctionMetadata]]] = [] |
no outgoing calls
no test coverage detected