feat: serve the Anthropic Messages API on /v1/messages

src/inference/app.py

@@ -232,6 +232,38 @@ async def create_completion(
     )
 
 
+@app.post("/v1/messages")
+async def create_message(
+    request: Request,
+    background_tasks: BackgroundTasks,
+    authorization: str = Header(None),
+    x_api_key: str = Header(None, alias="x-api-key"),
+    sandbox: str = Header(None, alias="x-sandbox"),
+):
+    """
+    Anthropic Messages API surface.
+    Translates in, runs the ordinary completion path, translates out.
+    """
+    is_sandbox = sandbox.lower() == "true" if sandbox else False
+    # Anthropic clients send the key as x-api-key rather than a bearer token.
+    # Sandbox still goes through extract_api_key, which verifies the JWT in the
+    # bearer header; x-api-key must not be a way around that.
+    if is_sandbox:
+        api_key = extract_api_key(authorization, True)
+    else:
+        api_key = x_api_key or extract_api_key(authorization, False)
+    body = await parse_json_body(request)
+    client_ip = extract_client_ip(request)
+
+    return await OrchestrationService.handle_messages(
+        api_key=api_key,
+        body=body,
+        background_tasks=background_tasks,
+        ip_address=client_ip,
+        sandbox=is_sandbox,
+    )
+
+
 @app.post("/v1/embeddings")
 async def create_embeddings(
     request: Request,

src/inference/core/handlers/__init__.py

@@ -3,11 +3,13 @@
 from .completion import CompletionHandler
 from .embedding import EmbeddingHandler
 from .image import ImageHandler
+from .messages import MessagesHandler
 from .video import VideoHandler
 
 __all__ = [
     "CompletionHandler",
     "EmbeddingHandler",
     "ImageHandler",
+    "MessagesHandler",
     "VideoHandler",
 ]

src/inference/core/handlers/messages.py

@@ -0,0 +1,75 @@
+"""Handler for `/v1/messages`, the Anthropic Messages API surface.
+
+Translates in, runs the ordinary completion path, translates out. Nothing in
+that path is aware of the surface, which is what keeps API key resolution,
+quotas, rate limiting, routing and logging identical to
+`/v1/chat/completions`.
+"""
+
+import codecs
+import logging
+from typing import Dict, Optional
+
+from fastapi import BackgroundTasks
+from fastapi.responses import StreamingResponse
+
+from ..surfaces import anthropic as surface
+from .completion import CompletionHandler
+
+logger = logging.getLogger(__name__)
+
+
+class MessagesHandler:
+    """The Anthropic Messages surface over the OpenAI completion path."""
+
+    @staticmethod
+    async def handle(
+        api_key: str,
+        body: Dict,
+        background_tasks: BackgroundTasks,
+        ip_address: Optional[str] = None,
+        sandbox: bool = False,
+    ):
+        result = await CompletionHandler.handle(
+            api_key=api_key,
+            body=surface.request_to_openai(body),
+            background_tasks=background_tasks,
+            ip_address=ip_address,
+            sandbox=sandbox,
+        )
+
+        if isinstance(result, StreamingResponse):
+            translator = surface.StreamTranslator(model=body.get("model") or "")
+            result.body_iterator = MessagesHandler._translate_stream(
+                result.body_iterator, translator
+            )
+            return result
+
+        return surface.response_to_anthropic(result)
+
+    @staticmethod
+    async def _translate_stream(source, translator: "surface.StreamTranslator"):
+        """OpenAI SSE in, Anthropic SSE out.
+
+        `process_stream` yields whole lines, so an event never arrives split.
+        The decoder is incremental anyway, because that guarantee belongs to a
+        different module and would break quietly here if it ever changed.
+        """
+        decoder = codecs.getincrementaldecoder("utf-8")(errors="ignore")
+        try:
+            async for chunk in source:
+                text = (
+                    decoder.decode(chunk)
+                    if isinstance(chunk, bytes)
+                    else str(chunk)
+                )
+                for event in surface.iter_openai_events(text):
+                    if event is None:
+                        continue  # `[DONE]`: Anthropic closes differently
+                    for out in translator.chunk(event):
+                        yield out
+        finally:
+            # The envelope must close even if the client disconnects or
+            # upstream dies, or a reader is left waiting on `message_stop`.
+            for out in translator.finish():
+                yield out

src/inference/core/orchestrator.py

@@ -8,7 +8,13 @@
 import logging
 
 from inference.client import api_gateway_client
-from .handlers import CompletionHandler, EmbeddingHandler, ImageHandler, VideoHandler
+from .handlers import (
+    CompletionHandler,
+    EmbeddingHandler,
+    ImageHandler,
+    MessagesHandler,
+    VideoHandler,
+)
 
 # Re-export dependencies that tests patch via this module's namespace.
 from .service import GatewayService  # noqa: F401
@@ -34,6 +40,9 @@ async def list_models(api_key: str, sandbox: bool = False):
     # --- Chat Completion ---
     handle_completion = CompletionHandler.handle
 
+    # --- Anthropic Messages surface ---
+    handle_messages = MessagesHandler.handle
+
     # --- Embeddings ---
     handle_embeddings = EmbeddingHandler.handle
 

src/inference/core/surfaces/__init__.py

@@ -0,0 +1,12 @@
+"""Client-facing API surfaces.
+
+A surface translates between the shape a client speaks and the OpenAI shape
+used internally. It is the mirror of a provider adapter, which translates
+between an upstream's shape and the same internal one.
+
+    client  -> [surface in ] -> internal (OpenAI) -> [provider out] -> upstream
+    client <-  [surface out] <- internal (OpenAI) <- [provider in ] <- upstream
+
+Keeping them separate is what lets a new client format cost a translator rather
+than a second request pipeline.
+"""