Function Calling

Extend AI capabilities by enabling models to call external functions and APIs, allowing them to access real-time data, perform calculations, and interact with external systems.

Overview

Function calling allows AI models to:

Access real-time data - Weather, stock prices, news
Perform calculations - Complex math, data analysis
Interact with APIs - Database queries, web services
Execute actions - Send emails, create calendar events
Use tools - Code execution, file operations

Tool Integration

Connect AI to external tools and services

Structured Output

Get properly formatted function calls with parameters

Real-time Data

Access live information during conversations

Action Execution

Perform real-world actions through AI commands

How Function Calling Works

1. Function Definition

Define available functions with their parameters and descriptions:

{
  "name": "get_weather",
  "description": "Get current weather for a location",
  "parameters": {
    "type": "object",
    "properties": {
      "location": {
        "type": "string",
        "description": "City and state, e.g. San Francisco, CA"
      },
      "unit": {
        "type": "string",
        "enum": ["celsius", "fahrenheit"],
        "description": "Temperature unit"
      }
    },
    "required": ["location"]
  }
}

2. Model Decision

The AI model decides when and how to call functions based on user input.

3. Function Execution

Your application executes the function with provided parameters.

4. Result Integration

Results are passed back to the model to generate the final response.

Basic Function Calling

import requests
import json

def get_weather(location, unit="celsius"):
    """Mock weather function"""
    # In reality, this would call a weather API
    return {
        "location": location,
        "temperature": "22°C" if unit == "celsius" else "72°F",
        "condition": "Sunny",
        "humidity": "45%"
    }

def calculate(expression):
    """Safe calculator function"""
    try:
        # Only allow basic math operations
        allowed_chars = set("0123456789+-*/.()")
        if all(c in allowed_chars for c in expression.replace(" ", "")):
            result = eval(expression)
            return {"expression": expression, "result": result}
        else:
            return {"error": "Invalid expression"}
    except:
        return {"error": "Calculation failed"}

def function_calling_chat(messages, available_functions):
    """Chat with function calling capability"""
    
    # Define functions for the API
    functions = [
        {
            "name": "get_weather",
            "description": "Get current weather information for a location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. San Francisco, CA"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["location"]
            }
        },
        {
            "name": "calculate",
            "description": "Perform mathematical calculations",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "Mathematical expression to evaluate"
                    }
                },
                "required": ["expression"]
            }
        }
    ]
    
    # Make initial request
    response = requests.post(
        "https://api.anyapi.ai/v1/chat/completions",
        headers={
            "Authorization": "Bearer YOUR_API_KEY",
            "Content-Type": "application/json"
        },
        json={
            "model": "gpt-4o",
            "messages": messages,
            "functions": functions,
            "function_call": "auto"
        }
    )
    
    response_data = response.json()
    message = response_data["choices"][0]["message"]
    
    # Check if the model wants to call a function
    if message.get("function_call"):
        function_name = message["function_call"]["name"]
        function_args = json.loads(message["function_call"]["arguments"])
        
        print(f"Calling function: {function_name}")
        print(f"Arguments: {function_args}")
        
        # Execute the function
        if function_name in available_functions:
            function_result = available_functions[function_name](**function_args)
        else:
            function_result = {"error": "Function not found"}
        
        # Add the function call and result to the conversation
        messages.append(message)  # Assistant's function call
        messages.append({
            "role": "function",
            "name": function_name,
            "content": json.dumps(function_result)
        })
        
        # Get the final response
        final_response = requests.post(
            "https://api.anyapi.ai/v1/chat/completions",
            headers={
                "Authorization": "Bearer YOUR_API_KEY",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4o",
                "messages": messages
            }
        )
        
        return final_response.json()["choices"][0]["message"]["content"]
    
    else:
        # No function call needed
        return message["content"]

# Available functions mapping
available_functions = {
    "get_weather": get_weather,
    "calculate": calculate
}

# Usage
messages = [
    {"role": "user", "content": "What's the weather like in Tokyo and what's 15 * 24?"}
]

response = function_calling_chat(messages, available_functions)
print(response)

Advanced Function Calling Patterns

Multiple Function Calls

