How the Magic Works
The Tool Calling Dance
A simple 6-step process that changes everything- 🛠️ You define the tools – Tell the AI what functions are available
- 🧠 AI analyzes the request – Model decides which tools it needs
- 📞 AI requests tool execution – “Hey, can you run
search_databasewith these parameters?” - ⚡ Your code executes tools – You run the actual functions and get results
- 📤 You return the results – Feed the data back to the AI
- ✨ AI generates final response – Model combines tool results into a helpful answer
💡 Key insight: The AI doesn’t execute tools directly – it’s like having a smart assistant who knows what to ask for, but you maintain complete control over what actually happens.
Quick Start: Your First Tool
Define a Simple Tool
Let’s start with something usefulimport requests
import json
def search_products(query, max_price=None, category=None):
"""Search your product database"""
# Your actual database logic here
products = [
{"id": 1, "name": f"Premium {query}", "price": 899, "category": category},
{"id": 2, "name": f"Budget {query}", "price": 299, "category": category}
]
# Filter by price if specified
if max_price:
products = [p for p in products if p["price"] <= max_price]
return products
# 🛠️ Tool definition - this is what the AI sees
tools = [
{
"type": "function",
"function": {
"name": "search_products",
"description": "Search the product database for items matching criteria",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "What to search for (e.g., 'laptop', 'headphones')"
},
"max_price": {
"type": "number",
"description": "Maximum price filter in dollars"
},
"category": {
"type": "string",
"enum": ["electronics", "clothing", "books", "home"],
"description": "Product category to search within"
}
},
"required": ["query"]
}
}
}
]
Execute the Tool-Enhanced Conversation
Watch the AI use your tools intelligentlyclass ToolEnabledChat:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.anyapi.ai/v1/chat/completions"
self.available_functions = {
"search_products": search_products
}
def chat_with_tools(self, user_message, tools):
"""Handle a complete tool-calling conversation"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
messages = [{"role": "user", "content": user_message}]
# 🎯 Initial request with tools available
payload = {
"model": "openai/gpt-4o-2024-11-20",
"messages": messages,
"tools": tools,
"tool_choice": "auto" # Let AI decide when to use tools
}
response = requests.post(self.base_url, headers=headers, json=payload)
result = response.json()
message = result["choices"][0]["message"]
messages.append(message)
# 🔧 Check if AI wants to use tools
if message.get("tool_calls"):
print(f"🛠️ AI wants to use {len(message['tool_calls'])} tool(s)")
# Execute each requested tool
for tool_call in message["tool_calls"]:
function_name = tool_call["function"]["name"]
function_args = json.loads(tool_call["function"]["arguments"])
print(f"⚡ Executing {function_name} with args: {function_args}")
# Execute the function
if function_name in self.available_functions:
try:
result = self.available_functions[function_name](**function_args)
print(f"✅ Tool result: {result}")
# Add tool result to conversation
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(result)
})
except Exception as e:
# Handle tool execution errors
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps({"error": str(e)})
})
else:
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps({"error": f"Unknown function: {function_name}"})
})
# 🎭 Send results back to AI for final response
payload["messages"] = messages
response = requests.post(self.base_url, headers=headers, json=payload)
final_result = response.json()
return final_result["choices"][0]["message"]["content"]
else:
# No tools needed, return the direct response
return message["content"]
# 🚀 Try it out
chat = ToolEnabledChat(API_KEY)
response = chat.chat_with_tools(
"I'm looking for laptops under $1000 in the electronics category",
tools
)
print(f"🤖 AI Response: {response}")
Real-World Power Tools
Multi-Function Assistant
Build an AI that can actually do thingsimport requests
import datetime
import os
class PowerfulAIAssistant:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.anyapi.ai/v1/chat/completions"
# 🛠️ Define all available functions
self.functions = {
"search_database": self.search_database,
"get_weather": self.get_weather,
"send_email": self.send_email,
"calculate": self.calculate,
"get_calendar_events": self.get_calendar_events,
"create_reminder": self.create_reminder
}
# 🎯 Tool definitions for the AI
self.tools = [
{
"type": "function",
"function": {
"name": "search_database",
"description": "Search internal database for products, users, or orders",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search term"},
"table": {"type": "string", "enum": ["products", "users", "orders"]},
"limit": {"type": "integer", "default": 10}
},
"required": ["query", "table"]
}
}
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather and forecast for any city",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "City name"},
"days": {"type": "integer", "description": "Forecast days (1-7)", "default": 1}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "Send an email to someone",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string", "description": "Recipient email"},
"subject": {"type": "string", "description": "Email subject"},
"body": {"type": "string", "description": "Email content"}
},
"required": ["to", "subject", "body"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "Perform mathematical calculations including complex equations",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "Math expression to calculate"}
},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "get_calendar_events",
"description": "Get upcoming calendar events",
"parameters": {
"type": "object",
"properties": {
"days_ahead": {"type": "integer", "description": "How many days to look ahead", "default": 7}
}
}
}
},
{
"type": "function",
"function": {
"name": "create_reminder",
"description": "Create a reminder for later",
"parameters": {
"type": "object",
"properties": {
"message": {"type": "string", "description": "Reminder text"},
"when": {"type": "string", "description": "When to remind (e.g., '2 hours', 'tomorrow 9am')"}
},
"required": ["message", "when"]
}
}
}
]
def search_database(self, query, table, limit=10):
"""Search your database"""
# Mock database results
results = {
"products": [
{"id": 1, "name": f"Product matching '{query}'", "price": 99},
{"id": 2, "name": f"Another {query} item", "price": 149}
],
"users": [
{"id": 1, "name": f"User {query}", "email": f"{query}@example.com"},
],
"orders": [
{"id": 1, "user": query, "total": 299, "status": "shipped"}
]
}
return results.get(table, [])[:limit]
def get_weather(self, city, days=1):
"""Get weather data"""
# Mock weather API
return {
"city": city,
"current": {"temp": 22, "condition": "Sunny"},
"forecast": [{"day": i+1, "temp": 20+i, "condition": "Partly cloudy"} for i in range(days)]
}
def send_email(self, to, subject, body):
"""Send email (mock implementation)"""
# In real implementation, integrate with email service
print(f"📧 EMAIL SENT:")
print(f"To: {to}")
print(f"Subject: {subject}")
print(f"Body: {body}")
return {"status": "sent", "message_id": "mock_123"}
def calculate(self, expression):
"""Safe calculator"""
try:
# Simple eval for demo - in production, use a proper math parser
result = eval(expression.replace('^', '**'))
return {"expression": expression, "result": result}
except:
return {"error": "Invalid expression"}
def get_calendar_events(self, days_ahead=7):
"""Get calendar events"""
# Mock calendar data
events = [
{"title": "Team Meeting", "date": "2024-01-15", "time": "10:00 AM"},
{"title": "Project Review", "date": "2024-01-16", "time": "2:00 PM"}
]
return events
def create_reminder(self, message, when):
"""Create a reminder"""
# Mock reminder system
return {
"reminder": message,
"scheduled_for": when,
"id": "reminder_456",
"status": "created"
}
def process_request(self, user_message):
"""Process a user request with full tool access"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
messages = [{"role": "user", "content": user_message}]
# Allow multiple tool calls if needed
max_iterations = 5
iteration = 0
while iteration < max_iterations:
payload = {
"model": "openai/gpt-4o-2024-11-20",
"messages": messages,
"tools": self.tools,
"tool_choice": "auto"
}
response = requests.post(self.base_url, headers=headers, json=payload)
result = response.json()
message = result["choices"][0]["message"]
messages.append(message)
if message.get("tool_calls"):
print(f"🔄 Iteration {iteration + 1}: AI using {len(message['tool_calls'])} tool(s)")
# Execute all requested tools
for tool_call in message["tool_calls"]:
function_name = tool_call["function"]["name"]
function_args = json.loads(tool_call["function"]["arguments"])
if function_name in self.functions:
try:
result = self.functions[function_name](**function_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(result)
})
except Exception as e:
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps({"error": str(e)})
})
iteration += 1
else:
# No more tool calls needed
return message["content"]
return "Sorry, the task required too many tool calls to complete."
# 🎯 Create your powerful AI assistant
assistant = PowerfulAIAssistant(API_KEY)
# 🚀 Test complex multi-tool requests
response = assistant.process_request(
"Check the weather in London, search for any orders from user 'john', and calculate 15% tip on a $85 bill. Also send me an email summary to john@example.com"
)
print(f"🤖 Assistant: {response}")
Smart E-commerce Bot
Build a shopping assistant that actually knows your inventoryclass EcommerceBot:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.anyapi.ai/v1/chat/completions"
# Mock inventory data
self.inventory = [
{"id": 1, "name": "MacBook Pro 14\"", "price": 1999, "category": "laptops", "stock": 5},
{"id": 2, "name": "ThinkPad X1 Carbon", "price": 1299, "category": "laptops", "stock": 3},
{"id": 3, "name": "Sony WH-1000XM4", "price": 299, "category": "headphones", "stock": 12},
{"id": 4, "name": "iPhone 15 Pro", "price": 999, "category": "phones", "stock": 8}
]
self.tools = [
{
"type": "function",
"function": {
"name": "search_products",
"description": "Search available products by name, category, or price range",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search term"},
"category": {"type": "string", "description": "Product category"},
"max_price": {"type": "number", "description": "Maximum price"},
"min_price": {"type": "number", "description": "Minimum price"}
}
}
}
},
{
"type": "function",
"function": {
"name": "check_stock",
"description": "Check stock availability for specific products",
"parameters": {
"type": "object",
"properties": {
"product_ids": {"type": "array", "items": {"type": "integer"}}
},
"required": ["product_ids"]
}
}
},
{
"type": "function",
"function": {
"name": "get_recommendations",
"description": "Get product recommendations based on user preferences",
"parameters": {
"type": "object",
"properties": {
"category": {"type": "string", "description": "Preferred category"},
"budget": {"type": "number", "description": "Budget limit"},
"use_case": {"type": "string", "description": "What the user needs it for"}
}
}
}
}
]
def search_products(self, query=None, category=None, max_price=None, min_price=None):
"""Search products with filters"""
results = self.inventory.copy()
if query:
results = [p for p in results if query.lower() in p["name"].lower()]
if category:
results = [p for p in results if p["category"] == category]
if max_price:
results = [p for p in results if p["price"] <= max_price]
if min_price:
results = [p for p in results if p["price"] >= min_price]
return results
def check_stock(self, product_ids):
"""Check stock for specific products"""
stock_info = []
for pid in product_ids:
product = next((p for p in self.inventory if p["id"] == pid), None)
if product:
stock_info.append({
"product_id": pid,
"name": product["name"],
"stock": product["stock"],
"available": product["stock"] > 0
})
return stock_info
def get_recommendations(self, category=None, budget=None, use_case=None):
"""Get smart recommendations"""
recommendations = self.inventory.copy()
if category:
recommendations = [p for p in recommendations if p["category"] == category]
if budget:
recommendations = [p for p in recommendations if p["price"] <= budget]
# Sort by price and stock
recommendations.sort(key=lambda x: (x["price"], -x["stock"]))
return recommendations[:3] # Top 3 recommendations
def chat(self, user_message):
"""Handle customer inquiries with tool support"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
messages = [
{
"role": "system",
"content": "You are a helpful e-commerce assistant. Use the available tools to search products, check stock, and provide recommendations. Be friendly and helpful."
},
{"role": "user", "content": user_message}
]
payload = {
"model": "openai/gpt-4o-2024-11-20",
"messages": messages,
"tools": self.tools,
"tool_choice": "auto"
}
response = requests.post(self.base_url, headers=headers, json=payload)
result = response.json()
message = result["choices"][0]["message"]
messages.append(message)
if message.get("tool_calls"):
# Execute tools
for tool_call in message["tool_calls"]:
function_name = tool_call["function"]["name"]
function_args = json.loads(tool_call["function"]["arguments"])
if function_name == "search_products":
result = self.search_products(**function_args)
elif function_name == "check_stock":
result = self.check_stock(**function_args)
elif function_name == "get_recommendations":
result = self.get_recommendations(**function_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(result)
})
# Get final response
payload["messages"] = messages
response = requests.post(self.base_url, headers=headers, json=payload)
final_result = response.json()
return final_result["choices"][0]["message"]["content"]
return message["content"]
# 🛍️ Test the e-commerce bot
bot = EcommerceBot(API_KEY)
# Try different queries
queries = [
"I'm looking for a laptop under $1500 for programming",
"Do you have the MacBook Pro in stock?",
"Can you recommend good headphones for under $400?",
"Show me all phones available"
]
for query in queries:
print(f"\n👤 Customer: {query}")
response = bot.chat(query)
print(f"🤖 Bot: {response}")
Advanced Tool Strategies
Tool Choice Control
Fine-tune when and how tools are useddef smart_tool_control(user_message, urgency="normal"):
"""Control tool usage based on context"""
# Different strategies for different situations
strategies = {
"force_tools": {
"tool_choice": "required", # Must use a tool
"use_case": "When you know tools are needed"
},
"specific_tool": {
"tool_choice": {
"type": "function",
"function": {"name": "search_database"}
},
"use_case": "Force a specific tool"
},
"no_tools": {
"tool_choice": "none",
"use_case": "Chat only, no external actions"
},
"auto_decide": {
"tool_choice": "auto",
"use_case": "Let AI decide (most common)"
}
}
# Choose strategy based on urgency and message content
if "emergency" in user_message.lower() or urgency == "high":
strategy = "force_tools"
elif "just chat" in user_message.lower():
strategy = "no_tools"
elif "search" in user_message.lower():
strategy = "specific_tool"
else:
strategy = "auto_decide"
return strategies[strategy]
# Usage examples
emergency_config = smart_tool_control("Emergency: find all critical alerts", urgency="high")
chat_config = smart_tool_control("Let's just chat about the weather")
search_config = smart_tool_control("Search for user data")
Error Handling & Resilience
Build bulletproof tool executionclass RobustToolExecutor:
def __init__(self, api_key):
self.api_key = api_key
self.retry_count = 3
self.timeout_seconds = 30
def safe_execute_tool(self, function_name, function_args, available_functions):
"""Execute tools with comprehensive error handling"""
try:
# Validate function exists
if function_name not in available_functions:
return {
"error": f"Function '{function_name}' not available",
"available_functions": list(available_functions.keys())
}
# Validate arguments
function = available_functions[function_name]
# Execute with timeout protection
import signal
def timeout_handler(signum, frame):
raise TimeoutError("Tool execution timed out")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(self.timeout_seconds)
try:
result = function(**function_args)
signal.alarm(0) # Cancel timeout
return {
"success": True,
"result": result,
"function": function_name,
"execution_time": "< 30s"
}
except TimeoutError:
return {
"error": f"Function '{function_name}' timed out after {self.timeout_seconds}s",
"suggestion": "Try with simpler parameters"
}
except Exception as e:
return {
"error": f"Function execution failed: {str(e)}",
"function": function_name,
"args_used": function_args
}
except Exception as e:
return {
"error": f"Tool execution system error: {str(e)}",
"contact_support": True
}
def execute_with_retry(self, function_name, function_args, available_functions):
"""Execute with automatic retry logic"""
last_error = None
for attempt in range(self.retry_count):
print(f"🔄 Attempt {attempt + 1}/{self.retry_count} for {function_name}")
result = self.safe_execute_tool(function_name, function_args, available_functions)
if result.get("success"):
if attempt > 0:
print(f"✅ Succeeded on attempt {attempt + 1}")
return result
last_error = result
# Wait before retry (exponential backoff)
import time
time.sleep(2 ** attempt)
return {
"error": f"Failed after {self.retry_count} attempts",
"last_error": last_error,
"suggestion": "Check your function implementation or arguments"
}
# 🛡️ Usage with error handling
executor = RobustToolExecutor(API_KEY)
result = executor.execute_with_retry(
"search_database",
{"query": "test", "table": "products"},
{"search_database": search_products}
)
if result.get("success"):
print(f"✅ Tool succeeded: {result['result']}")
else:
print(f"❌ Tool failed: {result['error']}")
Tool Security Best Practices
Input Validation & Sanitization
Never trust, always verifyimport re
from typing import Dict, Any, List
class SecureToolValidator:
def __init__(self):
self.dangerous_patterns = [
r'__.*__', # Python dunder methods
r'eval\s*\(', # eval calls
r'exec\s*\(', # exec calls
r'import\s+', # import statements
r'open\s*\(', # file operations
r'subprocess', # system calls
]
def validate_function_args(self, function_name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""Validate and sanitize tool arguments"""
cleaned_args = {}
for key, value in args.items():
# Check for dangerous patterns in string values
if isinstance(value, str):
for pattern in self.dangerous_patterns:
if re.search(pattern, value, re.IGNORECASE):
raise ValueError(f"Potentially dangerous content detected in {key}: {pattern}")
# Basic sanitization
value = value.strip()
# Length limits
if len(value) > 1000:
raise ValueError(f"Argument {key} too long (max 1000 chars)")
# Type validation based on function
if function_name == "search_database":
if key == "limit" and (not isinstance(value, int) or value > 100):
value = min(int(value), 100)
elif key == "query" and not isinstance(value, str):
value = str(value)
cleaned_args[key] = value
return cleaned_args
def is_function_allowed(self, function_name: str, user_permissions: List[str]) -> bool:
"""Check if user has permission to use this function"""
function_permissions = {
"search_database": ["read_data", "admin"],
"send_email": ["send_messages", "admin"],
"delete_data": ["admin"], # High privilege function
"get_weather": ["public"], # Anyone can use
}
required_perms = function_permissions.get(function_name, ["admin"])
return any(perm in user_permissions for perm in required_perms)
# 🔒 Secure tool execution
validator = SecureToolValidator()
def secure_tool_execution(function_name, function_args, user_permissions, available_functions):
"""Execute tools with security checks"""
try:
# 1. Check permissions
if not validator.is_function_allowed(function_name, user_permissions):
return {"error": f"Permission denied for function: {function_name}"}
# 2. Validate and sanitize arguments
clean_args = validator.validate_function_args(function_name, function_args)
# 3. Execute function
if function_name in available_functions:
result = available_functions[function_name](**clean_args)
return {"success": True, "result": result}
else:
return {"error": f"Function not found: {function_name}"}
except ValueError as e:
return {"error": f"Validation error: {str(e)}"}
except Exception as e:
return {"error": f"Execution error: {str(e)}"}
# Usage with security
user_perms = ["read_data", "send_messages"] # User's permissions
result = secure_tool_execution(
"search_database",
{"query": "laptops", "limit": 50},
user_perms,
{"search_database": search_products}
)
Performance Optimization
Parallel Tool Execution
Run multiple tools simultaneouslyimport asyncio
import aiohttp
class ParallelToolExecutor:
def __init__(self, api_key):
self.api_key = api_key
async def execute_tools_parallel(self, tool_calls, available_functions):
"""Execute multiple tools concurrently"""
async def execute_single_tool(tool_call):
function_name = tool_call["function"]["name"]
function_args = json.loads(tool_call["function"]["arguments"])
try:
# Convert sync