# Function Calling

Structured tool invocation: schemas, providers, patterns, and best practices.

---

## Function Calling Fundamentals

### What is Function Calling?
LLMs can invoke structured functions by outputting a special format that the system parses and executes, returning results to the model.

### Core Flow
```
User Query → LLM → Function Call(s) → Execute → Results → LLM → Final Answer
```

### Function Definition
```python
{
    "name": "search_web",
    "description": "Search the web for current information. Use for recent events, facts, data.",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Search query, optimized for search engines"
            },
            "max_results": {
                "type": "integer",
                "description": "Maximum number of results to return",
                "default": 10,
                "minimum": 1,
                "maximum": 50
            },
            "recency_days": {
                "type": "integer",
                "description": "Limit results to last N days",
                "minimum": 1
            }
        },
        "required": ["query"],
        "additionalProperties": false
    }
}
```

---

## Provider Implementations

### OpenAI (Responses API + Chat Completions)
```python
from openai import AsyncOpenAI

client = AsyncOpenAI()

tools = [{
    "type": "function",
    "function": {
        "name": "search_web",
        "description": "Search the web...",
        "parameters": SEARCH_SCHEMA
    }
}]

response = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
    tools=tools,
    tool_choice="auto",  # auto, none, required, {"type": "function", "function": {"name": "search_web"}}
    parallel_tool_calls=True
)

# Handle tool calls
for tool_call in response.choices[0].message.tool_calls:
    if tool_call.function.name == "search_web":
        args = json.loads(tool_call.function.arguments)
        result = await search_web(**args)
        
        # Send result back
        followup = await client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "user", "content": "What's the weather in Tokyo?"},
                response.choices[0].message,
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result)
                }
            ]
        )
```

### Anthropic (Claude)
```python
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

tools = [{
    "name": "search_web",
    "description": "Search the web...",
    "input_schema": SEARCH_SCHEMA
}]

response = await client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=4096,
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
    tools=tools,
    tool_choice={"type": "auto"}  # auto, any, tool name, none
)

# Handle tool use
for block in response.content:
    if block.type == "tool_use":
        result = await search_web(**block.input)
        
        followup = await client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4096,
            messages=[
                {"role": "user", "content": "What's the weather in Tokyo?"},
                {"role": "assistant", "content": response.content},
                {
                    "role": "user",
                    "content": [{
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    }]
                }
            ]
        )
```

### Google (Gemini)
```python
import google.generativeai as genai

model = genai.GenerativeModel(
    model_name="gemini-1.5-pro",
    tools=[search_web_function]
)

chat = model.start_chat()

response = chat.send_message("What's the weather in Tokyo?")

# Handle function calls
for part in response.candidates[0].content.parts:
    if part.function_call:
        result = search_web(**dict(part.function_call.args))
        
        # Send result
        response = chat.send_message(
            genai.protos.Part(
                function_response=genai.protos.FunctionResponse(
                    name="search_web",
                    response={"result": result}
                )
            )
        )
```

### Mistral
```python
from mistralai.async_client import MistralAsyncClient

client = MistralAsyncClient(api_key="...")

tools = [{
    "type": "function",
    "function": {
        "name": "search_web",
        "description": "Search the web...",
        "parameters": SEARCH_SCHEMA
    }
}]

response = await client.chat(
    model="mistral-large-latest",
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
    tools=tools,
    tool_choice="auto"
)
```

### OpenAI-Compatible (vLLM, SGLang, TGI, Ollama)
```python
# All support OpenAI-compatible function calling
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="http://localhost:8000/v1",
    api_key="EMPTY"
)

# Same code as OpenAI
```

---

## Schema Design Best Practices

### Parameter Design
```python
# Good: Specific, constrained
{
    "query": {"type": "string", "description": "Search query"},
    "max_results": {"type": "integer", "minimum": 1, "maximum": 50, "default": 10},
    "recency_days": {"type": "integer", "minimum": 1, "maximum": 365}
}

# Bad: Vague, unconstrained
{
    "query": {"type": "string"},
    "options": {"type": "object"}  # Too vague!
}
```

### Enum for Fixed Values
```python
{
    "severity": {
        "type": "string",
        "enum": ["low", "medium", "high", "critical"],
        "description": "Issue severity level"
    }
}
```

### Nested Objects
```python
{
    "filter": {
        "type": "object",
        "properties": {
            "date_range": {
                "type": "object",
                "properties": {
                    "start": {"type": "string", "format": "date"},
                    "end": {"type": "string", "format": "date"}
                }
            },
            "tags": {"type": "array", "items": {"type": "string"}}
        }
    }
}
```

### Array of Objects
```python
{
    "contacts": {
        "type": "array",
        "items": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "email": {"type": "string", "format": "email"},
                "role": {"type": "string", "enum": ["owner", "admin", "member"]}
            },
            "required": ["name", "email"]
        }
    }
}
```

---

## Tool Design Patterns

### 1. Single Responsibility
```python
# Good: One tool, one purpose
search_web
read_document
write_file
calculate

# Bad: Kitchen sink
do_everything(action="search", params={...})
```

### 2. Idempotency Keys
```python
{
    "name": "create_user",
    "parameters": {
        "email": {"type": "string"},
        "idempotency_key": {
            "type": "string",
            "description": "Unique key to prevent duplicate creation"
        }
    }
}
```

