AI Tool Use and Function Calling Complete Guide 2026

Introduction

The ability to use tools and call functions represents one of the most significant capabilities of modern AI systems. Function calling enables AI models to interact with external systems, access real-time information, and perform actions beyond their training data. This comprehensive guide covers everything you need to know about implementing tool use and function calling in AI applications.

Understanding Function Calling

What is Function Calling?

Function calling is a mechanism that allows AI models to invoke external functions or APIs to accomplish tasks. Instead of relying solely on knowledge from training data, AI systems can:

Query databases for real-time information
Call external APIs to fetch data
Perform calculations and computations
Interact with third-party services
Execute code in sandboxed environments

How Function Calling Works

The function calling process follows a structured flow:

User Query → AI Model → Identify Need → Generate Parameters → 
Execute Function → Return Results → Generate Response → User

Function Calling Architecture

Core Components

A function calling system consists of several key components:

Component	Description
Function Registry	Central catalog of available tools
Parameter Generator	Creates valid parameters for function calls
Executor	Runs the actual function
Result Processor	Formats output for the model
Error Handler	Manages failures gracefully

Function Definition Schema

Functions are defined using JSON schemas that describe the function’s interface:

function_schema = {
    "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., 'San Francisco, CA'"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit",
                    "default": "fahrenheit"
                }
            },
            "required": ["location"]
        }
    }
}

Implementing Function Calling

OpenAI Function Calling

import openai
import json
from typing import Optional, List

class FunctionCaller:
    def __init__(self, client: openai.OpenAI):
        self.client = client
        self.functions: dict = {}
    
    def register_function(self, name: str, func: callable, description: str, parameters: dict):
        """Register a function for AI calling."""
        self.functions[name] = {
            "function": func,
            "description": description,
            "parameters": parameters
        }
    
    def get_schema(self, name: str) -> dict:
        """Get the schema for a registered function."""
        func = self.functions.get(name)
        if not func:
            raise ValueError(f"Function {name} not registered")
        
        return {
            "type": "function",
            "function": {
                "name": name,
                "description": func["description"],
                "parameters": func["parameters"]
            }
        }
    
    def call_with_functions(self, messages: List[dict], 
                          function_names: Optional[List[str]] = None) -> dict:
        """Call the model with available functions."""
        tools = []
        
        if function_names:
            for name in function_names:
                tools.append(self.get_schema(name))
        else:
            # Use all registered functions
            for name in self.functions:
                tools.append(self.get_schema(name))
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        return response

# Example usage
def get_weather(location: str, unit: str = "fahrenheit") -> dict:
    """Get weather for a location."""
    # Call weather API
    return {
        "location": location,
        "temperature": 72,
        "condition": "Sunny",
        "unit": unit
    }

caller = FunctionCaller(openai.OpenAI())
caller.register_function(
    name="get_weather",
    func=get_weather,
    description="Get the current weather for a location",
    parameters={
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "City name"
            },
            "unit": {
                "type": "string",
                "enum": ["celsius", "fahrenheit"],
                "default": "fahrenheit"
            }
        },
        "required": ["location"]
    }
)

messages = [{"role": "user", "content": "What's the weather in New York?"}]
response = caller.call_with_functions(messages)

Anthropic Function Calling

import anthropic

class AnthropicFunctionCaller:
    def __init__(self, client: anthropic.Anthropic):
        self.client = client
        self.tools: List[dict] = []
    
    def add_tool(self, name: str, description: str, 
                input_schema: dict):
        """Add a tool definition."""
        self.tools.append({
            "name": name,
            "description": description,
            "input_schema": input_schema
        })
    
    def call(self, messages: List[dict], 
             max_tokens: int = 1024) -> dict:
        """Call Anthropic with tools."""
        response = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            messages=messages,
            tools=self.tools,
            max_tokens=max_tokens
        )
        
        # Check for tool use
        if response.stop_reason == "tool_use":
            tool_use = response.content[-1]
            return {
                "tool_use": {
                    "name": tool_use.name,
                    "input": tool_use.input
                }
            }
        
        return {"text": response.content[0].text}

Tool Execution Patterns

Synchronous Execution

For simple, fast functions:

from concurrent.futures import ThreadPoolExecutor
from typing import Any, Dict
import json

