Tool Use in AI Agents — The $12k Mistake We Made With Function Calling Schema Validation

import json
from typing import Any, Callable

# Minimal simulation of what the API does internally
def simulate_tool_call_parsing(model_output: str, tools: dict[str, dict]) -> list[dict]:
    """Simulate the parsing step that happens inside the API."""
    # In reality, the API does this parsing server-side.
    # Here we show what the model actually generates.
    try:
        # The model outputs a JSON object with name and arguments
        parsed = json.loads(model_output)
        if not isinstance(parsed, list):
            parsed = [parsed]  # handle single call
        
        calls = []
        for call in parsed:
            name = call.get('name')
            args_str = call.get('arguments', '{}')
            # The arguments are a JSON string, not a dict
            args = json.loads(args_str)
            calls.append({'name': name, 'arguments': args})
        return calls
    except json.JSONDecodeError as e:
        # This is what happens when the model hallucinates malformed JSON
        raise ValueError(f"Model output is not valid JSON: {e}")

# Example: what the model actually generates
def example_model_output():
    # This is what GPT-4 returns in the 'content' field when using tools
    return json.dumps([
        {
            "name": "get_weather",
            "arguments": json.dumps({"city": "London"})  # Note: arguments is a string!
        }
    ])

# The actual function you define
def get_weather(city: str) -> str:
    return f"The weather in {city} is sunny."

# Tool registry with validation
tools = {
    "get_weather": {
        "function": get_weather,
        "schema": {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Get the current weather for a city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "The city name, e.g. 'London'"
                        }
                    },
                    "required": ["city"]
                }
            }
        }
    }
}

# Simulate a full turn
def agent_turn(user_input: str, tools: dict) -> str:
    # Step 1: Model generates tool call (simulated)
    model_output = example_model_output()  # In reality, this comes from OpenAI API
    
    # Step 2: Parse the tool calls
    calls = simulate_tool_call_parsing(model_output, tools)
    
    # Step 3: Execute each tool
    results = []
    for call in calls:
        tool = tools.get(call['name'])
        if not tool:
            results.append(f"Error: Tool '{call['name']}' not found")
            continue
        try:
            result = tool['function'](**call['arguments'])
            results.append(result)
        except TypeError as e:
            # This is the validation failure we saw in the incident
            results.append(f"Error: Invalid arguments for {call['name']}: {e}")
    
    # Step 4: Return results as observations
    return json.dumps({"observations": results})

print(agent_turn("What's the weather in London?", tools))
# Output: {"observations": ["The weather in London is sunny."]}

import inspect
import json
import time
import logging
from typing import Any, Callable

logger = logging.getLogger(__name__)

class ToolRegistry:
    """Production-grade tool registry with validation, retries, and logging."""
    
    def __init__(self):
        self._tools: dict[str, dict] = {}
    
    def register(self, func: Callable, name: str = None, description: str = None):
        """Register a function as a tool. Generates schema from the function signature."""
        if name is None:
            name = func.__name__
        
        # Generate JSON schema from function signature
        sig = inspect.signature(func)
        properties = {}
        required = []
        for param_name, param in sig.parameters.items():
            # Map Python types to JSON schema types
            type_map = {
                str: "string",
                int: "integer",
                float: "number",
                bool: "boolean",
                list: "array",
                dict: "object"
            }
            json_type = type_map.get(param.annotation, "string")
            properties[param_name] = {
                "type": json_type,
                "description": f"The {param_name} parameter"
            }
            if param.default is inspect.Parameter.empty:
                required.append(param_name)
        
        schema = {
            "type": "function",
            "function": {
                "name": name,
                "description": description or func.__doc__ or f"Tool: {name}",
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required
                }
            }
        }
        
        self._tools[name] = {
            "function": func,
            "schema": schema,
            "signature": sig
        }
        logger.info(f"Registered tool: {name}")
    
    def validate_arguments(self, tool_name: str, arguments: dict) -> bool:
        """Validate that arguments match the function signature."""
        tool = self._tools.get(tool_name)
        if not tool:
            return False
        
        sig = tool["signature"]
        try:
            # This will raise TypeError if arguments don't match
            sig.bind(**arguments)
            return True
        except TypeError as e:
            logger.error(f"Argument validation failed for {tool_name}: {e}")
            return False
    
    def call(self, tool_name: str, arguments: dict, max_retries: int = 3) -> str:
        """Call a tool with retries and logging."""
        if not self.validate_arguments(tool_name, arguments):
            return json.dumps({"error": f"Invalid arguments for {tool_name}"})
        
        tool = self._tools[tool_name]
        func = tool["function"]
        
        for attempt in range(max_retries):
            try:
                start = time.time()
                result = func(**arguments)
                duration = time.time() - start
                logger.info(f"Tool {tool_name} succeeded in {duration:.2f}s")
                return json.dumps({"result": result, "duration": duration})
            except Exception as e:
                logger.warning(f"Tool {tool_name} failed attempt {attempt+1}: {e}")
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)  # exponential backoff
                else:
                    return json.dumps({"error": str(e)})
    
    def get_schemas(self) -> list[dict]:
        """Return all tool schemas for the API call."""
        return [t["schema"] for t in self._tools.values()]