class AdvancedFunctionCaller:
    def __init__(self, api_key):
        self.api_key = api_key
        self.functions = {}
        self.function_definitions = []
    
    def register_function(self, name, func, description, parameters):
        """Register a function for calling"""
        self.functions[name] = func
        self.function_definitions.append({
            "name": name,
            "description": description,
            "parameters": parameters
        })
    
    def chat_with_functions(self, messages, max_function_calls=5):
        """Handle multiple function calls in a conversation"""
        
        conversation = messages.copy()
        function_call_count = 0
        
        while function_call_count < max_function_calls:
            response = requests.post(
                "https://api.anyapi.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gpt-4o",
                    "messages": conversation,
                    "functions": self.function_definitions,
                    "function_call": "auto"
                }
            )
            
            response_data = response.json()
            message = response_data["choices"][0]["message"]
            
            if message.get("function_call"):
                function_name = message["function_call"]["name"]
                function_args = json.loads(message["function_call"]["arguments"])
                
                print(f"Function call #{function_call_count + 1}: {function_name}")
                
                # Execute function
                if function_name in self.functions:
                    try:
                        function_result = self.functions[function_name](**function_args)
                    except Exception as e:
                        function_result = {"error": str(e)}
                else:
                    function_result = {"error": "Function not available"}
                
                # Add to conversation
                conversation.append(message)
                conversation.append({
                    "role": "function",
                    "name": function_name,
                    "content": json.dumps(function_result)
                })
                
                function_call_count += 1
            else:
                # No more function calls needed
                return message["content"]
        
        # Max function calls reached
        return "I've reached the maximum number of function calls. Please ask me to continue if needed."

# Example functions
def get_stock_price(symbol):
    """Mock stock price function"""
    prices = {
        "AAPL": {"price": 175.84, "change": "+2.3%"},
        "GOOGL": {"price": 142.56, "change": "-0.8%"},
        "MSFT": {"price": 378.85, "change": "+1.2%"}
    }
    return prices.get(symbol.upper(), {"error": "Symbol not found"})

def search_news(query, limit=3):
    """Mock news search function"""
    return {
        "query": query,
        "articles": [
            {"title": f"Article about {query} #1", "summary": "Summary 1"},
            {"title": f"Article about {query} #2", "summary": "Summary 2"},
            {"title": f"Article about {query} #3", "summary": "Summary 3"}
        ][:limit]
    }

def send_email(to, subject, body):
    """Mock email function"""
    return {
        "status": "sent",
        "to": to,
        "subject": subject,
        "message": "Email sent successfully"
    }

# Setup
function_caller = AdvancedFunctionCaller("YOUR_API_KEY")

# Register functions
function_caller.register_function(
    "get_stock_price",
    get_stock_price,
    "Get current stock price for a given symbol",
    {
        "type": "object",
        "properties": {
            "symbol": {"type": "string", "description": "Stock symbol (e.g., AAPL)"}
        },
        "required": ["symbol"]
    }
)

function_caller.register_function(
    "search_news",
    search_news,
    "Search for news articles about a topic",
    {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"},
            "limit": {"type": "integer", "description": "Number of articles", "default": 3}
        },
        "required": ["query"]
    }
)

# Usage
messages = [
    {
        "role": "user", 
        "content": "Get me the stock price for Apple, then search for recent news about the company"
    }
]

response = function_caller.chat_with_functions(messages)
print(response)

Streaming with Function Calls

def stream_with_function_calls(messages, functions, available_functions):
    """Stream responses while handling function calls"""
    
    response = requests.post(
        "https://api.anyapi.ai/v1/chat/completions",
        headers={
            "Authorization": "Bearer YOUR_API_KEY",
            "Content-Type": "application/json"
        },
        json={
            "model": "gpt-4o",
            "messages": messages,
            "functions": functions,
            "function_call": "auto",
            "stream": True
        },
        stream=True
    )
    
    function_call_buffer = ""
    function_name = ""
    
    for line in response.iter_lines(decode_unicode=True):
        if line.startswith('data: '):
            data = line[6:]
            
            if data == '[DONE]':
                break
            
            try:
                chunk = json.loads(data)
                
                if chunk.get('choices') and len(chunk['choices']) > 0:
                    choice = chunk['choices'][0]
                    delta = choice.get('delta', {})
                    
                    # Handle regular content
                    if 'content' in delta and delta['content']:
                        yield {'type': 'content', 'data': delta['content']}
                    
                    # Handle function calls
                    if 'function_call' in delta:
                        func_call = delta['function_call']
                        
                        if 'name' in func_call:
                            function_name = func_call['name']
                            yield {'type': 'function_call_start', 'data': function_name}
                        
                        if 'arguments' in func_call:
                            function_call_buffer += func_call['arguments']
                    
                    # Handle completion
                    if choice.get('finish_reason') == 'function_call':
                        # Execute function
                        try:
                            args = json.loads(function_call_buffer)
                            result = available_functions[function_name](**args)
                            
                            yield {
                                'type': 'function_result',
                                'data': {
                                    'name': function_name,
                                    'arguments': args,
                                    'result': result
                                }
                            }
                            
                            # Continue conversation with function result
                            messages.append({
                                "role": "assistant",
                                "content": None,
                                "function_call": {
                                    "name": function_name,
                                    "arguments": function_call_buffer
                                }
                            })
                            
                            messages.append({
                                "role": "function",
                                "name": function_name,
                                "content": json.dumps(result)
                            })
                            
                            # Get continuation
                            yield from stream_with_function_calls(messages, functions, available_functions)
                            return
                            
                        except Exception as e:
                            yield {'type': 'error', 'data': f"Function execution failed: {e}"}
                            return
                    
            except json.JSONDecodeError:
                continue