class ToolExecutor:
    def __init__(self):
        self.tools: Dict[str, callable] = {}
        self.executor = ThreadPoolExecutor(max_workers=10)
    
    def register(self, name: str, func: callable):
        """Register a synchronous tool."""
        self.tools[name] = func
    
    def execute(self, name: str, arguments: Dict[str, Any]) -> Any:
        """Execute a tool with given arguments."""
        if name not in self.tools:
            raise ValueError(f"Unknown tool: {name}")
        
        tool = self.tools[name]
        
        # Validate required arguments
        required = getattr(tool, "_required_args", [])
        missing = [arg for arg in required if arg not in arguments]
        
        if missing:
            raise ValueError(f"Missing required arguments: {missing}")
        
        # Execute
        try:
            result = tool(**arguments)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def execute_with_retry(self, name: str, arguments: Dict,
                          max_retries: int = 3) -> Any:
        """Execute with automatic retry."""
        import time
        
        for attempt in range(max_retries):
            result = self.execute(name, arguments)
            
            if result["success"]:
                return result
            
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                time.sleep(wait_time)
        
        return result


# Example: Register tool with metadata
def get_stock_price(symbol: str) -> dict:
    """Get current stock price."""
    # Implementation
    return {"symbol": symbol, "price": 150.25}

# Mark required arguments
get_stock_price._required_args = ["symbol"]

executor = ToolExecutor()
executor.register("get_stock_price", get_stock_price)

Asynchronous Execution

For I/O-bound operations:

import asyncio
from typing import Any, Dict, List, Optional

