Markov
b096096f93
- Implemented full ReAct loop in agent.py with 20 iteration limit - Added Anthropic provider using httpx (no SDK dependency) - Created complete tool system: read, write, edit, bash - Added session management with JSONL format - Updated README with usage examples - Fixed tool registry imports and methods - All core functionality working
77 lines
2.2 KiB
Python
77 lines
2.2 KiB
Python
"""
|
|
Tool registry for managing available tools
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Any, List, Callable
|
|
from abc import ABC, abstractmethod
|
|
|
|
|
|
@dataclass
|
|
class Tool(ABC):
|
|
"""Base tool class"""
|
|
name: str
|
|
description: str
|
|
parameters: Dict[str, Any] # JSON Schema
|
|
|
|
@abstractmethod
|
|
async def execute(self, args: Dict[str, Any]) -> str:
|
|
"""Execute the tool with given arguments"""
|
|
pass
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert tool to dictionary for API"""
|
|
return {
|
|
'name': self.name,
|
|
'description': self.description,
|
|
'input_schema': {
|
|
'type': 'object',
|
|
'properties': self.parameters.get('properties', {}),
|
|
'required': self.parameters.get('required', [])
|
|
}
|
|
}
|
|
|
|
def to_anthropic_format(self) -> Dict[str, Any]:
|
|
"""Convert tool to Anthropic API format (alias)"""
|
|
return self.to_dict()
|
|
|
|
|
|
class ToolRegistry:
|
|
"""Registry for managing available tools"""
|
|
|
|
def __init__(self):
|
|
self.tools: Dict[str, Tool] = {}
|
|
|
|
def register(self, tool: Tool):
|
|
"""Register a tool"""
|
|
self.tools[tool.name] = tool
|
|
|
|
def get_tool(self, name: str) -> Tool:
|
|
"""Get a tool by name"""
|
|
if name not in self.tools:
|
|
raise ValueError(f"Tool '{name}' not found")
|
|
return self.tools[name]
|
|
|
|
def get_all_tools(self) -> List[Tool]:
|
|
"""Get all registered tools"""
|
|
return list(self.tools.values())
|
|
|
|
def get_tool_definitions(self) -> List[Dict[str, Any]]:
|
|
"""Get tool definitions for API"""
|
|
return [tool.to_dict() for tool in self.tools.values()]
|
|
|
|
def get_anthropic_tools(self) -> List[Dict[str, Any]]:
|
|
"""Get all tools in Anthropic API format (alias)"""
|
|
return self.get_tool_definitions()
|
|
|
|
async def execute_tool(self, name: str, args: Dict[str, Any]) -> str:
|
|
"""Execute a tool by name with arguments"""
|
|
try:
|
|
tool = self.get_tool(name)
|
|
return await tool.execute(args)
|
|
except Exception as e:
|
|
return f"Error executing {name}: {str(e)}"
|
|
|
|
|
|
# Global registry instance
|
|
registry = ToolRegistry() |