tanyar09
096d76430b
- Add timeout protection (120s) for LLM provider calls - Skip memory consolidation for CLI mode to avoid blocking - Add timeout protection for memory consolidation (120s) - Improve error handling with better logging - Add parameter type coercion before validation - Allow None values for optional parameters in validation - Fix type coercion for memory updates (handle dict responses)
76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
"""Tool registry for dynamic tool management."""
|
|
|
|
from typing import Any
|
|
|
|
from nanobot.agent.tools.base import Tool
|
|
|
|
|
|
class ToolRegistry:
|
|
"""
|
|
Registry for agent tools.
|
|
|
|
Allows dynamic registration and execution of tools.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._tools: dict[str, Tool] = {}
|
|
|
|
def register(self, tool: Tool) -> None:
|
|
"""Register a tool."""
|
|
self._tools[tool.name] = tool
|
|
|
|
def unregister(self, name: str) -> None:
|
|
"""Unregister a tool by name."""
|
|
self._tools.pop(name, None)
|
|
|
|
def get(self, name: str) -> Tool | None:
|
|
"""Get a tool by name."""
|
|
return self._tools.get(name)
|
|
|
|
def has(self, name: str) -> bool:
|
|
"""Check if a tool is registered."""
|
|
return name in self._tools
|
|
|
|
def get_definitions(self) -> list[dict[str, Any]]:
|
|
"""Get all tool definitions in OpenAI format."""
|
|
return [tool.to_schema() for tool in self._tools.values()]
|
|
|
|
async def execute(self, name: str, params: dict[str, Any]) -> str:
|
|
"""
|
|
Execute a tool by name with given parameters.
|
|
|
|
Args:
|
|
name: Tool name.
|
|
params: Tool parameters.
|
|
|
|
Returns:
|
|
Tool execution result as string.
|
|
|
|
Raises:
|
|
KeyError: If tool not found.
|
|
"""
|
|
tool = self._tools.get(name)
|
|
if not tool:
|
|
return f"Error: Tool '{name}' not found"
|
|
|
|
try:
|
|
# Coerce parameter types before validation
|
|
coerced_params = tool.coerce_params(params)
|
|
errors = tool.validate_params(coerced_params)
|
|
if errors:
|
|
return f"Error: Invalid parameters for tool '{name}': " + "; ".join(errors)
|
|
return await tool.execute(**coerced_params)
|
|
except Exception as e:
|
|
return f"Error executing {name}: {str(e)}"
|
|
|
|
@property
|
|
def tool_names(self) -> list[str]:
|
|
"""Get list of registered tool names."""
|
|
return list(self._tools.keys())
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._tools)
|
|
|
|
def __contains__(self, name: str) -> bool:
|
|
return name in self._tools
|