Tool Calling

import 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"]
            }
        }
    }
]

class ToolEnabledChat:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.anyapi.ai/api/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}")

import requests
import datetime
import os

class PowerfulAIAssistant:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.anyapi.ai/api/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}")

class EcommerceBot:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.anyapi.ai/api/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}")

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

class 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']}")

import 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}
)

import 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