### 3. Side Effect Declaration
```python
{
    "name": "send_email",
    "description": "Send an email",
    "parameters": {...},
    "x-side-effects": ["external_network", "cost_per_call:0.001"],
    "x-idempotent": false
}
```

### 4. Streaming Results
```python
{
    "name": "stream_logs",
    "description": "Stream logs from service",
    "parameters": {
        "service": {"type": "string"},
        "follow": {"type": "boolean", "default": true}
    },
    "x-streaming": true
}
```

---

## Error Handling

### Tool Errors
```python
# Return structured error
{
    "error": {
        "code": "RATE_LIMITED",
        "message": "Rate limit exceeded. Retry after 60 seconds.",
        "retry_after_seconds": 60,
        "details": {"limit": 100, "remaining": 0}
    }
}

# Or raise exception (framework handles)
raise ToolError(code="NOT_FOUND", message="User not found", retryable=False)
```

### Retry Logic
```python
class ToolExecutor:
    def __init__(self, max_retries=3, base_delay=1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    async def execute(self, tool, args):
        for attempt in range(self.max_retries + 1):
            try:
                return await tool.execute(args)
            except ToolError as e:
                if not e.retryable or attempt == self.max_retries:
                    raise
                delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
                await asyncio.sleep(delay)
```

---

## Parallel Tool Calls

### Automatic (OpenAI, SGLang)
```python
# Model automatically emits multiple tool_calls
response = await client.chat.completions.create(
    ...,
    parallel_tool_calls=True  # Default true
)

# Execute in parallel
async def execute_parallel(tool_calls):
    tasks = [execute_tool(tc) for tc in tool_calls]
    return await asyncio.gather(*tasks)
```

### Manual Parallel
```python
# Explicit parallel execution
async def execute_batch(tool_calls: list[ToolCall]) -> list[ToolResult]:
    semaphore = asyncio.Semaphore(10)  # Limit concurrency
    
    async def execute_one(tc):
        async with semaphore:
            return await execute_tool(tc)
    
    return await asyncio.gather(*[execute_one(tc) for tc in tool_calls])
```

---

## Security

### Input Validation
```python
def validate_tool_args(schema: dict, args: dict) -> ValidationResult:
    # 1. JSON Schema validation
    validator = Draft7Validator(schema)
    errors = list(validator.iter_errors(args))
    
    # 2. Custom validation
    # - Sanitize strings (no injection)
    # - Validate file paths (no traversal)
    # - Check URLs (allowlist)
    # - Validate IDs (format)
    
    return ValidationResult(valid=len(errors) == 0, errors=errors)
```

### Allowlists
```python
ALLOWED_DOMAINS = {"api.github.com", "api.openai.com", "api.anthropic.com"}
ALLOWED_COMMANDS = {"ls", "cat", "grep", "python", "node"}

def validate_external_call(url: str) -> bool:
    parsed = urlparse(url)
    return parsed.netloc in ALLOWED_DOMAINS
```

### Sandboxing
```python
# Always execute untrusted code in sandbox
async def execute_code(code: str) -> ExecutionResult:
    # Use: E2B, Modal, Daytona, Firecracker, gVisor, WASM
    # Never: eval(), exec(), subprocess directly
    return await sandbox.run(code)
```

---

## Testing Tools

### Unit Testing
```python
@pytest.fixture
def mock_search():
    with patch("tools.search_web") as mock:
        mock.return_value = [{"title": "Test", "url": "http://test.com", "snippet": "Test snippet"}]
        yield mock

async def test_agent_uses_search(agent, mock_search):
    result = await agent.run("Search for Python tutorials")
    mock_search.assert_called_once()
    assert "Python" in result
```

### Integration Testing
```python
async def test_tool_execution():
    # Test with real tool (use test doubles for external deps)
    result = await search_web.execute({"query": "test", "max_results": 1})
    assert len(result) > 0
    assert "title" in result[0]
    assert "url" in result[0]
```

### Contract Testing
```python
# Verify tool schema matches implementation
def test_tool_schema_matches_implementation():
    schema = get_tool_schema("search_web")
    sig = inspect.signature(search_web.execute)
    
    # Check required params match
    schema_required = set(schema["parameters"]["properties"].keys())
    sig_params = set(sig.parameters.keys())
    
    assert schema_required == sig_params
```

---

## Related Notes

- [[ToolUse/MCP-Protocol.md]] — Standardized tool protocol
- [[ToolUse/Code-Execution.md]] — Code execution tools
- [[ToolUse/Browser-Automation.md]] — Browser tools
- [[ToolUse/API-Integration.md]] — API tools
- [[ToolUse/Custom-Tools.md]] — Building custom tools
- [[ToolUse/Tool-Discovery.md]] — Dynamic tool discovery
- [[Frameworks/OpenAI-Agents-SDK.md]] — Function calling in SDK
- [[Frameworks/LangGraph.md]] — Tools in LangGraph
- [[Frameworks/PydanticAI.md]] — Type-safe tools
- [[Architectures/Single-Agent-Patterns.md]] — ReAct, tool use patterns
- [[Security/Secure-Coding.md]] — Tool security
- [[Production/Observability.md]] — Tool observability

---

*Sources: OpenAI function calling docs, Anthropic tool use docs, Gemini function calling, Mistral API, vLLM/SGLang/TGI function calling, JSON Schema spec, MCP spec, security best practices.*