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

{
    "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)

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)

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)

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

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)

# 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

# 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

{
    "severity": {
        "type": "string",
        "enum": ["low", "medium", "high", "critical"],
        "description": "Issue severity level"
    }
}

Nested Objects

{
    "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

{
    "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

# Good: One tool, one purpose
search_web
read_document
write_file
calculate

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

2. Idempotency Keys

{
    "name": "create_user",
    "parameters": {
        "email": {"type": "string"},
        "idempotency_key": {
            "type": "string",
            "description": "Unique key to prevent duplicate creation"
        }
    }
}

3. Side Effect Declaration

{
    "name": "send_email",
    "description": "Send an email",
    "parameters": {...},
    "x-side-effects": ["external_network", "cost_per_call:0.001"],
    "x-idempotent": false
}

4. Streaming Results

{
    "name": "stream_logs",
    "description": "Stream logs from service",
    "parameters": {
        "service": {"type": "string"},
        "follow": {"type": "boolean", "default": true}
    },
    "x-streaming": true
}

Error Handling

Tool Errors

# 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

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)

# 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

# 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

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

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

# 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

@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

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

# 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

MCP Protocol — Standardized tool protocol
Code Execution — Code execution tools
Browser Automation — Browser tools
API Integration — API tools
Custom Tools — Building custom tools
Tool Discovery — Dynamic tool discovery
OpenAI Agents SDK — Function calling in SDK
LangGraph — Tools in LangGraph
PydanticAI — Type-safe tools
Single Agent Patterns — ReAct, tool use patterns
Secure Coding — Tool security
Observability — 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.

From the Agentic Engineering research vault · CC BY 4.0 · all notes · raw markdown