# 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}