"""APIClient - Agent - Conversational AI co-pilot core.""" import json import re import httpx from app.core import storage MAX_HISTORY_MSGS = 30 _SYSTEM_PROMPT = """\ You are APIClient - Agent, an expert AI API testing co-pilot embedded in the APIClient - Agent desktop application. Your responsibilities: • Help craft and debug HTTP requests (REST, JSON-API, GraphQL, Odoo APIs) • Analyze HTTP responses - status codes, headers, body structure, errors • Specialize in the EKIKA Odoo API Framework: - JSON-API (Content-Type: application/vnd.api+json) - body format: {"data": {"type": model, "attributes": {...}}} - REST JSON (Content-Type: application/json) - GraphQL (POST with {"query": "..."} body) - Auth: x-api-key header, Basic Auth, OAuth2 Bearer, JWT Bearer - Odoo models: sale.order, res.partner, account.move, product.template, stock.picking, etc. • Generate request bodies, params, headers, and test scripts • Explain SSL/TLS errors, auth failures, and connection issues • Help with environment variable setup ({{base_url}}, {{api_key}}, etc.) When you produce content the user should apply to their request, use EXACTLY these fences: ```apply:body { "json": "here" } ``` ```apply:params page=1 limit=10 fields=id,name ``` ```apply:headers x-api-key: {{api_key}} Accept: application/vnd.api+json ``` ```apply:test pm.test('Status 200', lambda: pm.response.to_have_status(200)) pm.test('Has data', lambda: expect(pm.response.json()).to_have_key('data')) ``` Rules: - Be concise and actionable - explain WHY, not just WHAT - If you add apply blocks, briefly explain what each block does - For JSON-API responses: data is in response.data, errors in response.errors - For SSL cert errors: tell user to uncheck SSL verification in the Settings tab - For 401/403: check x-api-key header and environment variable values - For unresolved {{variable}}: tell user to set up environment via Tools → Environments """ class AIError(Exception): pass def get_api_key() -> str: return storage.get_setting("anthropic_api_key", "") def build_context(req=None, resp=None, env_vars: dict = None) -> str: """Build compact context string for system prompt injection.""" parts = [] if req: parts.append(f"METHOD: {req.method}") parts.append(f"URL: {req.url}") if req.headers: safe = { k: ("***" if any(s in k.lower() for s in ["key", "token", "secret", "auth", "pass"]) else v) for k, v in req.headers.items() } parts.append(f"HEADERS: {json.dumps(safe)}") if req.params: parts.append(f"PARAMS: {json.dumps(req.params)}") if req.body: preview = req.body[:1500] + ("…" if len(req.body) > 1500 else "") parts.append(f"BODY ({req.body_type}):\n{preview}") if req.content_type: parts.append(f"CONTENT-TYPE: {req.content_type}") if req.test_script: parts.append(f"TEST SCRIPT:\n{req.test_script}") if resp: if resp.error: parts.append(f"\nRESPONSE ERROR: {resp.error}") else: parts.append(f"\nRESPONSE: {resp.status} {resp.reason} ({resp.elapsed_ms:.0f} ms)") ct = (resp.headers or {}).get("content-type", (resp.headers or {}).get("Content-Type", "")) if ct: parts.append(f"RESPONSE CONTENT-TYPE: {ct}") if resp.body: preview = resp.body[:4000] + ("…" if len(resp.body) > 4000 else "") parts.append(f"RESPONSE BODY:\n{preview}") if env_vars: safe_vars = { k: ("***" if any(s in k.lower() for s in ["key", "token", "secret", "password", "pass"]) else v) for k, v in env_vars.items() } parts.append(f"\nENVIRONMENT VARIABLES: {json.dumps(safe_vars)}") return "\n".join(parts) def stream_chat(messages: list[dict], context: str = "", chunk_cb=None) -> str: """ Stream a multi-turn conversation to Claude. messages: list of {"role": "user"|"assistant", "content": str} chunk_cb(chunk: str): called for each streamed text chunk Returns full assistant response text. Raises AIError on failure. """ api_key = get_api_key() if not api_key: raise AIError( "No Anthropic API key configured.\n" "Go to Tools → AI Assistant → Settings to add your key." ) system = _SYSTEM_PROMPT if context: system += f"\n\n## Current Request Context\n{context}" headers = { "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", } payload = { "model": "claude-opus-4-6", "max_tokens": 2048, "system": system, "messages": messages[-MAX_HISTORY_MSGS:], } full_text = "" try: with httpx.stream( "POST", "https://api.anthropic.com/v1/messages", headers=headers, json=payload, timeout=60.0, ) as resp: if resp.status_code != 200: body = resp.read().decode() raise AIError(f"Claude API error {resp.status_code}: {body[:400]}") for line in resp.iter_lines(): if not line.startswith("data:"): continue data_str = line[5:].strip() if data_str == "[DONE]": break try: event = json.loads(data_str) delta = event.get("delta", {}) if delta.get("type") == "text_delta": chunk = delta.get("text", "") full_text += chunk if chunk_cb: chunk_cb(chunk) except json.JSONDecodeError: continue except httpx.TimeoutException: raise AIError("Request timed out. Try a shorter question or check your connection.") except httpx.RequestError as e: raise AIError(f"Network error: {e}") return full_text def parse_apply_blocks(text: str) -> list[dict]: """Parse ```apply:TYPE ... ``` blocks from AI response text.""" blocks = [] for m in re.finditer(r"```apply:(\w+)\n(.*?)```", text, re.DOTALL): blocks.append({"type": m.group(1), "content": m.group(2).strip()}) return blocks def strip_apply_blocks(text: str) -> str: """Remove apply fences from display text, leaving the explanation.""" return re.sub(r"```apply:\w+\n.*?```", "", text, flags=re.DOTALL).strip()