# Example usage
registry = ToolRegistry()

def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    # In production, call a real weather API
    return f"The weather in {city} is sunny."

registry.register(get_weather)

# Simulate a tool call
result = registry.call("get_weather", {"city": "London"})
print(result)
# Output: {"result": "The weather in London is sunny.", "duration": 0.001}

# This would fail validation
bad_result = registry.call("get_weather", {"city": "London", "amount_usd": 100})
print(bad_result)
# Output: {"error": "Invalid arguments for get_weather"}

# Example: When NOT to use tool use - use a simple API instead

# Bad: Using an agent for a simple calculation
import time
from openai import OpenAI

client = OpenAI()

def bad_calculator_agent(expression: str) -> str:
    """Don't do this. It's slow, expensive, and unreliable."""
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a calculator. Use the calculate tool."},
            {"role": "user", "content": expression}
        ],
        tools=[{
            "type": "function",
            "function": {
                "name": "calculate",
                "description": "Evaluate a mathematical expression",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "expression": {"type": "string"}
                    },
                    "required": ["expression"]
                }
            }
        }]
    )
    # Parse the tool call...
    return "Result: 4"  # Simplified

# Good: Just use eval()
def good_calculator(expression: str) -> float:
    """Simple, fast, deterministic. No agent needed."""
    # Use ast.literal_eval for safety in production
    return eval(expression)  # Note: use ast.literal_eval in real code

# Benchmark
start = time.time()
result = good_calculator("2 + 2")
print(f"Simple: {result} in {time.time() - start:.4f}s")
# Output: Simple: 4 in 0.0001s

# The agent version would take 2-3 seconds and cost $0.01 per call

import asyncio
import json
import time
from typing import Any
from openai import AsyncOpenAI

client = AsyncOpenAI()

class ParallelToolExecutor:
    """Executes multiple tool calls concurrently with rate limiting."""
    
    def __init__(self, registry, max_concurrent: int = 5):
        self.registry = registry
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def execute_one(self, tool_name: str, arguments: dict) -> dict:
        """Execute a single tool call with rate limiting."""
        async with self.semaphore:
            # Call the tool (synchronous in this example, but can be async)
            result = self.registry.call(tool_name, arguments)
            return {
                "role": "tool",
                "tool_call_id": tool_name,  # In production, use the actual ID from the API
                "content": result
            }
    
    async def execute_all(self, tool_calls: list[dict]) -> list[dict]:
        """Execute all tool calls in parallel."""
        tasks = []
        for call in tool_calls:
            task = self.execute_one(call["name"], json.loads(call["arguments"]))
            tasks.append(task)
        return await asyncio.gather(*tasks)

# Production agent loop
async def agent_loop(user_input: str, registry, max_turns: int = 10):
    """Full agent loop with parallel tool execution."""
    messages = [
        {"role": "system", "content": "You are a helpful assistant with access to tools."},
        {"role": "user", "content": user_input}
    ]
    
    executor = ParallelToolExecutor(registry)
    
    for turn in range(max_turns):
        response = await client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=registry.get_schemas(),
            parallel_tool_calls=True
        )
        
        message = response.choices[0].message
        
        # Check if the model wants to call tools
        if message.tool_calls:
            # Add the assistant's message with tool calls
            messages.append(message.to_dict())
            
            # Execute all tool calls in parallel
            tool_results = await executor.execute_all(
                [call.to_dict() for call in message.tool_calls]
            )
            
            # Add all tool results to messages
            messages.extend(tool_results)
        else:
            # Model wants to respond directly
            messages.append({"role": "assistant", "content": message.content})
            break
    
    return messages[-1]["content"]

# Example usage (requires async context)
# result = await agent_loop("What's the weather in London and Paris?", registry)
# print(result)

# Mistake 1: Using **kwargs (silent failure)
def bad_tool(**kwargs):
    """This tool silently ignores bad arguments."""
    # The model might pass 'amount_usd' but we ignore it
    return "default result"

# Fix: Explicit parameters
def good_tool(transaction_id: str, amount: float):
    """This tool will crash if the model passes wrong arguments."""
    return f"Processing transaction {transaction_id} for ${amount}"

# Mistake 2: Not handling errors
def fragile_tool():
    # If this API call fails, the agent crashes
    import requests
    response = requests.get("https://api.example.com/data")
    return response.json()

# Fix: Wrap in try-except
def robust_tool():
    import requests
    try:
        response = requests.get("https://api.example.com/data", timeout=5)
        return response.json()
    except Exception as e:
        return {"error": str(e)}

# Mistake 3: No max turns
# The agent loop runs forever if the model keeps calling tools
# Fix: Add a counter
MAX_TURNS = 10
for turn in range(MAX_TURNS):
    # ... agent logic
    pass
