| 1 | """OpenAI-compatible HTTP API server for a fixed nanobot session. |
| 2 | |
| 3 | Provides /v1/chat/completions and /v1/models endpoints. |
| 4 | All requests route to a single persistent API session. |
| 5 | """ |
| 6 | |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | import asyncio |
| 10 | import json as _json |
| 11 | import time |
| 12 | import uuid |
| 13 | from typing import Any |
| 14 | |
| 15 | from aiohttp import web |
| 16 | from loguru import logger |
| 17 | |
| 18 | from nanobot.config.paths import get_media_dir |
| 19 | from nanobot.utils.helpers import safe_filename |
| 20 | from nanobot.utils.media_decode import MAX_FILE_SIZE |
| 21 | from nanobot.utils.media_decode import ( |
| 22 | FileSizeExceeded as _FileSizeExceeded, |
| 23 | ) |
| 24 | from nanobot.utils.media_decode import ( |
| 25 | save_base64_data_url as _save_base64_data_url, |
| 26 | ) |
| 27 | from nanobot.utils.runtime import EMPTY_FINAL_RESPONSE_MESSAGE |
| 28 | |
| 29 | __all__ = ( |
| 30 | "MAX_FILE_SIZE", |
| 31 | "_FileSizeExceeded", |
| 32 | "_save_base64_data_url", |
| 33 | "create_app", |
| 34 | "handle_chat_completions", |
| 35 | ) |
| 36 | |
| 37 | |
| 38 | API_SESSION_KEY = "api:default" |
| 39 | API_CHAT_ID = "default" |
| 40 | |
| 41 | |
| 42 | # --------------------------------------------------------------------------- |
| 43 | # Response helpers |
| 44 | # --------------------------------------------------------------------------- |
| 45 | |
| 46 | |
| 47 | def _error_json(status: int, message: str, err_type: str = "invalid_request_error") -> web.Response: |
| 48 | return web.json_response( |
| 49 | {"error": {"message": message, "type": err_type, "code": status}}, |
| 50 | status=status, |
| 51 | ) |
| 52 | |
| 53 | |
| 54 | def _chat_completion_response(content: str, model: str) -> dict[str, Any]: |
| 55 | return { |
| 56 | "id": f"chatcmpl-{uuid.uuid4().hex[:12]}", |
| 57 | "object": "chat.completion", |
| 58 | "created": int(time.time()), |
| 59 | "model": model, |
| 60 | "choices": [ |
| 61 | { |
| 62 | "index": 0, |
| 63 | "message": {"role": "assistant", "content": content}, |
| 64 | "finish_reason": "stop", |
| 65 | } |
| 66 | ], |
| 67 | "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}, |
| 68 | } |
| 69 | |
| 70 | |
| 71 | def _response_text(value: Any) -> str: |
| 72 | """Normalize process_direct output to plain assistant text.""" |
| 73 | if value is None: |
| 74 | return "" |
| 75 | if hasattr(value, "content"): |
| 76 | return str(getattr(value, "content") or "") |
| 77 | return str(value) |
| 78 | |
| 79 | # --------------------------------------------------------------------------- |
| 80 | # SSE helpers |
| 81 | # --------------------------------------------------------------------------- |
| 82 | |
| 83 | |
| 84 | def _sse_chunk(delta: str, model: str, chunk_id: str, finish_reason: str | None = None) -> bytes: |
| 85 | """Format a single OpenAI-compatible SSE chunk.""" |
| 86 | payload = { |
| 87 | "id": chunk_id, |
| 88 | "object": "chat.completion.chunk", |
| 89 | "created": int(time.time()), |
| 90 | "model": model, |
| 91 | "choices": [ |
| 92 | { |
| 93 | "index": 0, |
| 94 | "delta": {"content": delta} if delta else {}, |
| 95 | "finish_reason": finish_reason, |
| 96 | } |
| 97 | ], |
| 98 | } |
| 99 | return f"data: {_json.dumps(payload)}\n\n".encode() |
| 100 | |
| 101 | |
| 102 | _SSE_DONE = b"data: [DONE]\n\n" |
| 103 | |
| 104 | # --------------------------------------------------------------------------- |
| 105 | # Upload helpers |
| 106 | # --------------------------------------------------------------------------- |
| 107 | |
| 108 | |
| 109 | def _parse_json_content(body: dict) -> tuple[str, list[str]]: |
| 110 | """Parse JSON request body. Returns (text, media_paths).""" |
| 111 | messages = body.get("messages") |
| 112 | if not isinstance(messages, list) or len(messages) != 1: |
| 113 | raise ValueError("Only a single user message is supported") |
| 114 | message = messages[0] |
| 115 | if not isinstance(message, dict) or message.get("role") != "user": |
| 116 | raise ValueError("Only a single user message is supported") |
| 117 | |
| 118 | user_content = message.get("content", "") |
| 119 | media_dir = get_media_dir("api") |
| 120 | media_paths: list[str] = [] |
| 121 | |
| 122 | if isinstance(user_content, list): |
| 123 | text_parts: list[str] = [] |
| 124 | for part in user_content: |
| 125 | if not isinstance(part, dict): |
| 126 | continue |
| 127 | if part.get("type") == "text": |
| 128 | text_parts.append(part.get("text", "")) |
| 129 | elif part.get("type") == "image_url": |
| 130 | url = part.get("image_url", {}).get("url", "") |
| 131 | if url.startswith("data:"): |
| 132 | saved = _save_base64_data_url(url, media_dir) |
| 133 | if saved: |
| 134 | media_paths.append(saved) |
| 135 | elif url: |
| 136 | raise ValueError( |
| 137 | "Remote image URLs are not supported. " |
| 138 | "Use base64 data URLs or upload files via multipart/form-data." |
| 139 | ) |
| 140 | text = " ".join(text_parts) |
| 141 | elif isinstance(user_content, str): |
| 142 | text = user_content |
| 143 | else: |
| 144 | raise ValueError("Invalid content format") |
| 145 | |
| 146 | return text, media_paths |
| 147 | |
| 148 | |
| 149 | async def _parse_multipart(request: web.Request) -> tuple[str, list[str], str | None, str | None]: |
| 150 | """Parse multipart/form-data. Returns (text, media_paths, session_id, model).""" |
| 151 | media_dir = get_media_dir("api") |
| 152 | reader = await request.multipart() |
| 153 | text = "" |
| 154 | session_id = None |
| 155 | model = None |
| 156 | media_paths: list[str] = [] |
| 157 | |
| 158 | while True: |
| 159 | part = await reader.next() |
| 160 | if part is None: |
| 161 | break |
| 162 | if part.name == "message": |
| 163 | text = (await part.read()).decode("utf-8") |
| 164 | elif part.name == "session_id": |
| 165 | session_id = (await part.read()).decode("utf-8").strip() |
| 166 | elif part.name == "model": |
| 167 | model = (await part.read()).decode("utf-8").strip() |
| 168 | elif part.name == "files": |
| 169 | raw = await part.read() |
| 170 | if len(raw) > MAX_FILE_SIZE: |
| 171 | raise _FileSizeExceeded( |
| 172 | f"File '{part.filename}' exceeds {MAX_FILE_SIZE // (1024 * 1024)}MB limit" |
| 173 | ) |
| 174 | base = safe_filename(part.filename or "upload.bin") |
| 175 | filename = f"{uuid.uuid4().hex[:12]}_{base}" |
| 176 | dest = media_dir / filename |
| 177 | dest.write_bytes(raw) |
| 178 | media_paths.append(str(dest)) |
| 179 | |
| 180 | if not text: |
| 181 | text = "请分析上传的文件" |
| 182 | |
| 183 | return text, media_paths, session_id, model |
| 184 | |
| 185 | |
| 186 | # --------------------------------------------------------------------------- |
| 187 | # Route handlers |
| 188 | # --------------------------------------------------------------------------- |
| 189 | |
| 190 | |
| 191 | async def handle_chat_completions(request: web.Request) -> web.Response: |
| 192 | """POST /v1/chat/completions — supports JSON and multipart/form-data.""" |
| 193 | content_type = request.content_type or "" |
| 194 | if not isinstance(content_type, str): |
| 195 | content_type = "" |
| 196 | |
| 197 | agent_loop = request.app["agent_loop"] |
| 198 | timeout_s: float = request.app.get("request_timeout", 120.0) |
| 199 | model_name: str = request.app.get("model_name", "nanobot") |
| 200 | |
| 201 | stream = False |
| 202 | try: |
| 203 | if content_type.startswith("multipart/"): |
| 204 | text, media_paths, session_id, requested_model = await _parse_multipart(request) |
| 205 | else: |
| 206 | try: |
| 207 | body = await request.json() |
| 208 | except Exception: |
| 209 | return _error_json(400, "Invalid JSON body") |
| 210 | stream = body.get("stream", False) |
| 211 | requested_model = body.get("model") |
| 212 | text, media_paths = _parse_json_content(body) |
| 213 | session_id = body.get("session_id") |
| 214 | except ValueError as e: |
| 215 | return _error_json(400, str(e)) |
| 216 | except _FileSizeExceeded as e: |
| 217 | return _error_json(413, str(e), err_type="invalid_request_error") |
| 218 | except Exception: |
| 219 | logger.exception("Error parsing upload") |
| 220 | return _error_json(413, "File too large or invalid upload") |
| 221 | |
| 222 | if requested_model and requested_model != model_name: |
| 223 | return _error_json(400, f"Only configured model '{model_name}' is available") |
| 224 | |
| 225 | session_key = f"api:{session_id}" if session_id else API_SESSION_KEY |
| 226 | session_locks: dict[str, asyncio.Lock] = request.app["session_locks"] |
| 227 | session_lock = session_locks.setdefault(session_key, asyncio.Lock()) |
| 228 | |
| 229 | logger.info( |
| 230 | "API request session_key={} media={} text={} stream={}", |
| 231 | session_key, len(media_paths), text[:80], stream, |
| 232 | ) |
| 233 | # -- streaming path -- |
| 234 | if stream: |
| 235 | resp = web.StreamResponse() |
| 236 | resp.content_type = "text/event-stream" |
| 237 | resp.headers["Cache-Control"] = "no-cache" |
| 238 | resp.headers["Connection"] = "keep-alive" |
| 239 | resp.enable_compression() |
| 240 | await resp.prepare(request) |
| 241 | |
| 242 | chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" |
| 243 | queue: asyncio.Queue[str | None] = asyncio.Queue() |
| 244 | stream_failed = False |
| 245 | |
| 246 | async def _on_stream(token: str) -> None: |
| 247 | await queue.put(token) |
| 248 | |
| 249 | async def _on_stream_end(*_a: Any, **_kw: Any) -> None: |
| 250 | await queue.put(None) |
| 251 | |
| 252 | async def _run() -> None: |
| 253 | nonlocal stream_failed |
| 254 | try: |
| 255 | async with session_lock: |
| 256 | await asyncio.wait_for( |
| 257 | agent_loop.process_direct( |
| 258 | content=text, |
| 259 | media=media_paths if media_paths else None, |
| 260 | session_key=session_key, |
| 261 | channel="api", |
| 262 | chat_id=API_CHAT_ID, |
| 263 | on_stream=_on_stream, |
| 264 | on_stream_end=_on_stream_end, |
| 265 | ), |
| 266 | timeout=timeout_s, |
| 267 | ) |
| 268 | except Exception: |
| 269 | stream_failed = True |
| 270 | logger.exception("Streaming error for session {}", session_key) |
| 271 | await queue.put(None) |
| 272 | |
| 273 | task = asyncio.create_task(_run()) |
| 274 | try: |
| 275 | while True: |
| 276 | token = await queue.get() |
| 277 | if token is None: |
| 278 | break |
| 279 | await resp.write(_sse_chunk(token, model_name, chunk_id)) |
| 280 | finally: |
| 281 | task.cancel() |
| 282 | |
| 283 | if not stream_failed: |
| 284 | await resp.write(_sse_chunk("", model_name, chunk_id, finish_reason="stop")) |
| 285 | await resp.write(_SSE_DONE) |
| 286 | return resp |
| 287 | |
| 288 | # -- non-streaming path (original logic) -- |
| 289 | fallback_message = EMPTY_FINAL_RESPONSE_MESSAGE |
| 290 | |
| 291 | try: |
| 292 | async with session_lock: |
| 293 | try: |
| 294 | response = await asyncio.wait_for( |
| 295 | agent_loop.process_direct( |
| 296 | content=text, |
| 297 | media=media_paths if media_paths else None, |
| 298 | session_key=session_key, |
| 299 | channel="api", |
| 300 | chat_id=API_CHAT_ID, |
| 301 | ), |
| 302 | timeout=timeout_s, |
| 303 | ) |
| 304 | response_text = _response_text(response) |
| 305 | |
| 306 | if not response_text or not response_text.strip(): |
| 307 | logger.warning("Empty response for session {}, retrying", session_key) |
| 308 | retry_response = await asyncio.wait_for( |
| 309 | agent_loop.process_direct( |
| 310 | content=text, |
| 311 | media=media_paths if media_paths else None, |
| 312 | session_key=session_key, |
| 313 | channel="api", |
| 314 | chat_id=API_CHAT_ID, |
| 315 | ), |
| 316 | timeout=timeout_s, |
| 317 | ) |
| 318 | response_text = _response_text(retry_response) |
| 319 | if not response_text or not response_text.strip(): |
| 320 | logger.warning("Empty response after retry, using fallback") |
| 321 | response_text = fallback_message |
| 322 | |
| 323 | except asyncio.TimeoutError: |
| 324 | return _error_json(504, f"Request timed out after {timeout_s}s") |
| 325 | except Exception: |
| 326 | logger.exception("Error processing request for session {}", session_key) |
| 327 | return _error_json(500, "Internal server error", err_type="server_error") |
| 328 | except Exception: |
| 329 | logger.exception("Unexpected API lock error for session {}", session_key) |
| 330 | return _error_json(500, "Internal server error", err_type="server_error") |
| 331 | |
| 332 | return web.json_response(_chat_completion_response(response_text, model_name)) |
| 333 | |
| 334 | |
| 335 | async def handle_models(request: web.Request) -> web.Response: |
| 336 | """GET /v1/models""" |
| 337 | model_name = request.app.get("model_name", "nanobot") |
| 338 | return web.json_response( |
| 339 | { |
| 340 | "object": "list", |
| 341 | "data": [ |
| 342 | { |
| 343 | "id": model_name, |
| 344 | "object": "model", |
| 345 | "created": 0, |
| 346 | "owned_by": "nanobot", |
| 347 | } |
| 348 | ], |
| 349 | } |
| 350 | ) |
| 351 | |
| 352 | |
| 353 | async def handle_health(request: web.Request) -> web.Response: |
| 354 | """GET /health""" |
| 355 | return web.json_response({"status": "ok"}) |
| 356 | |
| 357 | |
| 358 | # --------------------------------------------------------------------------- |
| 359 | # App factory |
| 360 | # --------------------------------------------------------------------------- |
| 361 | |
| 362 | |
| 363 | def create_app( |
| 364 | agent_loop, model_name: str = "nanobot", request_timeout: float = 120.0 |
| 365 | ) -> web.Application: |
| 366 | """Create the aiohttp application. |
| 367 | |
| 368 | Args: |
| 369 | agent_loop: An initialized AgentLoop instance. |
| 370 | model_name: Model name reported in responses. |
| 371 | request_timeout: Per-request timeout in seconds. |
| 372 | """ |
| 373 | app = web.Application(client_max_size=20 * 1024 * 1024) # 20MB for base64 images |
| 374 | app["agent_loop"] = agent_loop |
| 375 | app["model_name"] = model_name |
| 376 | app["request_timeout"] = request_timeout |
| 377 | app["session_locks"] = {} # per-user locks, keyed by session_key |
| 378 | |
| 379 | app.router.add_post("/v1/chat/completions", handle_chat_completions) |
| 380 | app.router.add_get("/v1/models", handle_models) |
| 381 | app.router.add_get("/health", handle_health) |
| 382 | return app |
| 383 |