WebashalarForML/scratch_chat
0
1#!/usr/bin/env python3
2"""
3Chat Service Client - For integrating with external applications.
4
5This client provides a simple interface for external applications to use
6the multi-language chat agent as a service.
7"""
8
9import requests
10import json
11import time
12from typing import Dict, List, Optional, Any
13from datetime import datetime
14
15
16class ChatServiceClient:
17 """
18 Client for interacting with the Chat Agent Service.
19
20 This client handles session management, message processing, and
21 maintains conversation context for external applications.
22 """
23
24 def __init__(self, base_url: str = "http://localhost:5000",
25 app_name: str = "ExternalApp", timeout: int = 30):
26 """
27 Initialize the chat service client.
28
29 Args:
30 base_url: Base URL of the chat service
31 app_name: Name of your application (for tracking)
32 timeout: Request timeout in seconds
33 """
34 self.base_url = base_url.rstrip('/')
35 self.app_name = app_name
36 self.timeout = timeout
37 self.api_base = f"{self.base_url}/api/v1/chat"
38
39 # Session cache for the client
40 self._sessions = {}
41
42 def create_session(self, user_id: str, language: str = "python",
43 metadata: Optional[Dict] = None) -> Dict[str, Any]:
44 """
45 Create a new chat session for a user.
46
47 Args:
48 user_id: Unique identifier for the user in your app
49 language: Programming language context
50 metadata: Additional metadata about the session
51
52 Returns:
53 Dict containing session information
54
55 Raises:
56 Exception: If session creation fails
57 """
58 url = f"{self.api_base}/sessions"
59
60 headers = {
61 "X-User-ID": user_id,
62 "Content-Type": "application/json"
63 }
64
65 payload = {
66 "language": language,
67 "metadata": {
68 "source": self.app_name,
69 "created_by": "chat_service_client",
70 **(metadata or {})
71 }
72 }
73
74 try:
75 response = requests.post(url, headers=headers, json=payload, timeout=self.timeout)
76 response.raise_for_status()
77
78 session_data = response.json()
79
80 # Cache session locally
81 self._sessions[session_data['session_id']] = {
82 'user_id': user_id,
83 'language': language,
84 'created_at': session_data['created_at'],
85 'message_count': session_data['message_count']
86 }
87
88 return session_data
89
90 except requests.exceptions.RequestException as e:
91 raise Exception(f"Failed to create session: {e}")
92
93 def send_message(self, session_id: str, message: str,
94 language: Optional[str] = None) -> Dict[str, Any]:
95 """
96 Send a message to the chat agent.
97
98 Args:
99 session_id: Session identifier
100 message: User's message
101 language: Optional language override
102
103 Returns:
104 Dict containing the response and metadata
105
106 Raises:
107 Exception: If message processing fails
108 """
109 # Get session info for user_id
110 if session_id not in self._sessions:
111 # Try to get session info from API
112 session_info = self.get_session(session_id)
113 if not session_info:
114 raise Exception(f"Session {session_id} not found")
115
116 user_id = self._sessions[session_id]['user_id']
117
118 url = f"{self.api_base}/sessions/{session_id}/message"
119
120 headers = {
121 "X-User-ID": user_id,
122 "Content-Type": "application/json"
123 }
124
125 payload = {
126 "content": message,
127 "timestamp": datetime.utcnow().isoformat()
128 }
129
130 if language:
131 payload["language"] = language
132
133 try:
134 response = requests.post(url, headers=headers, json=payload, timeout=self.timeout)
135 response.raise_for_status()
136
137 result = response.json()
138
139 # Update local session cache
140 if session_id in self._sessions:
141 self._sessions[session_id]['message_count'] += 1
142
143 return result
144
145 except requests.exceptions.RequestException as e:
146 raise Exception(f"Failed to send message: {e}")
147
148 def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
149 """
150 Get session information.
151
152 Args:
153 session_id: Session identifier
154
155 Returns:
156 Dict containing session information or None if not found
157 """
158 if session_id in self._sessions:
159 user_id = self._sessions[session_id]['user_id']
160 else:
161 # We need user_id to make the request, but we don't have it
162 # This is a limitation - in practice, you'd store user_id with session_id
163 return None
164
165 url = f"{self.api_base}/sessions/{session_id}"
166
167 headers = {
168 "X-User-ID": user_id,
169 "Content-Type": "application/json"
170 }
171
172 try:
173 response = requests.get(url, headers=headers, timeout=self.timeout)
174
175 if response.status_code == 404:
176 return None
177
178 response.raise_for_status()
179 return response.json()
180
181 except requests.exceptions.RequestException:
182 return None
183
184 def get_chat_history(self, session_id: str, limit: int = 50) -> List[Dict[str, Any]]:
185 """
186 Get chat history for a session.
187
188 Args:
189 session_id: Session identifier
190 limit: Maximum number of messages to retrieve
191
192 Returns:
193 List of messages
194 """
195 if session_id not in self._sessions:
196 raise Exception(f"Session {session_id} not found in cache")
197
198 user_id = self._sessions[session_id]['user_id']
199
200 url = f"{self.api_base}/sessions/{session_id}/history"
201
202 headers = {
203 "X-User-ID": user_id,
204 "Content-Type": "application/json"
205 }
206
207 params = {
208 "recent_only": "true",
209 "limit": limit
210 }
211
212 try:
213 response = requests.get(url, headers=headers, params=params, timeout=self.timeout)
214 response.raise_for_status()
215
216 result = response.json()
217 return result.get('messages', [])
218
219 except requests.exceptions.RequestException as e:
220 raise Exception(f"Failed to get chat history: {e}")
221
222 def switch_language(self, session_id: str, language: str) -> Dict[str, Any]:
223 """
224 Switch the programming language context for a session.
225
226 Args:
227 session_id: Session identifier
228 language: New programming language
229
230 Returns:
231 Dict containing switch confirmation
232 """
233 if session_id not in self._sessions:
234 raise Exception(f"Session {session_id} not found in cache")
235
236 user_id = self._sessions[session_id]['user_id']
237
238 url = f"{self.api_base}/sessions/{session_id}/language"
239
240 headers = {
241 "X-User-ID": user_id,
242 "Content-Type": "application/json"
243 }
244
245 payload = {
246 "language": language
247 }
248
249 try:
250 response = requests.put(url, headers=headers, json=payload, timeout=self.timeout)
251 response.raise_for_status()
252
253 result = response.json()
254
255 # Update local cache
256 if session_id in self._sessions:
257 self._sessions[session_id]['language'] = language
258
259 return result
260
261 except requests.exceptions.RequestException as e:
262 raise Exception(f"Failed to switch language: {e}")
263
264 def delete_session(self, session_id: str) -> bool:
265 """
266 Delete a chat session.
267
268 Args:
269 session_id: Session identifier
270
271 Returns:
272 True if successful, False otherwise
273 """
274 if session_id not in self._sessions:
275 return False
276
277 user_id = self._sessions[session_id]['user_id']
278
279 url = f"{self.api_base}/sessions/{session_id}"
280
281 headers = {
282 "X-User-ID": user_id,
283 "Content-Type": "application/json"
284 }
285
286 try:
287 response = requests.delete(url, headers=headers, timeout=self.timeout)
288 response.raise_for_status()
289
290 # Remove from local cache
291 if session_id in self._sessions:
292 del self._sessions[session_id]
293
294 return True
295
296 except requests.exceptions.RequestException:
297 return False
298
299 def health_check(self) -> Dict[str, Any]:
300 """
301 Check if the chat service is healthy.
302
303 Returns:
304 Dict containing health status
305 """
306 url = f"{self.api_base}/health"
307
308 try:
309 response = requests.get(url, timeout=self.timeout)
310 response.raise_for_status()
311 return response.json()
312
313 except requests.exceptions.RequestException as e:
314 return {"status": "unhealthy", "error": str(e)}
315
316
317# Convenience class for managing multiple user sessions
318class MultiUserChatManager:
319 """
320 Manager for handling multiple user sessions in an external application.
321
322 This class provides a higher-level interface for managing chat sessions
323 across multiple users in your application.
324 """
325
326 def __init__(self, chat_service_url: str = "http://localhost:5000",
327 app_name: str = "ExternalApp"):
328 """
329 Initialize the multi-user chat manager.
330
331 Args:
332 chat_service_url: URL of the chat service
333 app_name: Name of your application
334 """
335 self.client = ChatServiceClient(chat_service_url, app_name)
336 self.user_sessions = {} # user_id -> session_id mapping
337
338 def start_chat_for_user(self, user_id: str, language: str = "python",
339 user_metadata: Optional[Dict] = None) -> str:
340 """
341 Start a new chat session for a user.
342
343 Args:
344 user_id: Unique user identifier in your app
345 language: Programming language context
346 user_metadata: Additional user metadata
347
348 Returns:
349 Session ID for the created session
350 """
351 # End existing session if any
352 if user_id in self.user_sessions:
353 self.end_chat_for_user(user_id)
354
355 # Create new session
356 session_data = self.client.create_session(user_id, language, user_metadata)
357 session_id = session_data['session_id']
358
359 # Store mapping
360 self.user_sessions[user_id] = session_id
361
362 return session_id
363
364 def send_user_message(self, user_id: str, message: str,
365 language: Optional[str] = None) -> Dict[str, Any]:
366 """
367 Send a message for a specific user.
368
369 Args:
370 user_id: User identifier
371 message: User's message
372 language: Optional language override
373
374 Returns:
375 Response from the chat agent
376 """
377 if user_id not in self.user_sessions:
378 raise Exception(f"No active session for user {user_id}")
379
380 session_id = self.user_sessions[user_id]
381 return self.client.send_message(session_id, message, language)
382
383 def get_user_history(self, user_id: str, limit: int = 50) -> List[Dict[str, Any]]:
384 """
385 Get chat history for a specific user.
386
387 Args:
388 user_id: User identifier
389 limit: Maximum number of messages
390
391 Returns:
392 List of messages
393 """
394 if user_id not in self.user_sessions:
395 return []
396
397 session_id = self.user_sessions[user_id]
398 return self.client.get_chat_history(session_id, limit)
399
400 def switch_user_language(self, user_id: str, language: str) -> Dict[str, Any]:
401 """
402 Switch programming language for a user's session.
403
404 Args:
405 user_id: User identifier
406 language: New programming language
407
408 Returns:
409 Switch confirmation
410 """
411 if user_id not in self.user_sessions:
412 raise Exception(f"No active session for user {user_id}")
413
414 session_id = self.user_sessions[user_id]
415 return self.client.switch_language(session_id, language)
416
417 def end_chat_for_user(self, user_id: str) -> bool:
418 """
419 End chat session for a specific user.
420
421 Args:
422 user_id: User identifier
423
424 Returns:
425 True if successful
426 """
427 if user_id not in self.user_sessions:
428 return True
429
430 session_id = self.user_sessions[user_id]
431 success = self.client.delete_session(session_id)
432
433 if success:
434 del self.user_sessions[user_id]
435
436 return success
437
438 def get_active_users(self) -> List[str]:
439 """
440 Get list of users with active chat sessions.
441
442 Returns:
443 List of user IDs
444 """
445 return list(self.user_sessions.keys())
446
447
448if __name__ == "__main__":
449 # Example usage
450 print("🧪 Testing Chat Service Client")
451
452 # Initialize client
453 client = ChatServiceClient("http://localhost:5000", "TestApp")
454
455 # Check service health
456 health = client.health_check()
457 print(f"Service health: {health}")
458
459 if health.get('status') == 'healthy':
460 # Create session
461 session = client.create_session("test-user-123", "python")
462 print(f"Created session: {session['session_id']}")
463
464 # Send message
465 response = client.send_message(session['session_id'], "What is Python?")
466 print(f"Response: {response['response'][:100]}...")
467
468 # Switch language
469 switch_result = client.switch_language(session['session_id'], "javascript")
470 print(f"Language switched: {switch_result}")
471
472 # Send another message
473 response2 = client.send_message(session['session_id'], "What is JavaScript?")
474 print(f"JS Response: {response2['response'][:100]}...")
475
476 # Get history
477 history = client.get_chat_history(session['session_id'])
478 print(f"History length: {len(history)} messages")
479
480 # Clean up
481 client.delete_session(session['session_id'])
482 print("Session deleted")
483 else:
484 print("❌ Chat service is not healthy")