# Usage
functions = [
    {
        "name": "get_weather",
        "description": "Get weather information",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"}
            },
            "required": ["location"]
        }
    }
]

available_functions = {
    "get_weather": lambda location: {
        "location": location,
        "temperature": "22°C",
        "condition": "Sunny"
    }
}

messages = [
    {"role": "user", "content": "What's the weather like in Tokyo?"}
]

for event in stream_with_function_calls(messages, functions, available_functions):
    if event['type'] == 'content':
        print(event['data'], end='', flush=True)
    elif event['type'] == 'function_call_start':
        print(f"\n[Calling {event['data']}...]")
    elif event['type'] == 'function_result':
        print(f"[Got result: {event['data']['result']}]")

Real-World Function Examples

Database Operations

import sqlite3

class DatabaseFunctions:
    def __init__(self, db_path):
        self.db_path = db_path
    
    def query_users(self, limit=10, filter_active=True):
        """Query users from database"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            query = "SELECT id, name, email, created_at FROM users"
            params = []
            
            if filter_active:
                query += " WHERE active = ?"
                params.append(True)
            
            query += " LIMIT ?"
            params.append(limit)
            
            cursor.execute(query, params)
            results = cursor.fetchall()
            
            users = []
            for row in results:
                users.append({
                    "id": row[0],
                    "name": row[1],
                    "email": row[2],
                    "created_at": row[3]
                })
            
            conn.close()
            return {"users": users, "count": len(users)}
            
        except Exception as e:
            return {"error": str(e)}
    
    def get_user_stats(self):
        """Get user statistics"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute("SELECT COUNT(*) FROM users WHERE active = 1")
            active_users = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM users")
            total_users = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM users WHERE created_at > datetime('now', '-30 days')")
            new_users = cursor.fetchone()[0]
            
            conn.close()
            
            return {
                "total_users": total_users,
                "active_users": active_users,
                "new_users_30_days": new_users
            }
            
        except Exception as e:
            return {"error": str(e)}

# Function definitions for API
def get_db_function_definitions():
    return [
        {
            "name": "query_users",
            "description": "Query users from the database",
            "parameters": {
                "type": "object",
                "properties": {
                    "limit": {
                        "type": "integer",
                        "description": "Maximum number of users to return",
                        "default": 10
                    },
                    "filter_active": {
                        "type": "boolean",
                        "description": "Only return active users",
                        "default": True
                    }
                }
            }
        },
        {
            "name": "get_user_stats",
            "description": "Get user statistics and metrics",
            "parameters": {
                "type": "object",
                "properties": {}
            }
        }
    ]

API Integration

import requests
from datetime import datetime

