Coverage for src/lilbee/server/anthropic_api/routes.py: 100%
98 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""HTTP route for the Anthropic-compatible ``/v1/messages``."""
3from __future__ import annotations
5import asyncio
6import logging
7import uuid
8from collections.abc import AsyncGenerator
10from litestar import Request, Router, post
11from litestar.background_tasks import BackgroundTask
12from litestar.exceptions import NotAuthorizedException, ValidationException
13from litestar.response import Response, Stream
15from lilbee.app.services import get_services
16from lilbee.server.anthropic_api.errors import anthropic_error_body, anthropic_error_type
17from lilbee.server.anthropic_api.models import (
18 AnthropicEventType,
19 MessagesRequest,
20 MessagesResponse,
21)
22from lilbee.server.anthropic_api.streaming import encode_anthropic_event, encode_anthropic_sse
23from lilbee.server.anthropic_api.translate import (
24 canonical_stream_to_anthropic_events,
25 canonical_to_messages_response,
26 messages_to_canonical_request,
27)
28from lilbee.server.auth import auth_checked_in_handler, session_manager
29from lilbee.server.chat_completions_api.errors import (
30 CompletionsErrorCode,
31 classify_provider_error,
32)
33from lilbee.server.chat_dispatch.canonical import CanonicalChatRequest
34from lilbee.server.chat_dispatch.concurrency import (
35 ChatBusyError,
36 ChatSlotGuard,
37 acquire_chat_slot_or_busy,
38)
39from lilbee.server.chat_dispatch.dispatch import (
40 dispatch_chat,
41 dispatch_chat_stream,
42 preflight_chat_request,
43)
44from lilbee.server.handlers.sse import SSE_MEDIA_TYPE
45from lilbee.server.validation_format import format_validation
47log = logging.getLogger(__name__)
49_INTERNAL_ERROR_MESSAGE = "Internal server error. Check the server logs for details."
52async def _auth_before_request(request: Request) -> Response | None:
53 """Reject an unauthenticated caller before Litestar parses the body.
55 Same rationale as the completions surface: the auth answer must ride this
56 surface's own error envelope, and it must win over body validation.
57 """
58 return _auth_failure(request)
61@post("/v1/messages", status_code=200, before_request=_auth_before_request)
62@auth_checked_in_handler
63async def messages_endpoint(request: Request, data: MessagesRequest) -> Response | Stream:
64 """``/v1/messages`` (stream + non-stream + tools), Anthropic wire format.
66 ``stream: true`` switches the 200 response from JSON to the Anthropic SSE
67 event stream; the request body picks the arm, matching Anthropic's own
68 contract.
69 """
70 try:
71 req = messages_to_canonical_request(data)
72 except ValueError as exc:
73 # Wire-valid but untranslatable (image content, bare tool choice).
74 return _error_response(400, CompletionsErrorCode.INVALID_REQUEST, str(exc))
76 preflight = await _preflight_resolved_model(req)
77 if isinstance(preflight, Response):
78 return preflight
79 resolved_model = preflight
81 try:
82 await acquire_chat_slot_or_busy(get_services().provider.max_concurrent_chats())
83 except ChatBusyError:
84 return _error_response(
85 429,
86 CompletionsErrorCode.RATE_LIMIT_EXCEEDED,
87 "Backend is busy. Retry in a moment.",
88 headers={"Retry-After": "1"},
89 )
91 guard = ChatSlotGuard()
92 if req.stream:
93 # The after-send hook frees the slot when a disconnect lands before the
94 # generator's first iteration (its finally never runs in that case).
95 return Stream(
96 _gated_messages_stream(req, guard, model=resolved_model),
97 media_type=SSE_MEDIA_TYPE,
98 background=BackgroundTask(guard.release),
99 )
100 return await _run_non_stream(req, guard, canonical_model=resolved_model)
103async def _preflight_resolved_model(req: CanonicalChatRequest) -> str | Response:
104 """Validate *req* before any streaming response starts (see completions)."""
105 try:
106 return await asyncio.to_thread(preflight_chat_request, req)
107 except Exception as exc:
108 classified = classify_provider_error(exc)
109 if classified is None:
110 return _internal_error_response()
111 return _error_response(classified.http_status, classified.code, classified.message)
114async def _run_non_stream(
115 req: CanonicalChatRequest, guard: ChatSlotGuard, *, canonical_model: str
116) -> Response:
117 """Dispatch a non-streaming chat call, translating errors to the envelope."""
118 try:
119 resp = await asyncio.to_thread(dispatch_chat, req, canonical_model=canonical_model)
120 except Exception as exc:
121 classified = classify_provider_error(exc)
122 if classified is None:
123 return _internal_error_response()
124 return _error_response(classified.http_status, classified.code, classified.message)
125 finally:
126 await guard.release()
127 body: MessagesResponse = canonical_to_messages_response(resp, response_id=_response_id())
128 return Response(body.model_dump(), media_type="application/json")
131async def _gated_messages_stream(
132 req: CanonicalChatRequest,
133 guard: ChatSlotGuard,
134 *,
135 model: str,
136) -> AsyncGenerator[bytes, None]:
137 """Drive dispatch -> translate -> SSE-encode, freeing the slot on exit.
139 A mid-stream failure surfaces as Anthropic's ``event: error`` frame; the
140 headers are already flushed at 200 by then, so the frame is the only
141 channel left.
142 """
143 response_id = _response_id()
144 try:
145 try:
146 events = dispatch_chat_stream(req, canonical_model=model)
147 pairs = canonical_stream_to_anthropic_events(
148 events, model=model, response_id=response_id
149 )
150 async for frame in encode_anthropic_sse(pairs):
151 yield frame
152 except Exception as exc:
153 classified = classify_provider_error(exc)
154 if classified is None:
155 log.exception("anthropic messages stream failed")
156 body = anthropic_error_body("api_error", _INTERNAL_ERROR_MESSAGE)
157 else:
158 body = anthropic_error_body(
159 anthropic_error_type(classified.code), classified.message
160 )
161 yield encode_anthropic_event(AnthropicEventType.ERROR, body)
162 finally:
163 await guard.release()
166def _internal_error_response() -> Response:
167 """Log and return the generic api_error 500 envelope."""
168 log.exception("messages_endpoint failed")
169 return _error_response(500, CompletionsErrorCode.INTERNAL_ERROR, _INTERNAL_ERROR_MESSAGE)
172def _error_response(
173 status: int,
174 code: CompletionsErrorCode,
175 message: str,
176 *,
177 headers: dict[str, str] | None = None,
178) -> Response:
179 return Response(
180 anthropic_error_body(anthropic_error_type(code), message),
181 status_code=status,
182 headers=headers or {},
183 media_type="application/json",
184 )
187def _auth_failure(request: Request) -> Response | None:
188 """Return a 401 envelope if the bearer token is missing/wrong, else None.
190 Claude Code sends ``ANTHROPIC_AUTH_TOKEN`` as a bearer Authorization
191 header, which is exactly the session token check every /v1 route runs.
192 """
193 auth_header = request.headers.get("authorization", "")
194 try:
195 authorized = session_manager.validate(auth_header)
196 except NotAuthorizedException:
197 authorized = False
198 if authorized:
199 return None
200 return _error_response(401, CompletionsErrorCode.INVALID_API_KEY, "Missing or invalid API key.")
203def _validation_exception_handler(_: Request, exc: ValidationException) -> Response:
204 """Wrap Litestar's body-parse failures in the Anthropic error envelope."""
205 return _error_response(400, CompletionsErrorCode.INVALID_REQUEST, format_validation(exc))
208def _response_id() -> str:
209 """Anthropic-style ``msg_*`` id."""
210 return f"msg_{uuid.uuid4().hex[:24]}"
213anthropic_router = Router(
214 path="/",
215 route_handlers=[messages_endpoint],
216 exception_handlers={ValidationException: _validation_exception_handler},
217)