Markov
5417980b76
- Implemented ReAct loop in agent.py - Added Anthropic provider using httpx (no SDK) - Created tools: read, write, edit, bash - Added session management with JSONL format - Included configuration system with env var support - Added CLI interface and example usage - Minimal dependencies: only httpx
62 lines
1.8 KiB
Python
62 lines
1.8 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_anthropic_format(self) -> Dict[str, Any]:
|
|
"""Convert tool to Anthropic API format"""
|
|
return {
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": self.parameters.get("properties", {}),
|
|
"required": self.parameters.get("required", [])
|
|
}
|
|
}
|
|
|
|
|
|
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_anthropic_tools(self) -> List[Dict[str, Any]]:
|
|
"""Get all tools in Anthropic API format"""
|
|
return [tool.to_anthropic_format() for tool in self.tools.values()]
|
|
|
|
async def execute_tool(self, name: str, args: Dict[str, Any]) -> str:
|
|
"""Execute a tool by name with arguments"""
|
|
tool = self.get_tool(name)
|
|
return await tool.execute(args) |