Spaces:
Sleeping
Sleeping
File size: 4,858 Bytes
cca16c9 15bcec8 cca16c9 15bcec8 cca16c9 15bcec8 cca16c9 15bcec8 48e7fd1 15bcec8 cca16c9 15bcec8 cca16c9 15bcec8 cca16c9 15bcec8 cca16c9 15bcec8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
# 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}
|