else:
    # Return a fallback response
    print("Agent exceeded max turns. Returning fallback.")

# Mistake 4: Vague tool description
# Bad:"search" -> model doesn't know when to use it
# Good: "Search the web for information. Use this when the user asks about current events, news, or general knowledge. Example: 'What is the capital of France?' maps to search(query='capital of France')."

# Comparison: Same task done with three different patterns

# 1. Tool Use (Function Calling)
import json

def search_web_tool(query: str) -> str:
    """Search the web."""
    return f"Results for {query}: ..."

# The model generates: {"name": "search_web_tool", "arguments": "{\"query\": \"Python 3.13\"}"}
# Your code parses and executes

# 2. Code Execution (Code Interpreter)
# The model generates:
# import requests
# response = requests.get('https://api.duckduckgo.com/?q=Python+3.13&format=json')
# print(response.json())
# This runs in a sandbox. More flexible but riskier.

# 3. Direct API Call
import requests

def search_web_direct(query: str) -> str:
    """No model involved. Just call the API directly."""
    response = requests.get(f"https://api.example.com/search?q={query}")
    return response.json()

# Which to choose?
# - If the user says "search for Python 3.13 news", use direct API (fast, cheap)
# - If the user says "find the latest Python news and summarize them", use tool use (needs reasoning)
# - If the user says "analyze this CSV and plot the trend", use code execution (complex computation)

import json
import logging
import time
import uuid
from typing import Any

# Structured logging setup
logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger(__name__)

class AgentLogger:
    """Logs every step of the agent for debugging."""
    
    def __init__(self, trace_id: str = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.logs = []
    
    def log(self, event: str, data: dict):
        """Log an event with structured data."""
        entry = {
            "trace_id": self.trace_id,
            "timestamp": time.time(),
            "event": event,
            "data": data
        }
        self.logs.append(entry)
        logger.info(json.dumps(entry))
    
    def log_tool_call(self, tool_name: str, arguments: dict, result: Any, duration: float):
        self.log("tool_call", {
            "tool": tool_name,
            "arguments": arguments,
            "result": result,
            "duration": duration
        })
    
    def log_model_response(self, response: dict):
        self.log("model_response", {
            "model": response.get("model"),
            "usage": response.get("usage"),
            "choices": response.get("choices", [])
        })
    
    def get_summary(self) -> dict:
        """Return a summary of the agent run."""
        tool_calls = [l for l in self.logs if l["event"] == "tool_call"]
        return {
            "trace_id": self.trace_id,
            "total_tool_calls": len(tool_calls),
            "total_duration": sum(l["data"]["duration"] for l in tool_calls),
            "errors": [l for l in self.logs if "error" in l["data"]]
        }

# Usage in production
logger = AgentLogger()
start = time.time()
result = registry.call("get_weather", {"city": "London"})
duration = time.time() - start
logger.log_tool_call("get_weather", {"city": "London"}, result, duration)

# Later, you can replay the session
print(json.dumps(logger.get_summary(), indent=2))
# Output: {"trace_id": "...", "total_tool_calls": 1, "total_duration": 0.5, "errors": []}

from typing import Any, Callable, Dict
from pydantic import BaseModel, Field

class Tool(BaseModel):
    name: str
    description: str
    input_schema: dict
    fn: Callable

class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, Tool] = {}

    def register(self, tool: Tool):
        self._tools[tool.name] = tool

    def get_metadata(self) -> list[dict]:
        return [
            {"name": t.name, "description": t.description, "parameters": t.input_schema}
            for t in self._tools.values()
        ]

    def execute(self, name: str, **kwargs) -> Any:
        if name not in self._tools:
            raise ValueError(f"Tool '{name}' not found")
        return self._tools[name].fn(**kwargs)

# Usage
registry = ToolRegistry()
registry.register(
    Tool(
        name="get_weather",
        description="Fetch current weather for a city",
        input_schema={"type": "object", "properties": {"city": {"type": "string"}}},
        fn=lambda city: f"Sunny, 22°C in {city}"
    )
)
print(registry.get_metadata())

import json

def react_agent(prompt: str, registry: ToolRegistry, max_steps: int = 5):
    scratchpad = f"User: {prompt}\n"
    for step in range(max_steps):
        # Simulate LLM call (in prod, use OpenAI/Claude API)
        llm_response = f"Thought: I need to find the weather.\nAction: get_weather\nAction Input: {json.dumps({'city': 'Paris'})}"
        print(f"Step {step}: {llm_response}")
        
        # Parse action from LLM output
        if "Action:" in llm_response:
            tool_name = llm_response.split("Action:")[1].split("\n")[0].strip()
            input_str = llm_response.split("Action Input:")[1].split("\n")[0].strip()
            result = registry.execute(tool_name, **json.loads(input_str))
            scratchpad += f"Observation: {result}\n"
            print(f"Observation: {result}")
        else:
            print("Final answer:", llm_response)
            break

# Example run
react_agent("What's the weather in Paris?", registry)