class AsyncToolExecutor:
    def __init__(self):
        self.tools: Dict[str, callable] = {}
    
    def register(self, name: str, func: callable):
        """Register an async tool."""
        self.tools[name] = func
    
    async def execute(self, name: str, 
                    arguments: Dict[str, Any]) -> Any:
        """Execute an async tool."""
        if name not in self.tools:
            raise ValueError(f"Unknown tool: {name}")
        
        tool = self.tools[name]
        
        try:
            if asyncio.iscoroutinefunction(tool):
                result = await tool(**arguments)
            else:
                result = await asyncio.to_thread(tool, **arguments)
            
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    async def execute_parallel(self, 
                             calls: List[tuple]) -> List[Any]:
        """Execute multiple tool calls in parallel."""
        tasks = [
            self.execute(name, args)
            for name, args in calls
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Process results
        processed = []
        for result in results:
            if isinstance(result, Exception):
                processed.append({"success": False, "error": str(result)})
            else:
                processed.append(result)
        
        return processed


# Example: Async API calls
import aiohttp

async def fetch_url(url: str) -> dict:
    """Fetch a URL asynchronously."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return {
                "url": url,
                "status": response.status,
                "content": await response.text()
            }

async def fetch_multiple(urls: List[str]) -> List[dict]:
    """Fetch multiple URLs in parallel."""
    executor = AsyncToolExecutor()
    executor.register("fetch_url", fetch_url)
    
    calls = [("fetch_url", {"url": url}) for url in urls]
    return await executor.execute_parallel(calls)

Tool Categories

Information Retrieval Tools

def search_knowledge_base(query: str, filters: dict = None) -> List[dict]:
    """Search internal knowledge base."""
    # Implementation
    return [
        {"title": "Article 1", "snippet": "...", "relevance": 0.95},
        {"title": "Article 2", "snippet": "...", "relevance": 0.87}
    ]

def get_customer_data(customer_id: str) -> dict:
    """Retrieve customer information."""
    # Implementation
    return {
        "id": customer_id,
        "name": "John Doe",
        "email": "[email protected]",
        "tier": "premium"
    }

Action Execution Tools

def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email."""
    # Implementation
    return {"status": "sent", "message_id": "msg_123"}

def create_calendar_event(event: dict) -> dict:
    """Create a calendar event."""
    # Implementation
    return {"status": "created", "event_id": "evt_456"}

def update_database_record(table: str, id: str, data: dict) -> dict:
    """Update a database record."""
    # Implementation
    return {"status": "updated", "table": table, "id": id}

Computation Tools

def calculate_mortgage(principal: float, rate: float, 
                       years: int) -> dict:
    """Calculate mortgage payments."""
    monthly_rate = rate / 12 / 100
    n_payments = years * 12
    
    monthly_payment = principal * (
        monthly_rate * (1 + monthly_rate) ** n_payments
    ) / ((1 + monthly_rate) ** n_payments - 1)
    
    total_payment = monthly_payment * n_payments
    total_interest = total_payment - principal
    
    return {
        "principal": principal,
        "interest_rate": rate,
        "years": years,
        "monthly_payment": round(monthly_payment, 2),
        "total_payment": round(total_payment, 2),
        "total_interest": round(total_interest, 2)
    }

def run_sql_query(query: str) -> dict:
    """Execute a SQL query."""
    # Implementation with proper security
    # Only SELECT queries allowed
    return {"rows": [], "columns": []}

Error Handling

Retry Strategies

import time
from functools import wraps
from typing import TypeVar, Callable

T = TypeVar('T')

def with_retry(max_attempts: int = 3, 
               backoff_factor: float = 2.0,
               exceptions: tuple = (Exception,)):
    """Decorator for retry logic."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            last_exception = None
            
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        wait_time = backoff_factor ** attempt
                        time.sleep(wait_time)
            
            raise last_exception
        
        return wrapper
    return decorator

@with_retry(max_attempts=3, exceptions=(ConnectionError, TimeoutError))
def unreliable_api_call(url: str) -> dict:
    """API call that might fail transiently."""
    pass

Graceful Degradation

from typing import Optional
from enum import Enum

class ToolStatus(Enum):
    AVAILABLE = "available"
    DEGRADED = "degraded"
    UNAVAILABLE = "unavailable"

class FallbackTool:
    def __init__(self, primary: callable, fallback: callable):
        self.primary = primary
        self.fallback = fallback
    
    def execute(self, *args, **kwargs):
        try:
            result = self.primary(*args, **kwargs)
            return {
                "result": result,
                "source": "primary",
                "status": ToolStatus.AVAILABLE
            }
        except Exception as primary_error:
            try:
                result = self.fallback(*args, **kwargs)
                return {
                    "result": result,
                    "source": "fallback",
                    "status": ToolStatus.DEGRADED,
                    "warning": f"Primary failed: {str(primary_error)}"
                }
            except Exception as fallback_error:
                return {
                    "result": None,
                    "source": None,
                    "status": ToolStatus.UNAVAILABLE,
                    "error": str(fallback_error)
                }

# Usage
primary_search = search_premium_api
fallback_search = search_free_api

search_tool = FallbackTool(primary_search, fallback_search)
result = search_tool.execute("query")

Security Considerations

Input Validation

import re
from typing import Any, Dict

class InputValidator:
    @staticmethod
    def validate_email(email: str) -> bool:
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, email))
    
    @staticmethod
    def validate_url(url: str) -> bool:
        pattern = r'^https?://[^\s/$.?#].[^\s]*$'
        return bool(re.match(pattern, url))
    
    @staticmethod
    def sanitize_sql(query: str) -> str:
        # Remove potentially dangerous patterns
        dangerous = ['DROP', 'DELETE', 'INSERT', 'UPDATE', 
                    'CREATE', 'ALTER', 'EXEC', 'EXECUTE']
        
        for keyword in dangerous:
            if keyword in query.upper():
                raise ValueError(f"SQL keyword '{keyword}' not allowed")
        
        return query
    
    @staticmethod
    def validate_function_params(schema: dict, 
                                params: Dict[str, Any]) -> Dict[str, Any]:
        """Validate parameters against schema."""
        validated = {}
        
        # Check required
        required = schema.get("required", [])
        for field in required:
            if field not in params:
                raise ValueError(f"Missing required field: {field}")
        
        # Validate types and constraints
        properties = schema.get("properties", {})
        for key, value in params.items():
            if key in properties:
                prop_schema = properties[key]
                
                # Type check
                expected_type = prop_schema.get("type")
                if expected_type:
                    if not isinstance(value, eval(expected_type.capitalize())):
                        raise TypeError(
                            f"Invalid type for {key}: expected {expected_type}"
                        )
                
                # Enum check
                if "enum" in prop_schema:
                    if value not in prop_schema["enum"]:
                        raise ValueError(
                            f"Invalid value for {key}: must be one of {prop_schema['enum']}"
                        )
                
                validated[key] = value
        
        return validated

Rate Limiting

from datetime import datetime, timedelta
from collections import defaultdict
from threading import Lock

class RateLimiter:
    def __init__(self, max_calls: int, window_seconds: int):
        self.max_calls = max_calls
        self.window = timedelta(seconds=window_seconds)
        self.calls = defaultdict(list)
        self.lock = Lock()
    
    def is_allowed(self, key: str) -> bool:
        """Check if call is allowed."""
        with self.lock:
            now = datetime.utcnow()
            cutoff = now - self.window
            
            # Clean old calls
            self.calls[key] = [
                t for t in self.calls[key] if t > cutoff
            ]
            
            if len(self.calls[key]) >= self.max_calls:
                return False
            
            self.calls[key].append(now)
            return True
    
    def wait_time(self, key: str) -> float:
        """Get seconds until next call allowed."""
        with self.lock:
            if not self.calls[key]:
                return 0
            
            oldest = min(self.calls[key])
            return (oldest + self.window - datetime.utcnow()).total_seconds()


# Usage
limiter = RateLimiter(max_calls=100, window_seconds=60)

def rate_limited_tool(param: str) -> dict:
    if not limiter.is_allowed("user_123"):
        wait = limiter.wait_time("user_123")
        raise RuntimeError(f"Rate limit exceeded. Wait {wait:.1f}s")
    
    return {"result": "success"}

Best Practices

1. Design Clear Function Schemas

# Good: Clear, descriptive schema
{
    "name": "create_calendar_event",
    "description": "Create a new calendar event. Use this when users want to schedule meetings or events.",
    "parameters": {
        "type": "object",
        "properties": {
            "title": {
                "type": "string",
                "description": "Event title, e.g., 'Team Meeting'"
            },
            "start_time": {
                "type": "string", 
                "description": "Start time in ISO 8601 format, e.g., '2024-01-15T10:00:00Z'"
            },
            "duration_minutes": {
                "type": "integer",
                "description": "Event duration in minutes",
                "minimum": 15,
                "maximum": 480
            },
            "attendees": {
                "type": "array",
                "items": {"type": "string"},
                "description": "List of attendee email addresses"
            }
        },
        "required": ["title", "start_time"]
    }
}

2. Handle Edge Cases

@function_tool
def robust_search(query: str, max_results: int = 10) -> dict:
    """Search with comprehensive error handling."""
    if not query or not query.strip():
        return {"error": "Query cannot be empty", "results": []}
    
    if max_results < 1:
        max_results = 10
    if max_results > 100:
        max_results = 100  # Cap at reasonable limit
    
    try:
        results = perform_search(query.strip(), max_results)
        return {"results": results, "count": len(results)}
    except TimeoutError:
        return {"error": "Search timed out", "results": [], "retryable": True}
    except PermissionError:
        return {"error": "Access denied", "results": []}
    except Exception as e:
        return {"error": f"Search failed: {str(e)}", "results": []}

3. Log Function Calls

import structlog
from datetime import datetime

logger = structlog.get_logger()

def logged_tool(func):
    """Decorator to log tool executions."""
    def wrapper(*args, **kwargs):
        call_id = f"{datetime.utcnow().timestamp()}"
        
        logger.info(
            "tool_call_start",
            tool=func.__name__,
            call_id=call_id,
            args=str(kwargs)[:200]  # Truncate long args
        )
        
        start = datetime.utcnow()
        
        try:
            result = func(*args, **kwargs)
            duration = (datetime.utcnow() - start).total_seconds()
            
            logger.info(
                "tool_call_success",
                tool=func.__name__,
                call_id=call_id,
                duration_ms=duration * 1000
            )
            
            return result
        except Exception as e:
            duration = (datetime.utcnow() - start).total_seconds()
            
            logger.error(
                "tool_call_failed",
                tool=func.__name__,
                call_id=call_id,
                duration_ms=duration * 1000,
                error=str(e)
            )
            
            raise
    
    return wrapper

Common Pitfalls

1. Not Handling Tool Failures

# Bad: No error handling
def get_data(url):
    return requests.get(url).json()

# Good: Proper error handling
def get_data(url):
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return {"data": response.json()}
    except requests.Timeout:
        return {"error": "Request timed out", "retryable": True}
    except requests.RequestException as e:
        return {"error": f"Request failed: {e}", "retryable": False}

2. Vague Function Descriptions

# Bad: Unclear description
{
    "name": "process",
    "description": "Process something"
}

# Good: Clear description
{
    "name": "process_refund",
    "description": "Process a customer refund. Calculate the refund amount including any applicable fees and initiate the refund to the original payment method."
}

3. Missing Input Validation

Always validate and sanitize inputs:

@function_tool
def transfer_funds(from_account: str, to_account: str, amount: float) -> dict:
    # Validate inputs
    if amount <= 0:
        return {"error": "Amount must be positive"}
    if amount > 10000:
        return {"error": "Amount exceeds limit"}
    
    # Validate accounts (format check)
    if not re.match(r'^[A-Z]{2}\d{10}$', from_account):
        return {"error": "Invalid source account format"}
    
    # Proceed with transfer
    return {"status": "processing"}

External Resources

Conclusion

Function calling represents a fundamental capability for building AI systems that can interact with the real world. By understanding the patterns and practices outlined in this guide, you can create robust AI applications that:

Access real-time information from external sources
Perform actions through APIs and services
Handle errors gracefully with retry and fallback mechanisms
Maintain security through proper validation and rate limiting

Key takeaways:

Design clear, well-documented function schemas
Implement comprehensive error handling
Add logging and monitoring for observability
Validate all inputs rigorously
Test tool behavior extensively

As AI systems become more sophisticated, mastering function calling will enable you to build applications that go beyond passive responses and become active participants in accomplishing real-world tasks.