class ExternalAPIFunctions:
    def __init__(self, weather_api_key, news_api_key):
        self.weather_api_key = weather_api_key
        self.news_api_key = news_api_key
    
    def get_weather(self, location, unit="metric"):
        """Get real weather data from OpenWeatherMap API"""
        try:
            url = f"http://api.openweathermap.org/data/2.5/weather"
            params = {
                "q": location,
                "appid": self.weather_api_key,
                "units": unit
            }
            
            response = requests.get(url, params=params, timeout=5)
            response.raise_for_status()
            
            data = response.json()
            
            return {
                "location": data["name"],
                "country": data["sys"]["country"],
                "temperature": f"{data['main']['temp']}°{'C' if unit == 'metric' else 'F'}",
                "description": data["weather"][0]["description"],
                "humidity": f"{data['main']['humidity']}%",
                "pressure": f"{data['main']['pressure']} hPa",
                "wind_speed": f"{data['wind']['speed']} {'m/s' if unit == 'metric' else 'mph'}"
            }
            
        except requests.exceptions.RequestException as e:
            return {"error": f"Weather API error: {str(e)}"}
        except KeyError as e:
            return {"error": f"Invalid weather data format: {str(e)}"}
    
    def search_news(self, query, limit=5, language="en"):
        """Search news using NewsAPI"""
        try:
            url = "https://newsapi.org/v2/everything"
            params = {
                "q": query,
                "apiKey": self.news_api_key,
                "language": language,
                "sortBy": "publishedAt",
                "pageSize": limit
            }
            
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            
            data = response.json()
            
            articles = []
            for article in data.get("articles", []):
                articles.append({
                    "title": article["title"],
                    "description": article["description"],
                    "source": article["source"]["name"],
                    "url": article["url"],
                    "published_at": article["publishedAt"]
                })
            
            return {
                "query": query,
                "total_results": data.get("totalResults", 0),
                "articles": articles
            }
            
        except requests.exceptions.RequestException as e:
            return {"error": f"News API error: {str(e)}"}
    
    def get_stock_price(self, symbol):
        """Get stock price from Alpha Vantage (free tier)"""
        try:
            # Using a free API for demo purposes
            url = f"https://api.twelvedata.com/price"
            params = {
                "symbol": symbol.upper(),
                "apikey": "demo"  # Use demo key for testing
            }
            
            response = requests.get(url, params=params, timeout=5)
            response.raise_for_status()
            
            data = response.json()
            
            if "price" in data:
                return {
                    "symbol": symbol.upper(),
                    "price": f"${float(data['price']):.2f}",
                    "timestamp": datetime.now().isoformat()
                }
            else:
                return {"error": "Stock data not available"}
                
        except requests.exceptions.RequestException as e:
            return {"error": f"Stock API error: {str(e)}"}

# Function definitions
def get_api_function_definitions():
    return [
        {
            "name": "get_weather",
            "description": "Get current weather information for any location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name or 'City, Country' format"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["metric", "imperial"],
                        "description": "Temperature unit (metric for Celsius, imperial for Fahrenheit)",
                        "default": "metric"
                    }
                },
                "required": ["location"]
            }
        },
        {
            "name": "search_news",
            "description": "Search for recent news articles on any topic",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query for news articles"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Number of articles to return (max 10)",
                        "default": 5,
                        "minimum": 1,
                        "maximum": 10
                    },
                    "language": {
                        "type": "string",
                        "description": "Language code (e.g., 'en', 'es', 'fr')",
                        "default": "en"
                    }
                },
                "required": ["query"]
            }
        },
        {
            "name": "get_stock_price",
            "description": "Get current stock price for a given symbol",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "Stock symbol (e.g., AAPL, GOOGL, MSFT)"
                    }
                },
                "required": ["symbol"]
            }
        }
    ]

Error Handling and Validation

import jsonschema
from jsonschema import validate

class SafeFunctionCaller:
    def __init__(self, api_key):
        self.api_key = api_key
        self.functions = {}
        self.schemas = {}
    
    def register_function(self, name, func, schema):
        """Register function with input validation"""
        self.functions[name] = func
        self.schemas[name] = schema
    
    def validate_function_args(self, function_name, args):
        """Validate function arguments against schema"""
        if function_name not in self.schemas:
            return False, "Function not registered"
        
        try:
            validate(instance=args, schema=self.schemas[function_name]["parameters"])
            return True, None
        except jsonschema.exceptions.ValidationError as e:
            return False, f"Validation error: {e.message}"
    
    def safe_execute_function(self, function_name, args):
        """Safely execute function with validation"""
        # Validate arguments
        is_valid, error = self.validate_function_args(function_name, args)
        if not is_valid:
            return {"error": error}
        
        # Execute function with timeout and error handling
        try:
            if function_name in self.functions:
                result = self.functions[function_name](**args)
                return result
            else:
                return {"error": "Function not found"}
        except TypeError as e:
            return {"error": f"Invalid arguments: {str(e)}"}
        except Exception as e:
            return {"error": f"Function execution failed: {str(e)}"}
    
    def chat_with_safe_functions(self, messages):
        """Chat with safe function execution"""
        function_definitions = [
            {
                "name": name,
                "description": schema["description"],
                "parameters": schema["parameters"]
            }
            for name, schema in self.schemas.items()
        ]
        
        response = requests.post(
            "https://api.anyapi.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4o",
                "messages": messages,
                "functions": function_definitions,
                "function_call": "auto"
            }
        )
        
        response_data = response.json()
        message = response_data["choices"][0]["message"]
        
        if message.get("function_call"):
            function_name = message["function_call"]["name"]
            
            try:
                function_args = json.loads(message["function_call"]["arguments"])
            except json.JSONDecodeError:
                function_result = {"error": "Invalid function arguments JSON"}
            else:
                function_result = self.safe_execute_function(function_name, function_args)
            
            # Continue conversation
            messages.append(message)
            messages.append({
                "role": "function",
                "name": function_name,
                "content": json.dumps(function_result)
            })
            
            # Get final response
            final_response = requests.post(
                "https://api.anyapi.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gpt-4o",
                    "messages": messages
                }
            )
            
            return final_response.json()["choices"][0]["message"]["content"]
        
        return message["content"]

