Spaces:
Runtime error
Runtime error
| #!/usr/bin/env python3 | |
| """ | |
| Chat Service Client - For integrating with external applications. | |
| This client provides a simple interface for external applications to use | |
| the multi-language chat agent as a service. | |
| """ | |
| import requests | |
| import json | |
| import time | |
| from typing import Dict, List, Optional, Any | |
| from datetime import datetime | |
| class ChatServiceClient: | |
| """ | |
| Client for interacting with the Chat Agent Service. | |
| This client handles session management, message processing, and | |
| maintains conversation context for external applications. | |
| """ | |
| def __init__(self, base_url: str = "http://localhost:5000", | |
| app_name: str = "ExternalApp", timeout: int = 30): | |
| """ | |
| Initialize the chat service client. | |
| Args: | |
| base_url: Base URL of the chat service | |
| app_name: Name of your application (for tracking) | |
| timeout: Request timeout in seconds | |
| """ | |
| self.base_url = base_url.rstrip('/') | |
| self.app_name = app_name | |
| self.timeout = timeout | |
| self.api_base = f"{self.base_url}/api/v1/chat" | |
| # Session cache for the client | |
| self._sessions = {} | |
| def create_session(self, user_id: str, language: str = "python", | |
| metadata: Optional[Dict] = None) -> Dict[str, Any]: | |
| """ | |
| Create a new chat session for a user. | |
| Args: | |
| user_id: Unique identifier for the user in your app | |
| language: Programming language context | |
| metadata: Additional metadata about the session | |
| Returns: | |
| Dict containing session information | |
| Raises: | |
| Exception: If session creation fails | |
| """ | |
| url = f"{self.api_base}/sessions" | |
| headers = { | |
| "X-User-ID": user_id, | |
| "Content-Type": "application/json" | |
| } | |
| payload = { | |
| "language": language, | |
| "metadata": { | |
| "source": self.app_name, | |
| "created_by": "chat_service_client", | |
| **(metadata or {}) | |
| } | |
| } | |
| try: | |
| response = requests.post(url, headers=headers, json=payload, timeout=self.timeout) | |
| response.raise_for_status() | |
| session_data = response.json() | |
| # Cache session locally | |
| self._sessions[session_data['session_id']] = { | |
| 'user_id': user_id, | |
| 'language': language, | |
| 'created_at': session_data['created_at'], | |
| 'message_count': session_data['message_count'] | |
| } | |
| return session_data | |
| except requests.exceptions.RequestException as e: | |
| raise Exception(f"Failed to create session: {e}") | |
| def send_message(self, session_id: str, message: str, | |
| language: Optional[str] = None) -> Dict[str, Any]: | |
| """ | |
| Send a message to the chat agent. | |
| Args: | |
| session_id: Session identifier | |
| message: User's message | |
| language: Optional language override | |
| Returns: | |
| Dict containing the response and metadata | |
| Raises: | |
| Exception: If message processing fails | |
| """ | |
| # Get session info for user_id | |
| if session_id not in self._sessions: | |
| # Try to get session info from API | |
| session_info = self.get_session(session_id) | |
| if not session_info: | |
| raise Exception(f"Session {session_id} not found") | |
| user_id = self._sessions[session_id]['user_id'] | |
| url = f"{self.api_base}/sessions/{session_id}/message" | |
| headers = { | |
| "X-User-ID": user_id, | |
| "Content-Type": "application/json" | |
| } | |
| payload = { | |
| "content": message, | |
| "timestamp": datetime.utcnow().isoformat() | |
| } | |
| if language: | |
| payload["language"] = language | |
| try: | |
| response = requests.post(url, headers=headers, json=payload, timeout=self.timeout) | |
| response.raise_for_status() | |
| result = response.json() | |
| # Update local session cache | |
| if session_id in self._sessions: | |
| self._sessions[session_id]['message_count'] += 1 | |
| return result | |
| except requests.exceptions.RequestException as e: | |
| raise Exception(f"Failed to send message: {e}") | |
| def get_session(self, session_id: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| Get session information. | |
| Args: | |
| session_id: Session identifier | |
| Returns: | |
| Dict containing session information or None if not found | |
| """ | |
| if session_id in self._sessions: | |
| user_id = self._sessions[session_id]['user_id'] | |
| else: | |
| # We need user_id to make the request, but we don't have it | |
| # This is a limitation - in practice, you'd store user_id with session_id | |
| return None | |
| url = f"{self.api_base}/sessions/{session_id}" | |
| headers = { | |
| "X-User-ID": user_id, | |
| "Content-Type": "application/json" | |
| } | |
| try: | |
| response = requests.get(url, headers=headers, timeout=self.timeout) | |
| if response.status_code == 404: | |
| return None | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.exceptions.RequestException: | |
| return None | |
| def get_chat_history(self, session_id: str, limit: int = 50) -> List[Dict[str, Any]]: | |
| """ | |
| Get chat history for a session. | |
| Args: | |
| session_id: Session identifier | |
| limit: Maximum number of messages to retrieve | |
| Returns: | |
| List of messages | |
| """ | |
| if session_id not in self._sessions: | |
| raise Exception(f"Session {session_id} not found in cache") | |
| user_id = self._sessions[session_id]['user_id'] | |
| url = f"{self.api_base}/sessions/{session_id}/history" | |
| headers = { | |
| "X-User-ID": user_id, | |
| "Content-Type": "application/json" | |
| } | |
| params = { | |
| "recent_only": "true", | |
| "limit": limit | |
| } | |
| try: | |
| response = requests.get(url, headers=headers, params=params, timeout=self.timeout) | |
| response.raise_for_status() | |
| result = response.json() | |
| return result.get('messages', []) | |
| except requests.exceptions.RequestException as e: | |
| raise Exception(f"Failed to get chat history: {e}") | |
| def switch_language(self, session_id: str, language: str) -> Dict[str, Any]: | |
| """ | |
| Switch the programming language context for a session. | |
| Args: | |
| session_id: Session identifier | |
| language: New programming language | |
| Returns: | |
| Dict containing switch confirmation | |
| """ | |
| if session_id not in self._sessions: | |
| raise Exception(f"Session {session_id} not found in cache") | |
| user_id = self._sessions[session_id]['user_id'] | |
| url = f"{self.api_base}/sessions/{session_id}/language" | |
| headers = { | |
| "X-User-ID": user_id, | |
| "Content-Type": "application/json" | |
| } | |
| payload = { | |
| "language": language | |
| } | |
| try: | |
| response = requests.put(url, headers=headers, json=payload, timeout=self.timeout) | |
| response.raise_for_status() | |
| result = response.json() | |
| # Update local cache | |
| if session_id in self._sessions: | |
| self._sessions[session_id]['language'] = language | |
| return result | |
| except requests.exceptions.RequestException as e: | |
| raise Exception(f"Failed to switch language: {e}") | |
| def delete_session(self, session_id: str) -> bool: | |
| """ | |
| Delete a chat session. | |
| Args: | |
| session_id: Session identifier | |
| Returns: | |
| True if successful, False otherwise | |
| """ | |
| if session_id not in self._sessions: | |
| return False | |
| user_id = self._sessions[session_id]['user_id'] | |
| url = f"{self.api_base}/sessions/{session_id}" | |
| headers = { | |
| "X-User-ID": user_id, | |
| "Content-Type": "application/json" | |
| } | |
| try: | |
| response = requests.delete(url, headers=headers, timeout=self.timeout) | |
| response.raise_for_status() | |
| # Remove from local cache | |
| if session_id in self._sessions: | |
| del self._sessions[session_id] | |
| return True | |
| except requests.exceptions.RequestException: | |
| return False | |
| def health_check(self) -> Dict[str, Any]: | |
| """ | |
| Check if the chat service is healthy. | |
| Returns: | |
| Dict containing health status | |
| """ | |
| url = f"{self.api_base}/health" | |
| try: | |
| response = requests.get(url, timeout=self.timeout) | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.exceptions.RequestException as e: | |
| return {"status": "unhealthy", "error": str(e)} | |
| # Convenience class for managing multiple user sessions | |
| class MultiUserChatManager: | |
| """ | |
| Manager for handling multiple user sessions in an external application. | |
| This class provides a higher-level interface for managing chat sessions | |
| across multiple users in your application. | |
| """ | |
| def __init__(self, chat_service_url: str = "http://localhost:5000", | |
| app_name: str = "ExternalApp"): | |
| """ | |
| Initialize the multi-user chat manager. | |
| Args: | |
| chat_service_url: URL of the chat service | |
| app_name: Name of your application | |
| """ | |
| self.client = ChatServiceClient(chat_service_url, app_name) | |
| self.user_sessions = {} # user_id -> session_id mapping | |
| def start_chat_for_user(self, user_id: str, language: str = "python", | |
| user_metadata: Optional[Dict] = None) -> str: | |
| """ | |
| Start a new chat session for a user. | |
| Args: | |
| user_id: Unique user identifier in your app | |
| language: Programming language context | |
| user_metadata: Additional user metadata | |
| Returns: | |
| Session ID for the created session | |
| """ | |
| # End existing session if any | |
| if user_id in self.user_sessions: | |
| self.end_chat_for_user(user_id) | |
| # Create new session | |
| session_data = self.client.create_session(user_id, language, user_metadata) | |
| session_id = session_data['session_id'] | |
| # Store mapping | |
| self.user_sessions[user_id] = session_id | |
| return session_id | |
| def send_user_message(self, user_id: str, message: str, | |
| language: Optional[str] = None) -> Dict[str, Any]: | |
| """ | |
| Send a message for a specific user. | |
| Args: | |
| user_id: User identifier | |
| message: User's message | |
| language: Optional language override | |
| Returns: | |
| Response from the chat agent | |
| """ | |
| if user_id not in self.user_sessions: | |
| raise Exception(f"No active session for user {user_id}") | |
| session_id = self.user_sessions[user_id] | |
| return self.client.send_message(session_id, message, language) | |
| def get_user_history(self, user_id: str, limit: int = 50) -> List[Dict[str, Any]]: | |
| """ | |
| Get chat history for a specific user. | |
| Args: | |
| user_id: User identifier | |
| limit: Maximum number of messages | |
| Returns: | |
| List of messages | |
| """ | |
| if user_id not in self.user_sessions: | |
| return [] | |
| session_id = self.user_sessions[user_id] | |
| return self.client.get_chat_history(session_id, limit) | |
| def switch_user_language(self, user_id: str, language: str) -> Dict[str, Any]: | |
| """ | |
| Switch programming language for a user's session. | |
| Args: | |
| user_id: User identifier | |
| language: New programming language | |
| Returns: | |
| Switch confirmation | |
| """ | |
| if user_id not in self.user_sessions: | |
| raise Exception(f"No active session for user {user_id}") | |
| session_id = self.user_sessions[user_id] | |
| return self.client.switch_language(session_id, language) | |
| def end_chat_for_user(self, user_id: str) -> bool: | |
| """ | |
| End chat session for a specific user. | |
| Args: | |
| user_id: User identifier | |
| Returns: | |
| True if successful | |
| """ | |
| if user_id not in self.user_sessions: | |
| return True | |
| session_id = self.user_sessions[user_id] | |
| success = self.client.delete_session(session_id) | |
| if success: | |
| del self.user_sessions[user_id] | |
| return success | |
| def get_active_users(self) -> List[str]: | |
| """ | |
| Get list of users with active chat sessions. | |
| Returns: | |
| List of user IDs | |
| """ | |
| return list(self.user_sessions.keys()) | |
| if __name__ == "__main__": | |
| # Example usage | |
| print("🧪 Testing Chat Service Client") | |
| # Initialize client | |
| client = ChatServiceClient("http://localhost:5000", "TestApp") | |
| # Check service health | |
| health = client.health_check() | |
| print(f"Service health: {health}") | |
| if health.get('status') == 'healthy': | |
| # Create session | |
| session = client.create_session("test-user-123", "python") | |
| print(f"Created session: {session['session_id']}") | |
| # Send message | |
| response = client.send_message(session['session_id'], "What is Python?") | |
| print(f"Response: {response['response'][:100]}...") | |
| # Switch language | |
| switch_result = client.switch_language(session['session_id'], "javascript") | |
| print(f"Language switched: {switch_result}") | |
| # Send another message | |
| response2 = client.send_message(session['session_id'], "What is JavaScript?") | |
| print(f"JS Response: {response2['response'][:100]}...") | |
| # Get history | |
| history = client.get_chat_history(session['session_id']) | |
| print(f"History length: {len(history)} messages") | |
| # Clean up | |
| client.delete_session(session['session_id']) | |
| print("Session deleted") | |
| else: | |
| print("❌ Chat service is not healthy") |