User: "What's the weather in Tokyo?" LLM: {"tool": "get_weather", "args": {"city": "Tokyo"}} System: [executes function, returns result] LLM: "It's 22°C and sunny in Tokyo." 

# tools.py import os import requests import json from typing import Dict, Any, List from abc import ABC, abstractmethod class Tool(ABC): @property @abstractmethod def name(self) -> str: ... @property @abstractmethod def description(self) -> str: ... @property @abstractmethod def parameters(self) -> Dict: ... @abstractmethod def execute(self, **kwargs) -> Any: ... # --- Concrete Tools ---  class WebSearchTool(Tool): name = "web_search" description = "Search the web for current information" parameters = { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"} }, "required": ["query"] } def execute(self, query: str) -> str: # Using DuckDuckGo HTML scrape (free, no API key)  url = "https://html.duckduckgo.com/html/" params = {"q": query} headers = {"User-Agent": "Mozilla/5.0"} try: response = requests.post(url, data=params, headers=headers, timeout=10) from bs4 import BeautifulSoup soup = BeautifulSoup(response.text, 'html.parser') results = soup.find_all('a', class_='result__snippet') snippets = [r.get_text()[:200] for r in results[:3]] return "\n".join(snippets) if snippets else "No results found" except Exception as e: return f"Search error: {e}" class PythonExecutorTool(Tool): name = "execute_python" description = "Execute Python code and return output" parameters = { "type": "object", "properties": { "code": {"type": "string", "description": "Python code to execute"} }, "required": ["code"] } def execute(self, code: str) -> str: import subprocess import tempfile import sys # Safety: only allow specific imports  allowed_imports = {'json', 'math', 'random', 'datetime', 'statistics', 'itertools', 'collections'} with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: f.write(code) temp_path = f.name try: result = subprocess.run( [sys.executable, temp_path], capture_output=True, text=True, timeout=10 ) os.unlink(temp_path) output = result.stdout if result.stderr: output += f"\nSTDERR: {result.stderr}" return output[:2000] if output else "(no output)" except subprocess.TimeoutExpired: os.unlink(temp_path) return "Error: Execution timeout (10s)" except Exception as e: return f"Error: {e}" class CalculatorTool(Tool): name = "calculate" description = "Perform mathematical calculations" parameters = { "type": "object", "properties": { "expression": {"type": "string", "description": "Math expression to evaluate"} }, "required": ["expression"] } def execute(self, expression: str) -> str: try: # Safe eval for math only  allowed_names = {"abs", "round", "min", "max", "sum", "pow"} allowed_names.update({k: v for k, v in __import__('math').__dict__.items() if not k.startswith('_')}) code = compile(expression, "<string>", "eval") for name in code.co_names: if name not in allowed_names: return f"Error: '{name}' not allowed" result = eval(code, {"__builtins__": {}}, allowed_names) return str(result) except Exception as e: return f"Error: {e}" class FileReadTool(Tool): name = "read_file" description = "Read contents of a file" parameters = { "type": "object", "properties": { "path": {"type": "string", "description": "File path to read"} }, "required": ["path"] } def execute(self, path: str) -> str: try: with open(path, 'r') as f: return f.read()[:5000] except Exception as e: return f"Error reading file: {e}" class FileWriteTool(Tool): name = "write_file" description = "Write content to a file" parameters = { "type": "object", "properties": { "path": {"type": "string", "description": "File path to write"}, "content": {"type": "string", "description": "Content to write"} }, "required": ["path", "content"] } def execute(self, path: str, content: str) -> str: try: os.makedirs(os.path.dirname(path) or '.', exist_ok=True) with open(path, 'w') as f: f.write(content) return f"File written: {path}" except Exception as e: return f"Error writing file: {e}" # Registry TOOLS = { "web_search": WebSearchTool(), "execute_python": PythonExecutorTool(), "calculate": CalculatorTool(), "read_file": FileReadTool(), "write_file": FileWriteTool() } def get_tool_schemas() -> List[Dict]: """Return OpenAI-compatible tool schemas.""" return [ { "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.parameters } } for tool in TOOLS.values() ] def execute_tool(name: str, args: Dict) -> str: if name not in TOOLS: return f"Error: Unknown tool '{name}'" try: return TOOLS[name].execute(**args) except Exception as e: return f"Tool error: {e}" 

# agent.py import os import json import requests from typing import List, Dict, Any, Optional from tools import get_tool_schemas, execute_tool class Agent: def __init__(self, provider: str = "openrouter", model: str = None): self.provider = provider self.configs = { "openrouter": { "url": "https://openrouter.ai/api/v1/chat/completions", "key": os.getenv("OPENROUTER_API_KEY"), "model": model or "google/gemini-2.0-flash-exp:free" }, "groq": { "url": "https://api.groq.com/openai/v1/chat/completions", "key": os.getenv("GROQ_API_KEY"), "model": model or "llama-3.1-70b-versatile" } } self.messages = [] self.tools = get_tool_schemas() self.max_turns = 10 def add_system_prompt(self, prompt: str): self.messages.insert(0, {"role": "system", "content": prompt}) def run(self, user_input: str) -> str: self.messages.append({"role": "user", "content": user_input}) for turn in range(self.max_turns): response = self._call_llm() if "tool_calls" in response: # Execute tools  tool_results = [] for tc in response["tool_calls"]: name = tc["function"]["name"] args = json.loads(tc["function"]["arguments"]) result = execute_tool(name, args) tool_results.append({ "tool_call_id": tc["id"], "role": "tool", "name": name, "content": result }) # Add tool results to conversation  self.messages.append({"role": "assistant", "tool_calls": response["tool_calls"]}) self.messages.extend(tool_results) else: # Final answer  content = response.get("content", "") self.messages.append({"role": "assistant", "content": content}) return content return "Max turns reached" def _call_llm(self) -> Dict: config = self.configs[self.provider] headers = { "Authorization": f"Bearer {config['key']}", "Content-Type": "application/json" } data = { "model": config["model"], "messages": self.messages, "tools": self.tools, "tool_choice": "auto", "temperature": 0.1, "max_tokens": 2000 } response = requests.post(config["url"], headers=headers, json=data, timeout=60) response.raise_for_status() return response.json()["choices"][0]["message"] def run_stream(self, user_input: str): """Streaming version.""" self.messages.append({"role": "user", "content": user_input}) for turn in range(self.max_turns): # First, check if we need tools (non-streaming for tool calls)  response = self._call_llm() if "tool_calls" in response: tool_results = [] for tc in response["tool_calls"]: name = tc["function"]["name"] args = json.loads(tc["function"]["arguments"]) result = execute_tool(name, args) tool_results.append({ "tool_call_id": tc["id"], "role": "tool...