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:Copy
{
"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
Copy
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
Copy
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
Copy
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
Copy
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
Copy
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
Copy
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
Copy
# 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
Copy
# 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."}