"""OpenAI-compatible HTTP surface for a mesh node. Only the subset ThoxRoute and the ThoxOS apps actually call is implemented: ``/v1/chat/completions``, ``/v1/models`` and health. Implementing the surface narrowly and honestly is deliberate — a node that advertises streaming or embeddings it cannot serve would be selected by the router and then fail, which is worse for the mesh than not advertising them. Every request is timed and recorded into the ``TelemetryRecorder``, because that recording *is* how the node earns or loses its place in the router's ranking. """ from __future__ import annotations import logging import time import uuid from typing import Any, Literal from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from pydantic import BaseModel, Field, field_validator from .backends import Backend from .errors import BackendError from .telemetry import TelemetryRecorder logger = logging.getLogger("thox.server") class ChatMessage(BaseModel): """One message in an OpenAI-style chat request.""" role: Literal["system", "user", "assistant", "tool"] content: str = "" @field_validator("content", mode="before") @classmethod def _coerce_content(cls, value: Any) -> str: """Accept the content-parts array form and flatten it to text. Newer OpenAI clients send ``[{"type": "text", "text": "..."}]``. A node that 422s on that shape looks broken to a caller that is, by the spec, well-behaved. """ if value is None: return "" if isinstance(value, str): return value if isinstance(value, list): parts = [ str(part.get("text", "")) for part in value if isinstance(part, dict) and part.get("type") in (None, "text") ] return "".join(parts) return str(value) class ChatCompletionRequest(BaseModel): """Request body for ``/v1/chat/completions``.""" model: str | None = None messages: list[ChatMessage] = Field(min_length=1) max_tokens: int | None = Field(default=None, ge=1) temperature: float = Field(default=0.7, ge=0.0, le=2.0) stop: list[str] | str | None = None stream: bool = False def create_app( *, backend: Backend, recorder: TelemetryRecorder, model_id: str, context_window: int, max_tokens_cap: int, node_state: dict[str, Any] | None = None, ) -> FastAPI: """Build the ASGI app. ``node_state`` is a live dict owned by the agent; the health endpoint reads it so operators (and the controller's health loop) can see registration state without a second channel. """ app = FastAPI(title="THOX Mesh Node", version="1.0.0") state = node_state if node_state is not None else {} @app.get("/health") def health() -> dict[str, Any]: """Liveness plus current mesh registration and telemetry.""" return { "status": "ok", "model_id": model_id, "context_window": context_window, "mesh": { "transport": state.get("transport"), "endpoint_id": state.get("endpoint_id"), "registered": bool(state.get("endpoint_id")), "public_url": state.get("public_url"), "last_heartbeat_at": state.get("last_heartbeat_at"), "last_error": state.get("last_error"), }, "telemetry": recorder.snapshot(), } @app.get("/v1/models") def list_models() -> dict[str, Any]: """Model list in OpenAI's shape, so generic clients can discover us.""" return { "object": "list", "data": [ { "id": model_id, "object": "model", "owned_by": "thox", "created": int(state.get("started_at", time.time())), "context_window": context_window, } ], } @app.post("/v1/chat/completions") def chat_completions(body: ChatCompletionRequest) -> JSONResponse: """Generate one completion and record its outcome as telemetry.""" if body.stream: # Refused explicitly rather than silently returning a non-stream body, # which would leave a streaming client hanging on a parse it can never finish. raise HTTPException( status_code=400, detail="streaming is not supported by this node; retry with stream=false", ) requested = body.max_tokens or max_tokens_cap max_tokens = max(1, min(requested, max_tokens_cap)) stop = [body.stop] if isinstance(body.stop, str) else body.stop recorder.request_started() started = time.perf_counter() try: completion = backend.generate( [m.model_dump() for m in body.messages], max_tokens=max_tokens, temperature=body.temperature, stop=stop, ) except BackendError as exc: recorder.request_finished(latency_ms=(time.perf_counter() - started) * 1000, ok=False) logger.error("generation failed: %s", exc) raise HTTPException(status_code=503, detail=str(exc)) from exc except Exception as exc: # noqa: BLE001 - never leak a 500 without telemetry recorder.request_finished(latency_ms=(time.perf_counter() - started) * 1000, ok=False) logger.exception("unexpected generation error") raise HTTPException(status_code=500, detail="internal generation error") from exc latency_ms = (time.perf_counter() - started) * 1000 recorder.request_finished( latency_ms=latency_ms, ok=True, completion_tokens=completion.completion_tokens, ) return JSONResponse( { "id": f"chatcmpl-{uuid.uuid4().hex[:24]}", "object": "chat.completion", "created": int(time.time()), "model": body.model or model_id, "choices": [ { "index": 0, "message": {"role": "assistant", "content": completion.text}, "finish_reason": "stop", } ], "usage": { "prompt_tokens": completion.prompt_tokens, "completion_tokens": completion.completion_tokens, "total_tokens": completion.total_tokens, }, "thox_node": { "latency_ms": round(latency_ms, 1), "endpoint_id": state.get("endpoint_id"), "node_kind": state.get("node_kind"), }, } ) @app.get("/") def root(request: Request) -> dict[str, Any]: """Human-facing landing payload; also what a Space's health probe hits.""" return { "service": "THOX Mesh Node", "model_id": model_id, "openai_base_url": str(request.base_url).rstrip("/") + "/v1", "endpoints": ["/v1/chat/completions", "/v1/models", "/health"], "registered": bool(state.get("endpoint_id")), } return app