tool_handler.py

# tool_handler.py
import warnings, logging, requests
from config import BASE_URL

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Suppress the InsecureRequestWarning
warnings.filterwarnings("ignore", category=requests.urllib3.exceptions.InsecureRequestWarning)

# Define tools
tools = [
    {
        "name": "get_user",
        "description": "Looks up a user by email, phone, or username.",
        "input_schema": {
            "type": "object",
            "properties": {
                "key": {
                    "type": "string",
                    "enum": ["email", "phone", "username"],
                    "description": "The attribute to search for a user by (email, phone, or username)."
                },
                "value": {
                    "type": "string",
                    "description": "The value to match for the specified attribute."
                }
            },
            "required": ["key", "value"]
        }
    },
    {
        "name": "get_order_by_id",
        "description": "Retrieves the details of a specific order based on the order ID.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "The unique identifier for the order."
                }
            },
            "required": ["order_id"]
        }
    },
    {
        "name": "get_customer_orders",
        "description": "Retrieves the list of orders belonging to a user based on a user's customer id.",
        "input_schema": {
            "type": "object",
            "properties": {
                "customer_id": {
                    "type": "string",
                    "description": "The customer_id belonging to the user"
                }
            },
            "required": ["customer_id"]
        }
    },
    {
        "name": "cancel_order",
        "description": "Cancels an order based on a provided order_id. Only orders that are 'processing' can be cancelled.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "The order_id pertaining to a particular order"
                }
            },
            "required": ["order_id"]
        }
    },
    {
        "name": "update_user_contact",
        "description": "Updates a user's email and/or phone number.",
        "input_schema": {
            "type": "object",
            "properties": {
                "user_id": {
                    "type": "string",
                    "description": "The ID of the user"
                },
                "email": {
                    "type": "string",
                    "description": "The new email address of the user"
                },
                "phone": {
                    "type": "string",
                    "description": "The new phone number of the user"
                }
            },
            "required": ["user_id"]
        }
    },
    {
        "name": "get_user_info",
        "description": "Retrieves a user's information along with their order history based on email, phone, or username.",
        "input_schema": {
            "type": "object",
            "properties": {
                "key": {
                    "type": "string",
                    "enum": ["email", "phone", "username"],
                    "description": "The attribute to search for a user by (email, phone, or username)."
                },
                "value": {
                    "type": "string",
                    "description": "The value to match for the specified attribute."
                }
            },
            "required": ["key", "value"]
        }
    }
]

def process_tool_call(tool_name, tool_input):
    tool_endpoints = {
        "get_user": "get_user",
        "get_order_by_id": "get_order_by_id",
        "get_customer_orders": "get_customer_orders",
        "cancel_order": "cancel_order",
        "update_user_contact": "update_user",
        "get_user_info": "get_user_info"
    }
    
    if tool_name in tool_endpoints:
        logger.info(f"tool_handler Calling tool: {tool_name}")
        response = requests.post(f"{BASE_URL}/{tool_endpoints[tool_name]}", json=tool_input, verify=False)
    else:
        logger.error(f"tool_handle Invalid tool name: {tool_name}")
        return {"error": "Invalid tool name"}

    if response.status_code == 200:
        logger.info(f"tool_handler Tool call successful: {response.json()}")
        return response.json()
    else:
        logger.error(f"tool_handler Tool call failed: {response.text}")
        return {"error": response.text}