feat: implement raycast backend

src/beaver_gateway/frontends/anthropic.py

@@ -0,0 +1,299 @@
+"""``POST /v1/messages`` frontend.
+
+Exposes the gateway as an Anthropic-compatible Messages endpoint, so any
+client that already speaks Anthropic (Cursor, Cline, the official SDK,
+``curl``) can hit a configured agent by passing its name as ``model``.
+
+The frontend is intentionally thin: it authenticates the bearer token,
+resolves ``body.model`` to an agent + its backend, and then either
+streams the backend's events straight to SSE or accumulates them into a
+single ``Message`` for ``stream=false`` callers. All provider quirks
+already live in the backend adapters; we don't translate here.
+
+Phase 1.4 wires only ``RaycastAgent`` through ``RaycastBackend``;
+``ClaudeAgent`` lands in Phase 2 and will plug into the same dispatch
+table without changes to this module.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import TYPE_CHECKING, Any
+
+from anthropic.types import (
+    InputJSONDelta,
+    Message,
+    RawContentBlockDeltaEvent,
+    RawContentBlockStartEvent,
+    RawContentBlockStopEvent,
+    RawMessageDeltaEvent,
+    RawMessageStartEvent,
+    SignatureDelta,
+    TextBlock,
+    TextDelta,
+    ThinkingBlock,
+    ThinkingDelta,
+    ToolUseBlock,
+    Usage,
+)
+from fastapi import FastAPI, HTTPException, Request, status
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from beaver_gateway.frontends.base import Frontend
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from beaver_gateway.core.events import MessageStreamEvent
+    from beaver_gateway.frontends.base import GatewayRuntime
+
+
+_log = logging.getLogger("beaver_gateway.frontends.anthropic")
+
+
+__all__ = ["AnthropicMessagesFrontend"]
+
+
+class AnthropicMessagesFrontend(Frontend):
+    """FastAPI app behind ``POST /v1/messages`` + ``GET /v1/models``."""
+
+    def __init__(self, *, host: str = "0.0.0.0", port: int = 8000) -> None:  # noqa: S104
+        self.host = host
+        self.port = port
+        self._runtime: GatewayRuntime | None = None
+        self._app: FastAPI | None = None
+
+    def configure(self, runtime: GatewayRuntime) -> None:
+        self._runtime = runtime
+        self._app = self._build_app(runtime)
+
+    async def serve(self) -> None:
+        # Local import: uvicorn pulls in a lot, no reason to load it when
+        # something else (a test, a script) imports this module just for
+        # the FastAPI factory.
+        import uvicorn
+
+        if self._app is None:
+            msg = "configure() must be called before serve()"
+            raise RuntimeError(msg)
+
+        config = uvicorn.Config(
+            self._app, host=self.host, port=self.port, log_level="info"
+        )
+        server = uvicorn.Server(config)
+        await server.serve()
+
+    def _build_app(self, runtime: GatewayRuntime) -> FastAPI:
+        app = FastAPI(title="beaver-gateway / Anthropic Messages")
+
+        @app.get("/healthz")
+        async def healthz() -> dict[str, str]:
+            return {"status": "ok"}
+
+        @app.get("/v1/models")
+        async def list_models(request: Request) -> dict[str, Any]:
+            _require_token(request, runtime)
+            data = [
+                {
+                    "type": "model",
+                    "id": a.name,
+                    "display_name": a.name,
+                    "created_at": None,
+                }
+                for a in runtime.agents
+            ]
+            return {"data": data, "has_more": False, "first_id": None, "last_id": None}
+
+        @app.post("/v1/messages")
+        async def create_message(request: Request) -> Any:
+            token_name = _require_token(request, runtime)
+            try:
+                body = await request.json()
+            except json.JSONDecodeError as exc:
+                raise HTTPException(
+                    status.HTTP_400_BAD_REQUEST, f"invalid JSON: {exc}"
+                ) from exc
+
+            model = body.get("model")
+            if not isinstance(model, str):
+                raise HTTPException(
+                    status.HTTP_400_BAD_REQUEST, "missing or non-string `model`"
+                )
+
+            agent = runtime.agents.get(model)
+            if agent is None:
+                raise HTTPException(
+                    status.HTTP_404_NOT_FOUND, f"unknown agent: {model!r}"
+                )
+
+            backend = runtime.backends.get(agent.name)
+            if backend is None:
+                # Agent exists in config but its backend isn't wired in
+                # this phase (e.g. ClaudeAgent before Phase 2, or a
+                # RaycastAgent without RAYCAST_BEARER set at startup).
+                raise HTTPException(
+                    status.HTTP_503_SERVICE_UNAVAILABLE,
+                    f"no backend configured for agent {agent.name!r}",
+                )
+
+            messages = body.get("messages") or []
+            system = body.get("system")
+            stream_flag = bool(body.get("stream", False))
+
+            _log.info(
+                "messages: actor=%s agent=%s stream=%s msgs=%d",
+                token_name,
+                agent.name,
+                stream_flag,
+                len(messages),
+            )
+
+            # Forward per-request knobs the Anthropic body may carry —
+            # backend adapters layer these over per-agent defaults. Only
+            # values explicitly present (not Anthropic-defaulted ones we
+            # never received) are forwarded, so the agent's default still
+            # wins when the caller omits the field.
+            options: dict[str, Any] = {}
+            if isinstance(body.get("temperature"), int | float):
+                options["temperature"] = body["temperature"]
+
+            events = backend.complete(
+                agent=agent,
+                messages=messages,
+                system=system if isinstance(system, str) else None,
+                **options,
+            )
+
+            if stream_flag:
+                return StreamingResponse(
+                    _sse(events), media_type="text/event-stream"
+                )
+            message = await _accumulate(events, model=model)
+            return JSONResponse(content=message.model_dump(mode="json"))
+
+        return app
+
+
+def _require_token(request: Request, runtime: GatewayRuntime) -> str:
+    """Verify the request's bearer and return the token's audit name.
+
+    Accepts both ``X-Api-Key: <token>`` (what the official Anthropic
+    SDK sends — LibreChat, the CLI, third-party clients) and
+    ``Authorization: Bearer <token>`` (curl, Cursor). Raises 401 on
+    miss. ``TokenStore`` doesn't know about HTTP, so response shape
+    is owned here.
+    """
+    api_key = request.headers.get("x-api-key")
+    name = (
+        runtime.token_store.verify(api_key)
+        if api_key
+        else runtime.token_store.verify_bearer(request.headers.get("authorization"))
+    )
+    if name is None:
+        raise HTTPException(
+            status.HTTP_401_UNAUTHORIZED,
+            "invalid or missing bearer token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    return name
+
+
+async def _sse(events: AsyncIterator[MessageStreamEvent]) -> AsyncIterator[bytes]:
+    r"""Serialize an event stream into Anthropic's ``text/event-stream`` form.
+
+    Each event becomes ``event: <type>\ndata: <json>\n\n`` — the shape
+    the Anthropic SDK's SSE decoder expects. Errors mid-stream are
+    swallowed into a synthetic ``error`` event so the client sees the
+    failure rather than a hung connection.
+    """
+    try:
+        async for ev in events:
+            payload = ev.model_dump_json()
+            yield f"event: {ev.type}\ndata: {payload}\n\n".encode()
+    except Exception as exc:  # noqa: BLE001
+        _log.exception("backend stream failed")
+        err = json.dumps(
+            {"type": "error", "error": {"type": "api_error", "message": str(exc)}}
+        )
+        yield f"event: error\ndata: {err}\n\n".encode()
+
+
+async def _accumulate(
+    events: AsyncIterator[MessageStreamEvent], *, model: str
+) -> Message:
+    """Collapse a stream-event sequence into one ``Message`` (``stream=false``).
+
+    Mirrors the Anthropic SDK's own accumulator: walk events, build
+    block dicts indexed by their ``content_block`` index, fold text /
+    thinking deltas in, buffer ``input_json_delta`` chunks until the
+    block closes (then JSON-parse them once).
+    """
+    message_id = ""
+    role = "assistant"
+    usage = Usage(input_tokens=0, output_tokens=0)
+    blocks: dict[int, dict[str, Any]] = {}
+    json_buffers: dict[int, str] = {}
+    stop_reason: str | None = None
+    stop_sequence: str | None = None
+
+    async for ev in events:
+        # isinstance, not ``ev.type == "..."``: ty narrows on the
+        # discriminator only via the class, and the raw event union
+        # carries its own discriminators (``Raw*Event``) the SDK
+        # already promises.
+        if isinstance(ev, RawMessageStartEvent):
+            message_id = ev.message.id
+            role = ev.message.role
+            usage = ev.message.usage
+        elif isinstance(ev, RawContentBlockStartEvent):
+            blocks[ev.index] = ev.content_block.model_dump()
+            if blocks[ev.index].get("type") == "tool_use":
+                json_buffers[ev.index] = ""
+        elif isinstance(ev, RawContentBlockDeltaEvent):
+            blk = blocks.setdefault(ev.index, {})
+            delta = ev.delta
+            if isinstance(delta, TextDelta):
+                blk["text"] = blk.get("text", "") + delta.text
+            elif isinstance(delta, InputJSONDelta):
+                json_buffers[ev.index] = (
+                    json_buffers.get(ev.index, "") + delta.partial_json
+                )
+            elif isinstance(delta, ThinkingDelta):
+                blk["thinking"] = blk.get("thinking", "") + delta.thinking
+            elif isinstance(delta, SignatureDelta):
+                blk["signature"] = delta.signature
+        elif isinstance(ev, RawContentBlockStopEvent):
+            blk = blocks.get(ev.index, {})
+            if blk.get("type") == "tool_use":
+                raw = json_buffers.pop(ev.index, "")
+                blk["input"] = json.loads(raw) if raw.strip() else {}
+        elif isinstance(ev, RawMessageDeltaEvent):
+            stop_reason = ev.delta.stop_reason
+            stop_sequence = ev.delta.stop_sequence
+            if ev.usage.output_tokens:
+                usage = Usage.model_validate(
+                    {**usage.model_dump(), "output_tokens": ev.usage.output_tokens}
+                )
+
+    content = []
+    for idx in sorted(blocks):
+        bd = blocks[idx]
+        btype = bd.get("type")
+        if btype == "text":
+            content.append(TextBlock.model_validate(bd))
+        elif btype == "tool_use":
+            content.append(ToolUseBlock.model_validate(bd))
+        elif btype == "thinking":
+            content.append(ThinkingBlock.model_validate(bd))
+
+    return Message(
+        id=message_id or "msg_unknown",
+        type="message",
+        role=role,
+        model=model,
+        content=content,
+        stop_reason=stop_reason,  # type: ignore[arg-type]
+        stop_sequence=stop_sequence,
+        usage=usage,
+    )