""" HTTP REST API handler for the Qwen Pipeline. Serves static files + REST APIs for profiles, conversations, recordings, ACRS proxy, TTS test, translation, and tool generation. """ import json import os import re import time import threading import base64 import urllib.parse from http.server import SimpleHTTPRequestHandler import requests import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) from dashscope.audio.qwen_tts_realtime import ( QwenTtsRealtime, QwenTtsRealtimeCallback, AudioFormat ) from openai import OpenAI # Diagnostic add-on: audio-piece timing tracer (safe no-op if unavailable). try: import audio_timing_trace except Exception: audio_timing_trace = None # These are set by the server at startup via init_http_routes() HTTP_DIR = None RECORDINGS_DIR = None PROFILES_COLLECTION = None CONVERSATIONS_COLLECTION = None API_KEY = None LLM_BASE = None LLM_MODEL = None TTS_MODEL = None TTS_VC_MODEL = None LANG_NAMES = None _cloned_voices_ref = None # reference to the mutable list in the main module def init_http_routes(*, http_dir, recordings_dir, profiles_collection, conversations_collection, api_key, llm_base, llm_model, tts_model, tts_vc_model, lang_names, cloned_voices_ref): """Initialize module-level references. Called once at server startup.""" global HTTP_DIR, RECORDINGS_DIR, PROFILES_COLLECTION, CONVERSATIONS_COLLECTION global API_KEY, LLM_BASE, LLM_MODEL, TTS_MODEL, TTS_VC_MODEL, LANG_NAMES global _cloned_voices_ref HTTP_DIR = http_dir RECORDINGS_DIR = recordings_dir PROFILES_COLLECTION = profiles_collection CONVERSATIONS_COLLECTION = conversations_collection API_KEY = api_key LLM_BASE = llm_base LLM_MODEL = llm_model TTS_MODEL = tts_model TTS_VC_MODEL = tts_vc_model LANG_NAMES = lang_names _cloned_voices_ref = cloned_voices_ref class PipelineHTTPHandler(SimpleHTTPRequestHandler): """Serves static files + REST API for ChromaDB profile storage.""" def __init__(self, *args, **kwargs): super().__init__(*args, directory=HTTP_DIR, **kwargs) # Suppress noisy static file logs, keep API logs def log_message(self, format, *args): if self.path.startswith('/api/'): super().log_message(format, *args) def do_GET(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == '/api/profiles': return self._list_profiles() if parsed.path == '/api/conversations': return self._list_conversations() if parsed.path.startswith('/api/conversations/'): cid = urllib.parse.unquote(parsed.path[len('/api/conversations/'):]) return self._get_conversation(cid) if parsed.path.startswith('/api/recordings/'): cid = urllib.parse.unquote(parsed.path[len('/api/recordings/'):]) return self._get_recording(cid) if parsed.path == '/api/config-vars': return self._get_config_vars() if parsed.path == '/api/audio-timing': return self._audio_timing_list() if parsed.path.startswith('/api/audio-timing/'): cid = urllib.parse.unquote(parsed.path[len('/api/audio-timing/'):]) return self._audio_timing_series(cid) return super().do_GET() def do_POST(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == '/api/profiles': return self._save_profile() if parsed.path == '/api/acrs/sync': return self._acrs_sync() if parsed.path == '/api/acrs/create': return self._acrs_create() if parsed.path == '/api/tts-test': return self._tts_test() if parsed.path == '/api/translate': return self._translate() if parsed.path == '/api/tools/generate': return self._generate_tool() if parsed.path == '/api/tools/generate-handler': return self._generate_handler() if parsed.path == '/api/tools/save-handler': return self._save_handler() if parsed.path == '/api/tools/check-model': return self._check_model() if parsed.path == '/api/prompt/generate': return self._generate_prompt() if parsed.path.startswith('/api/recordings/'): cid = urllib.parse.unquote(parsed.path[len('/api/recordings/'):]) return self._save_recording(cid) self.send_error(404) def do_DELETE(self): parsed = urllib.parse.urlparse(self.path) if parsed.path.startswith('/api/profiles/'): name = urllib.parse.unquote(parsed.path[len('/api/profiles/'):]) return self._delete_profile(name) if parsed.path.startswith('/api/conversations/'): cid = urllib.parse.unquote(parsed.path[len('/api/conversations/'):]) return self._delete_conversation(cid) self.send_error(404) def _json_response(self, data, status=200): body = json.dumps(data).encode('utf-8') self.send_response(status) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', len(body)) self.end_headers() self.wfile.write(body) def _list_profiles(self): if not PROFILES_COLLECTION: return self._json_response({'error': 'ChromaDB not available'}, 503) try: result = PROFILES_COLLECTION.get() profiles = {} for i, pid in enumerate(result['ids']): try: profiles[pid] = json.loads(result['documents'][i]) except (json.JSONDecodeError, IndexError): pass self._json_response(profiles) except Exception as e: self._json_response({'error': str(e)}, 500) def _save_profile(self): if not PROFILES_COLLECTION: return self._json_response({'error': 'ChromaDB not available'}, 503) try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) name = body['name'] data = body['data'] PROFILES_COLLECTION.upsert( ids=[name], documents=[json.dumps(data)], embeddings=[[0.0]] ) print(f'[Profiles] Saved: {name}') self._json_response({'ok': True}) except Exception as e: print(f'[Profiles] Save error: {e}') self._json_response({'error': str(e)}, 500) def _delete_profile(self, name): if not PROFILES_COLLECTION: return self._json_response({'error': 'ChromaDB not available'}, 503) try: PROFILES_COLLECTION.delete(ids=[name]) print(f'[Profiles] Deleted: {name}') self._json_response({'ok': True}) except Exception as e: print(f'[Profiles] Delete error: {e}') self._json_response({'error': str(e)}, 500) # ── Conversations API ────────────────────────────────────────── def _list_conversations(self): if not CONVERSATIONS_COLLECTION: return self._json_response({'error': 'ChromaDB not available'}, 503) try: result = CONVERSATIONS_COLLECTION.get() conversations = [] for i, cid in enumerate(result['ids']): try: doc = json.loads(result['documents'][i]) rec_file = self._recording_path(cid) caller_file = rec_file.replace('.webm', '_caller.wav') _fa = doc.get('faithfulness') _fa_summary = None if isinstance(_fa, dict): _fa_summary = { 'status': _fa.get('status'), 'flagged_count': _fa.get('flagged_count', 0), } conversations.append({ 'id': cid, 'profile': doc.get('profile', ''), 'agent_a': doc.get('agent_a', ''), 'agent_b': doc.get('agent_b', ''), 'transfer_agents': doc.get('transfer_agents', []), 'message_count': doc.get('message_count', 0), 'has_recording': os.path.isfile(rec_file), 'has_caller_recording': os.path.isfile(caller_file), 'cost_usd': doc.get('cost_usd'), 'faithfulness': _fa_summary, }) except (json.JSONDecodeError, IndexError): pass conversations.sort(key=lambda c: c['id'], reverse=True) self._json_response(conversations) except Exception as e: self._json_response({'error': str(e)}, 500) def _get_conversation(self, cid): if not CONVERSATIONS_COLLECTION: return self._json_response({'error': 'ChromaDB not available'}, 503) try: result = CONVERSATIONS_COLLECTION.get(ids=[cid]) if result['ids']: doc = json.loads(result['documents'][0]) self._json_response(doc) else: self._json_response({'error': 'Not found'}, 404) except Exception as e: self._json_response({'error': str(e)}, 500) def _delete_conversation(self, cid): if not CONVERSATIONS_COLLECTION: return self._json_response({'error': 'ChromaDB not available'}, 503) try: CONVERSATIONS_COLLECTION.delete(ids=[cid]) # Also delete recordings if they exist rec_file = self._recording_path(cid) caller_file = rec_file.replace('.webm', '_caller.wav') for f in (rec_file, caller_file): if os.path.isfile(f): os.remove(f) self._json_response({'ok': True}) except Exception as e: self._json_response({'error': str(e)}, 500) @staticmethod def _recording_path(cid): """Return safe filesystem path for a recording.""" safe = re.sub(r'[^\w\-]', '_', cid) return os.path.join(RECORDINGS_DIR, safe + '.webm') def _save_recording(self, cid): """Save uploaded audio recording for a conversation.""" try: length = int(self.headers.get('Content-Length', 0)) if length == 0: return self._json_response({'error': 'No data'}, 400) data = self.rfile.read(length) filepath = self._recording_path(cid) with open(filepath, 'wb') as f: f.write(data) print(f'[Recording] Saved: {os.path.basename(filepath)} ({len(data)} bytes)') self._json_response({'ok': True}) except Exception as e: self._json_response({'error': str(e)}, 500) def _get_recording(self, cid): """Serve a recording file (browser .webm or caller .wav).""" if cid.endswith('_caller'): # Caller-only recording (server-side WAV) real_cid = cid[:-len('_caller')] filepath = self._recording_path(real_cid).replace('.webm', '_caller.wav') content_type = 'audio/wav' else: # Full call recording (browser-uploaded WebM) filepath = self._recording_path(cid) content_type = 'audio/webm' if not os.path.isfile(filepath): return self.send_error(404) try: filename = os.path.basename(filepath) with open(filepath, 'rb') as f: data = f.read() self.send_response(200) self.send_header('Content-Type', content_type) self.send_header('Content-Length', len(data)) self.send_header('Content-Disposition', f'attachment; filename="{filename}"') self.end_headers() self.wfile.write(data) except Exception as e: self._json_response({'error': str(e)}, 500) # ── Call config variables ──────────────────────────────────── def _get_config_vars(self): """Return last-seen config vars per profile for the dashboard.""" import __main__ _LAST_CONFIG_VARS = getattr(__main__, '_LAST_CONFIG_VARS', {}) print(f'[ConfigVars] GET /api/config-vars — profiles stored: {list(_LAST_CONFIG_VARS.keys())}') self._json_response({ 'builtins': ['CURRENTTIME', 'CURRENTDATE', 'CALLERPHONE', 'AGENTDATA', 'EXTENSIONS', 'OUTGOING'], 'lastConfig': _LAST_CONFIG_VARS }) # ── Audio-piece timing tracer (diagnostic add-on) ──────────── def _audio_timing_list(self): """Summary of recorded calls: recv (from cloud) vs send (to bridge) gaps.""" if not audio_timing_trace: return self._json_response({'error': 'tracer unavailable'}, 503) try: self._json_response(audio_timing_trace.get_calls()) except Exception as e: self._json_response({'error': str(e)}, 500) def _audio_timing_series(self, cid): """Full per-piece timeline for one call.""" if not audio_timing_trace: return self._json_response({'error': 'tracer unavailable'}, 503) try: data = audio_timing_trace.get_series(cid) if data is None: return self._json_response({'error': 'not found'}, 404) self._json_response(data) except Exception as e: self._json_response({'error': str(e)}, 500) # ── ACRS proxy ───────────────────────────────────────────── def _acrs_sync(self): """Proxy: fetch all ACRS entities and return to browser.""" try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) api_url = body.get('api_url', '') api_key = body.get('api_key', '') if not api_url or not api_key: return self._json_response({'error': 'api_url and api_key required'}, 400) base = api_url.rsplit('/dispatches', 1)[0] headers = {'X-API-KEY': api_key, 'Accept': 'application/json'} result = {} for entity in ['departments', 'categories', 'types', 'sub-categories', 'ratings']: try: resp = requests.get(f'{base}/{entity}', headers=headers, timeout=10, verify=False) result[entity.replace('-', '_')] = resp.json() if resp.status_code == 200 else [] except Exception: result[entity.replace('-', '_')] = [] self._json_response(result) except Exception as e: self._json_response({'error': str(e)}, 500) def _acrs_create(self): """Proxy: create an entity in ACRS.""" try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) api_url = body.get('api_url', '') api_key = body.get('api_key', '') entity_type = body.get('entity', '') data = body.get('data', {}) if not api_url or not api_key or not entity_type: return self._json_response({'error': 'api_url, api_key, entity required'}, 400) base = api_url.rsplit('/dispatches', 1)[0] headers = {'X-API-KEY': api_key, 'Content-Type': 'application/json'} resp = requests.post(f'{base}/{entity_type}', json=data, headers=headers, timeout=10, verify=False) resp_data = {} try: resp_data = resp.json() except Exception: resp_data = {'raw': resp.text[:500]} if resp.status_code in (200, 201): self._json_response(resp_data) else: self._json_response({'error': resp_data}, resp.status_code) except Exception as e: self._json_response({'error': str(e)}, 500) # ── TTS Test API ────────────────────────────────────────────── def _tts_test(self): """Synthesize text and return base64 PCM audio.""" try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) text = body.get('text', '').strip() voice = body.get('voice', 'Ethan') speech_rate = float(body.get('speech_rate', 1.0)) pitch_rate = float(body.get('pitch_rate', 1.0)) volume = int(body.get('volume', 50)) if not text: return self._json_response({'error': 'text is required'}, 400) audio_chunks = [] done_event = threading.Event() class CollectorCallback(QwenTtsRealtimeCallback): def on_open(self): pass def on_close(self, *args): done_event.set() def on_event(self, response): try: etype = response.get('type', '') if etype == 'response.audio.delta': audio = response.get('delta', '') if audio: audio_chunks.append(audio) elif etype == 'session.finished': done_event.set() except Exception: pass callback = CollectorCallback() # Resolve model: cloned voices need VC model tts_model = TTS_MODEL for cv in _cloned_voices_ref: if cv['id'] == voice: tts_model = cv.get('target_model', TTS_VC_MODEL) break tts = QwenTtsRealtime( model=tts_model, callback=callback, url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime' ) tts.connect() tts.update_session( voice=voice, response_format=AudioFormat.PCM_24000HZ_MONO_16BIT, mode='server_commit', speech_rate=speech_rate, pitch_rate=pitch_rate, volume=volume, ) chunk_size = 80 for i in range(0, len(text), chunk_size): tts.append_text(text[i:i + chunk_size]) time.sleep(0.03) tts.finish() done_event.wait(timeout=30) # Concatenate all base64 audio chunks all_audio = b'' for chunk in audio_chunks: all_audio += base64.b64decode(chunk) combined_b64 = base64.b64encode(all_audio).decode('ascii') self._json_response({'audio': combined_b64}) except Exception as e: print(f'[TTS Test] Error: {e}') self._json_response({'error': str(e)}, 500) def _translate(self): """Translate text from one language to another using the LLM.""" try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) text = body.get('text', '').strip() source = (body.get('source_lang') or '').strip() target = (body.get('target_lang') or '').strip() if not text or not source or not target: return self._json_response( {'error': 'text, source_lang, target_lang required'}, 400) if source == target: return self._json_response({'translation': text}) lang_names = dict(LANG_NAMES) lang_names['zh'] = 'Chinese (Simplified)' # Translation needs qualified name src_name = lang_names.get(source, source) tgt_name = lang_names.get(target, target) prompt = ( f"You are a professional translator. Translate the following " f"{src_name} text into {tgt_name}.\n\n" f"Rules:\n" f"- Preserve the original formatting, line breaks, bullet points, " f"and section headings exactly.\n" f"- Keep placeholders, variable names, code, URLs, and identifiers " f"unchanged (e.g. {{name}}, $var, REFERENCE DATA, TRANSFER CAPABILITY).\n" f"- Translate only natural-language content.\n" f"- Output ONLY the translated text. No preamble, no explanations, " f"no surrounding quotes.\n\n" f"Source text:\n{text}" ) llm = OpenAI(api_key=API_KEY, base_url=LLM_BASE) response = llm.chat.completions.create( model=LLM_MODEL, messages=[{'role': 'user', 'content': prompt}], temperature=0.2, extra_body={'enable_thinking': False}, ) translation = (response.choices[0].message.content or '').strip() # Strip wrapping quotes if the model added them if len(translation) >= 2 and ( (translation[0] == '"' and translation[-1] == '"') or (translation[0] == "'" and translation[-1] == "'") ): translation = translation[1:-1].strip() print(f'[Translate] {source} -> {target} ({len(text)} chars)') self._json_response({'translation': translation}) except Exception as e: print(f'[Translate] Error: {e}') self._json_response({'error': str(e)}, 500) def _generate_tool(self): """Use LLM to generate a structured tool definition from natural language.""" raw = '' try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) description = body.get('description', '').strip() model = body.get('model', '').strip() or LLM_MODEL if not description: return self._json_response( {'error': 'description is required'}, 400) prompt = ( "You are a tool definition generator for an OpenAI-compatible function calling system.\n\n" "The user will describe an API tool in natural language. Your job is to extract structured " "information and return ONLY a JSON object (no markdown, no explanation, no code fences) " "with exactly these fields:\n\n" "{\n" ' "name": "snake_case_function_name",\n' ' "description": "Clear description of what this tool does",\n' ' "parameters": [\n' ' {\n' ' "name": "param_name",\n' ' "type": "string",\n' ' "description": "What this parameter is for",\n' ' "required": true\n' ' }\n' ' ],\n' ' "endpoint": "https://api.example.com/path",\n' ' "headers": {}\n' "}\n\n" "Rules:\n" "- name MUST be snake_case (lowercase, underscores only, no spaces)\n" "- type must be one of: string, number, integer, boolean\n" "- Extract the API endpoint URL if mentioned; leave as empty string if not mentioned\n" "- Extract any custom headers if mentioned (e.g. Authorization); default to empty object\n" "- Infer reasonable parameter names, types, and descriptions from context\n" "- Mark parameters as required=true unless they are clearly optional\n" "- Output ONLY the JSON object, nothing else\n\n" f"User description:\n{description}" ) llm = OpenAI(api_key=API_KEY, base_url=LLM_BASE) response = llm.chat.completions.create( model=model, messages=[{'role': 'user', 'content': prompt}], temperature=0.2, extra_body={'enable_thinking': False}, ) raw = (response.choices[0].message.content or '').strip() # Strip markdown code fences if present if raw.startswith('```'): raw = re.sub(r'^```(?:json)?\s*\n?', '', raw) raw = re.sub(r'\n?```\s*$', '', raw) raw = raw.strip() result = json.loads(raw) print(f'[ToolGen] Generated: {result.get("name", "?")} ' f'({len(result.get("parameters", []))} params, model={model})') self._json_response(result) except json.JSONDecodeError as e: print(f'[ToolGen] JSON parse error: {e}\nRaw: {raw[:500]}') self._json_response( {'error': f'LLM returned invalid JSON: {str(e)}'}, 502) except Exception as e: print(f'[ToolGen] Error: {e}') self._json_response({'error': str(e)}, 500) def _generate_handler(self): """Use LLM to generate a Python handler function for a tool.""" raw = '' try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) tool_name = body.get('tool_name', '').strip() description = body.get('description', '').strip() parameters = body.get('parameters', []) handler_description = body.get('handler_description', '').strip() model = body.get('model', '').strip() or LLM_MODEL if not tool_name or not handler_description: return self._json_response( {'error': 'tool_name and handler_description are required'}, 400) params_text = '\n'.join( f" - {p['name']} ({p.get('type', 'string')}): {p.get('description', '')}" for p in parameters ) or ' (none)' prompt = ( "You are a Python code generator for a phone system's tool handler.\n\n" "Generate a handler function that will be registered with @register_handler.\n\n" "SIGNATURE (must be exactly this):\n" f"@register_handler('{tool_name}')\n" f"def handle_{tool_name}(args, tool_def=None):\n\n" "AVAILABLE (already imported, do NOT add import statements):\n" " - requests (for HTTP calls)\n" " - json (for JSON handling)\n" " - re (for regex)\n\n" "HANDLER RULES:\n" " - args is a dict containing the tool parameters\n" " - Must return a string (the result sent back to the AI)\n" " - Use args.get('param_name', '') to safely get parameters\n" " - For API calls, use requests.post/get with timeout=15, verify=False\n" " - Handle errors with try/except and return error messages as strings\n" " - Do NOT use: import, os, subprocess, exec, eval, open, __import__, sys\n" " - Do NOT include any import statements\n\n" f"TOOL DESCRIPTION: {description}\n\n" f"PARAMETERS:\n{params_text}\n\n" f"HANDLER LOGIC: {handler_description}\n\n" "Output ONLY the Python function code (with the @register_handler decorator line). " "No markdown code fences, no explanation, no imports." ) llm = OpenAI(api_key=API_KEY, base_url=LLM_BASE) response = llm.chat.completions.create( model=model, messages=[{'role': 'user', 'content': prompt}], temperature=0.2, extra_body={'enable_thinking': False}, ) raw = (response.choices[0].message.content or '').strip() # Strip markdown code fences if present if raw.startswith('```'): raw = re.sub(r'^```(?:python)?\s*\n?', '', raw) raw = re.sub(r'\n?```\s*$', '', raw) raw = raw.strip() print(f'[HandlerGen] Generated handler for: {tool_name} (model={model})') self._json_response({'tool_name': tool_name, 'handler_code': raw}) except Exception as e: print(f'[HandlerGen] Error: {e}') self._json_response({'error': str(e)}, 500) def _generate_prompt(self): """Use LLM to generate a system prompt for a transfer agent.""" try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) description = body.get('description', '').strip() tools = body.get('tools', []) agent_name = body.get('agent_name', '').strip() or 'Transfer Agent' model = body.get('model', '').strip() or LLM_MODEL if not description and not tools: return self._json_response( {'error': 'Provide a description, tools, or both'}, 400) # Build context about tools tools_context = '' if tools: tool_lines = [] for t in tools: fn = t.get('function', {}) fn_name = fn.get('name', '') fn_desc = fn.get('description', '') if fn_name and fn_name != 'transfer_back': params = fn.get('parameters', {}).get('properties', {}) param_names = ', '.join(params.keys()) if params else 'none' tool_lines.append( f'- {fn_name}({param_names}): {fn_desc}') if tool_lines: tools_context = ( '\n\nTOOLS AVAILABLE TO THIS AGENT:\n' + '\n'.join(tool_lines) ) desc_context = '' if description: desc_context = f'\n\nAGENT PURPOSE (from user):\n{description}' prompt = ( "You are a system prompt writer for a phone call AI agent.\n\n" "Generate a concise system prompt for a transfer agent named " f'"{agent_name}". This agent handles live phone calls after ' "being transferred from a main receptionist.\n" f"{desc_context}" f"{tools_context}\n\n" "RULES FOR THE GENERATED PROMPT:\n" "- Write in second person (\"You are...\")\n" "- Keep it concise but comprehensive (a few paragraphs max)\n" "- Mention this is a live phone call — responses must be brief " "(2-3 sentences per turn)\n" "- If tools are provided, explain WHEN to use each one " "(not the parameter details — the agent already has tool definitions)\n" "- Include a transfer_back instruction: tell the agent to transfer " "back to the main receptionist if the caller's request is outside " "this agent's scope\n" "- Do NOT include greetings or opening lines — those are handled " "separately\n" "- Output ONLY the system prompt text, no markdown formatting, " "no code fences, no explanation\n" ) llm = OpenAI(api_key=API_KEY, base_url=LLM_BASE) response = llm.chat.completions.create( model=model, messages=[{'role': 'user', 'content': prompt}], temperature=0.4, extra_body={'enable_thinking': False}, ) raw = (response.choices[0].message.content or '').strip() # Strip markdown code fences if present if raw.startswith('```'): raw = re.sub(r'^```(?:\w+)?\s*\n?', '', raw) raw = re.sub(r'\n?```\s*$', '', raw) raw = raw.strip() print(f'[PromptGen] Generated prompt for: {agent_name} ' f'({len(raw)} chars, model={model})') self._json_response({'prompt': raw}) except Exception as e: print(f'[PromptGen] Error: {e}') self._json_response({'error': str(e)}, 500) def _save_handler(self): """Validate, save, and activate a generated handler.""" try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) tool_name = body.get('tool_name', '').strip() import textwrap handler_code = textwrap.dedent(body.get('handler_code', '')).strip() if not tool_name or not handler_code: return self._json_response( {'error': 'tool_name and handler_code are required'}, 400) # Safety validation — reject dangerous patterns forbidden = [ 'import ', '__import__', 'exec(', 'eval(', 'compile(', 'open(', 'os.', 'sys.', 'subprocess', 'shutil', 'globals(', 'locals(', 'setattr(', 'getattr(', 'delattr(', '__builtins__', ] for pattern in forbidden: if pattern in handler_code: return self._json_response( {'error': f'Forbidden pattern detected: {pattern}'}, 400) # Syntax check try: compile(handler_code, '', 'exec') except SyntaxError as e: return self._json_response( {'error': f'Syntax error: {e}'}, 400) # Verify decorator if f"@register_handler('{tool_name}')" not in handler_code: return self._json_response( {'error': f"Handler must use @register_handler('{tool_name}')"}, 400) # Save to file from datetime import datetime gen_file = os.path.join(os.path.dirname(__file__), 'generated_handlers.py') timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') _replace_or_append_handler(gen_file, tool_name, handler_code, timestamp) # Hot-load into CUSTOM_HANDLERS from tool_handlers import register_handler, CUSTOM_HANDLERS safe_builtins = { 'True': True, 'False': False, 'None': None, 'int': int, 'float': float, 'str': str, 'bool': bool, 'list': list, 'dict': dict, 'tuple': tuple, 'set': set, 'len': len, 'range': range, 'enumerate': enumerate, 'isinstance': isinstance, 'print': print, 'min': min, 'max': max, 'abs': abs, 'round': round, 'sorted': sorted, 'reversed': reversed, 'zip': zip, 'map': map, 'filter': filter, 'ValueError': ValueError, 'TypeError': TypeError, 'KeyError': KeyError, 'Exception': Exception, } exec_ns = { '__builtins__': safe_builtins, 'register_handler': register_handler, 'CUSTOM_HANDLERS': CUSTOM_HANDLERS, 'requests': requests, 'json': json, 're': re, } exec(handler_code, exec_ns) print(f'[HandlerGen] Saved and activated handler: {tool_name}') self._json_response({'ok': True, 'tool_name': tool_name}) except Exception as e: print(f'[HandlerGen] Save error: {e}') self._json_response({'error': str(e)}, 500) def _check_model(self): """Test if a model exists and can generate responses.""" try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) model = body.get('model', '').strip() if not model: return self._json_response({'error': 'Model name is required'}, 400) llm = OpenAI(api_key=API_KEY, base_url=LLM_BASE) response = llm.chat.completions.create( model=model, messages=[{'role': 'user', 'content': 'Reply with only the word "ok".'}], temperature=0, max_tokens=5, extra_body={'enable_thinking': False}, ) reply = (response.choices[0].message.content or '').strip().lower() print(f'[ModelCheck] {model}: ok (reply={reply})') self._json_response({'ok': True, 'message': f'{model} is available'}) except Exception as e: print(f'[ModelCheck] {model}: failed ({e})') self._json_response({'ok': False, 'error': f'Model not available: {e}'}) def _replace_or_append_handler(filepath, tool_name, code, timestamp): """Replace existing handler for tool_name in generated_handlers.py, or append.""" marker = f'# --- Generated: {tool_name} ' new_block = f'\n\n{marker}({timestamp}) ---\n{code}\n' if os.path.isfile(filepath): with open(filepath, 'r', encoding='utf-8') as f: content = f.read() if marker in content: # Remove old block (from marker to next marker or EOF) pattern = re.escape(marker) + r'\(.*?\) ---\n.*?(?=\n# --- Generated:|\Z)' content = re.sub(pattern, '', content, flags=re.DOTALL) content = content.rstrip() + new_block else: content = content.rstrip() + new_block with open(filepath, 'w', encoding='utf-8') as f: f.write(content) else: header = ( '"""\n' 'Auto-generated tool handlers.\n' 'Managed by AI Tool Builder. Each handler below was generated by the LLM\n' 'and reviewed by the user before activation.\n' '"""\n\n' 'import json\n' 'import re\n' 'import requests\n' 'from tool_handlers import register_handler\n\n' '# Generated handlers below\n' '# ' + '=' * 60 + '\n' ) with open(filepath, 'w', encoding='utf-8') as f: f.write(header + new_block)