Define Your Tools: Schemas, Permissions, and Safe Interfaces

You: "What's the weather in Tokyo right now?"

Without tools: "I don't have access to real-time weather data..."

With tools:
  LLM: I should call get_weather(location="Tokyo")
  Tool: {"temperature": 22, "condition": "Partly cloudy", "humidity": 68}
  LLM: "The current weather in Tokyo is 22°C (72°F) and partly cloudy with 68% humidity."

from openai import OpenAI
import json

client = OpenAI()

# Define your tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name, e.g. 'Tokyo' or 'New York, NY'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

# The actual implementation
def get_weather(location: str, unit: str = "celsius") -> dict:
    # In production: call a real weather API
    return {
        "location": location,
        "temperature": 22,
        "unit": unit,
        "condition": "partly cloudy"
    }

# Run the agent loop
def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        choice = response.choices[0]
        messages.append(choice.message)  # Add assistant's response
        
        # If no tool calls, we're done
        if choice.finish_reason == "stop":
            return choice.message.content
        
        # Process tool calls
        for tool_call in choice.message.tool_calls:
            function_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            
            # Call the actual function
            if function_name == "get_weather":
                result = get_weather(**arguments)
            else:
                result = {"error": f"Unknown function: {function_name}"}
            
            # Add tool result to messages
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

print(run_agent("What's the weather in Tokyo and London?"))

import anthropic
import json

client = anthropic.Anthropic()

# Define tools (slightly different schema)
tools = [
    {
        "name": "search_web",
        "description": "Search the web for current information",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Search query"
                },
                "num_results": {
                    "type": "integer",
                    "description": "Number of results to return",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    }
]

def search_web(query: str, num_results: int = 5) -> list:
    # In production: call a real search API (e.g., Brave Search, Serper)
    return [{"title": f"Result for: {query}", "url": "https://example.com", "snippet": "..."}]

def run_claude_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model="claude-opus-4-7",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )
        
        # Add response to messages
        messages.append({"role": "assistant", "content": response.content})
        
        # Check if we should stop
        if response.stop_reason == "end_turn":
            # Find the text block in the response
            for block in response.content:
                if hasattr(block, 'text'):
                    return block.text
        
        # Process tool use blocks
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                if block.name == "search_web":
                    result = search_web(**block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    })
        
        if tool_results:
            messages.append({"role": "user", "content": tool_results})

import requests

def search_web(query: str, num_results: int = 5) -> list[dict]:
    """Search the web using Brave Search API."""
    headers = {"X-Subscription-Token": BRAVE_API_KEY}
    params = {"q": query, "count": num_results}
    
    response = requests.get(
        "https://api.search.brave.com/res/v1/web/search",
        headers=headers,
        params=params
    )
    
    results = response.json().get("web", {}).get("results", [])
    return [
        {"title": r["title"], "url": r["url"], "snippet": r["description"]}
        for r in results
    ]

import subprocess
import tempfile
import os

def execute_python(code: str) -> dict:
    """Execute Python code safely and return output."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        temp_path = f.name
    
    try:
        result = subprocess.run(
            ["python", temp_path],
            capture_output=True,
            text=True,
            timeout=30  # Kill if takes more than 30 seconds
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "returncode": result.returncode
        }
    except subprocess.TimeoutExpired:
        return {"error": "Code execution timed out after 30 seconds"}
    finally:
        os.unlink(temp_path)

import os

def read_file(path: str) -> dict:
    """Read a file and return its contents."""
    try:
        with open(path, 'r', encoding='utf-8') as f:
            content = f.read()
        return {"success": True, "content": content, "size": len(content)}
    except FileNotFoundError:
        return {"success": False, "error": f"File not found: {path}"}
    except Exception as e:
        return {"success": False, "error": str(e)}

def write_file(path: str, content: str) -> dict:
    """Write content to a file."""
    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, 'w', encoding='utf-8') as f:
            f.write(content)
        return {"success": True, "bytes_written": len(content)}
    except Exception as e:
        return {"success": False, "error": str(e)}

# Too vague
{"name": "get_data", "description": "Gets some data"}

# Clear and specific
{"name": "get_stock_price", "description": "Get the current stock price for a publicly traded company. Use this when the user asks about a stock price, market cap, or financial data for a specific company."}

def safe_tool(arg: str) -> dict:
    try:
        result = do_work(arg)
        return {"success": True, "data": result}
    except Exception as e:
        return {"success": False, "error": str(e), "arg_received": arg}