Spaces:
Paused
Paused
File size: 4,539 Bytes
a5784e9 | 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 | import time
import requests
def verify_native_fc_e2e():
"""
E2E verification script for Native Function Calling.
This script connects to a running local proxy and sends a request with tool definitions
to verify that the proxy correctly handles the request and returns tool calls.
"""
base_url = "http://127.0.0.1:2048"
chat_url = f"{base_url}/v1/chat/completions"
health_url = f"{base_url}/health"
print("--- Native Function Calling E2E Verification ---")
# 1. Check if proxy is running
print(f"Checking proxy health at {health_url}...")
try:
health_resp = requests.get(health_url, timeout=5)
if health_resp.status_code == 200:
print(f"Proxy is UP. Status: {health_resp.json().get('status', 'Unknown')}")
else:
print(f"Proxy health check failed with status {health_resp.status_code}.")
print(
"Please ensure the proxy server is running before executing this script."
)
return
except requests.exceptions.ConnectionError:
print(f"CRITICAL: Could not connect to proxy at {base_url}.")
print("Please start the proxy server (e.g., python server.py) first.")
return
except Exception as e:
print(f"Warning: Health check failed with error: {e}")
# Continue anyway, health check might not be implemented exactly as expected
# 2. Prepare request with tools
headers = {"Content-Type": "application/json", "Authorization": "Bearer any-key"}
payload = {
"model": "gemini-3-flash-preview",
"messages": [
{"role": "user", "content": "What is the weather in Tokyo and Paris?"}
],
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["location"],
},
},
}
],
"tool_choice": "auto",
}
print(f"\nSending request to {chat_url}...")
print(f"Tools defined: {[t['function']['name'] for t in payload['tools']]}")
start_time = time.time()
try:
response = requests.post(chat_url, headers=headers, json=payload, timeout=60)
duration = time.time() - start_time
print(f"Response received in {duration:.2f}s (Status: {response.status_code})")
if response.status_code == 200:
data = response.json()
message = data["choices"][0]["message"]
if "tool_calls" in message:
tool_calls = message["tool_calls"]
print(f"\nSUCCESS: Received {len(tool_calls)} tool call(s)")
for i, call in enumerate(tool_calls):
func_name = call["function"]["name"]
args = call["function"]["arguments"]
print(f" Call #{i + 1}: {func_name}({args})")
# Verify format
if all(k in tool_calls[0] for k in ["id", "type", "function"]):
print(" Format verification: PASSED (OpenAI compatible)")
else:
print(" Format verification: FAILED (Missing required fields)")
else:
print("\nFAILURE: No tool calls received.")
if "content" in message and message["content"]:
print(f" Assistant Content: {message['content']}")
else:
print(" Response was empty.")
else:
print(f"\nERROR: Request failed with status {response.status_code}")
print(f"Response: {response.text}")
except requests.exceptions.Timeout:
print("\n❌ ERROR: Request timed out after 60s.")
except Exception as e:
print(f"\n❌ ERROR: An unexpected error occurred: {e}")
if __name__ == "__main__":
verify_native_fc_e2e()
|