# Usage with safe execution
safe_caller = SafeFunctionCaller("YOUR_API_KEY")

# Register function with schema
safe_caller.register_function(
    "divide_numbers",
    lambda x, y: {"result": x / y if y != 0 else "Cannot divide by zero"},
    {
        "description": "Divide two numbers",
        "parameters": {
            "type": "object",
            "properties": {
                "x": {"type": "number"},
                "y": {"type": "number"}
            },
            "required": ["x", "y"]
        }
    }
)

Best Practices

1. Function Design

Clear descriptions: Explain what the function does and when to use it
Proper parameters: Use appropriate types and include descriptions
Error handling: Always handle and return errors gracefully
Input validation: Validate all inputs before processing

2. Security Considerations

Validate inputs: Never trust function arguments without validation
Limit permissions: Functions should have minimal required permissions
Rate limiting: Implement rate limiting for expensive operations
Sanitize outputs: Clean sensitive data from function results

3. Performance Optimization

Timeout handling: Set reasonable timeouts for external API calls
Caching: Cache results when appropriate
Async execution: Use async functions for I/O operations
Batch operations: Group related operations when possible

4. User Experience

Clear feedback: Provide status updates for long-running functions
Graceful degradation: Handle API failures without breaking conversation
Helpful errors: Return user-friendly error messages

Common Patterns

Function Chaining

# Functions can call other functions through the AI model
def multi_step_analysis(user_query):
    """Example of chained function calls"""
    
    # Step 1: AI decides to get stock price
    # Step 2: AI gets news about the company
    # Step 3: AI analyzes both and provides recommendation
    
    pass

Conditional Execution

# Functions with conditional logic
def smart_weather_advice(location, activity):
    """Get weather and provide activity-specific advice"""
    weather = get_weather(location)
    
    if weather["temperature"] > 25:
        if activity == "running":
            return {"advice": "Great weather for running! Stay hydrated."}
        elif activity == "skiing":
            return {"advice": "Too warm for skiing. Consider other activities."}
    
    return {"advice": "Check current conditions and dress appropriately."}

Getting Started

Quick Start

Start with function calling

API Reference

Explore function calling parameters

Use Cases

See function calling examples

Chat Models

Learn about chat completions

Get started

Features

Use Cases

Developer guides

API Reference

Integrations

Function Calling

Function Calling

Overview

Tool Integration

Structured Output

Real-time Data

Action Execution

How Function Calling Works

1. Function Definition

2. Model Decision

3. Function Execution

4. Result Integration

Basic Function Calling

Advanced Function Calling Patterns

Multiple Function Calls

Streaming with Function Calls

Real-World Function Examples

Database Operations

API Integration

Error Handling and Validation

Best Practices

1. Function Design

2. Security Considerations

3. Performance Optimization

4. User Experience

Common Patterns

Function Chaining

Conditional Execution

Getting Started

Quick Start

API Reference

Use Cases

Chat Models

Get started

Features

Use Cases

Developer guides

API Reference

Integrations

​Function Calling

​Overview

Tool Integration

Structured Output

Real-time Data

Action Execution

​How Function Calling Works

​1. Function Definition

​2. Model Decision

​3. Function Execution

​4. Result Integration

​Basic Function Calling

​Advanced Function Calling Patterns

​Multiple Function Calls

​Streaming with Function Calls

​Real-World Function Examples

​Database Operations

​API Integration

​Error Handling and Validation

​Best Practices

​1. Function Design

​2. Security Considerations

​3. Performance Optimization

​4. User Experience

​Common Patterns

​Function Chaining

​Conditional Execution

​Getting Started

Quick Start

API Reference

Use Cases

Chat Models

Function Calling

Overview

How Function Calling Works

1. Function Definition

2. Model Decision

3. Function Execution

4. Result Integration

Basic Function Calling

Advanced Function Calling Patterns

Multiple Function Calls

Streaming with Function Calls

Real-World Function Examples

Database Operations

API Integration

Error Handling and Validation

Best Practices

1. Function Design

2. Security Considerations

3. Performance Optimization

4. User Experience

Common Patterns

Function Chaining

Conditional Execution

Getting Started