File size: 6,579 Bytes
b30f068 | 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 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 | #!/usr/bin/env python3
"""
Example client for the Agentic AI System API.
This demonstrates how to interact with the real-world API.
"""
import requests
import json
import time
from typing import Dict, Any
class AgenticAIClient:
"""Client for the Agentic AI System API."""
def __init__(self, base_url: str = "http://localhost:8000", api_key: str = "demo-key-12345"):
"""Initialize the client."""
self.base_url = base_url.rstrip('/')
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def health_check(self) -> Dict[str, Any]:
"""Check system health."""
response = requests.get(f"{self.base_url}/health")
return response.json()
def process_query(self, query: str, include_debug: bool = False, max_api_calls: int = 5) -> Dict[str, Any]:
"""Process a natural language query."""
data = {
"query": query,
"include_debug": include_debug,
"max_api_calls": max_api_calls,
"metadata": {
"client": "python_client",
"timestamp": time.time()
}
}
response = requests.post(
f"{self.base_url}/query",
headers=self.headers,
json=data
)
return response.json()
def list_apis(self) -> Dict[str, Any]:
"""List available API operations."""
response = requests.get(
f"{self.base_url}/apis",
headers=self.headers
)
return response.json()
def register_api(self, name: str, method: str, endpoint: str, description: str, tags: list = None) -> Dict[str, Any]:
"""Register a new API operation."""
data = {
"name": name,
"method": method,
"endpoint": endpoint,
"description": description,
"tags": tags or [],
"parameters": []
}
response = requests.post(
f"{self.base_url}/apis",
headers=self.headers,
json=data
)
return response.json()
def extract_keywords(self, query: str) -> Dict[str, Any]:
"""Extract keywords from a query."""
response = requests.get(
f"{self.base_url}/keywords/{query}",
headers=self.headers
)
return response.json()
def get_stats(self) -> Dict[str, Any]:
"""Get system statistics (requires admin permissions)."""
response = requests.get(
f"{self.base_url}/stats",
headers=self.headers
)
return response.json()
def demo_client():
"""Demonstrate the API client."""
print("🚀 Agentic AI System - API Client Demo")
print("=" * 50)
# Initialize client
client = AgenticAIClient()
try:
# Health check
print("\\n1. Health Check:")
health = client.health_check()
print(f" Status: {health['status']}")
print(f" Uptime: {health['uptime']:.2f}s")
print(f" Total Requests: {health['total_requests']}")
# List available APIs
print("\\n2. Available APIs:")
apis = client.list_apis()
for api in apis[:3]:
print(f" • {api['name']} ({api['method']} {api['endpoint']})")
# Process queries
test_queries = [
"Find machine learning datasets for image classification",
"Get user profile information",
"Retrieve CDMS labels for natural language processing"
]
print("\\n3. Processing Queries:")
for i, query in enumerate(test_queries, 1):
print(f"\\n Query {i}: {query}")
result = client.process_query(query, include_debug=True)
if result['success']:
print(f" ✅ Success ({result['processing_time']:.3f}s)")
# Show results summary
results = result['results']
query_info = results.get('query', {})
api_matches = results.get('api_matches', {})
execution_results = results.get('results', {})
print(f" 📝 Keywords: {query_info.get('keywords', [])[:3]}")
print(f" 🎯 Intent: {query_info.get('intent', 'unknown')}")
print(f" 🔗 API Matches: {api_matches.get('count', 0)}")
print(f" ⚡ Executed: {execution_results.get('executed_count', 0)} APIs")
print(f" 📊 Summary: {results.get('summary', 'N/A')[:80]}...")
else:
print(f" ❌ Failed: {result.get('error_message', 'Unknown error')}")
# Keyword extraction
print("\\n4. Keyword Extraction:")
keywords_result = client.extract_keywords("Create a new machine learning model for classification")
print(f" Keywords: {keywords_result['keywords'][:5]}")
# Register a new API
print("\\n5. Registering New API:")
try:
register_result = client.register_api(
name="testNewAPI",
method="POST",
endpoint="/test/new",
description="Test API for demonstration",
tags=["test", "demo"]
)
print(f" ✅ {register_result['message']}")
except Exception as e:
print(f" ⚠️ Registration: {e}")
print("\\n" + "=" * 50)
print("🎉 API Client Demo Completed Successfully!")
print("\\n💡 Next Steps:")
print(" • Deploy to cloud (AWS, GCP, Azure)")
print(" • Add real API integrations")
print(" • Implement authentication & rate limiting")
print(" • Add monitoring & analytics")
print(" • Scale with container orchestration")
except requests.exceptions.ConnectionError:
print("❌ Error: Could not connect to API server")
print("💡 Make sure to start the server first:")
print(" python api_server.py")
except Exception as e:
print(f"❌ Error: {e}")
if __name__ == "__main__":
demo_client()
|