[serve][llm] Add Native Anthropic Messages API (/v1/messages) Support - #65486
[serve][llm] Add Native Anthropic Messages API (/v1/messages) Support#65486wanadzhar913 wants to merge 3 commits into
Conversation
Signed-off-by: wanadzhar913 <adzhar.faiq@gmail.com>
Signed-off-by: wanadzhar913 <adzhar.faiq@gmail.com>
Signed-off-by: wanadzhar913 <adzhar.faiq@gmail.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces support for the Anthropic Messages API and token counting endpoints, integrating them with the Ray Serve LLM engine and vLLM. It adds new ingress routing, Pydantic models, and helper utilities, alongside comprehensive tests. The review feedback highlights several critical robustness improvements, including handling potential null values and non-integer status codes in error translation, catching stream-time exceptions during message generation, gracefully handling empty generators or StopAsyncIteration in the ingress endpoints, and supporting string-based class names in configuration comparisons.
| def translate_error_response(response: ErrorResponse) -> JSONResponse: | ||
| anthropic_error = AnthropicErrorResponse( | ||
| error=AnthropicError( | ||
| type=response.error.type, | ||
| message=response.error.message, | ||
| ) | ||
| ) | ||
| return JSONResponse( | ||
| status_code=response.error.code, | ||
| content=anthropic_error.model_dump(exclude_none=True), | ||
| ) |
There was a problem hiding this comment.
Accessing properties on response.error without a None check violates defensive programming guidelines. Additionally, response.error.code can be a string or None in OpenAI/vLLM error responses, which will cause a runtime error in Starlette's JSONResponse since status_code must be an integer. We should safely convert the code to an integer and handle potential None values.
def translate_error_response(response: ErrorResponse) -> JSONResponse:
error_info = response.error
if error_info is None:
return JSONResponse(
status_code=500,
content={"error": {"type": "internal_error", "message": "An unknown error occurred."}},
)
try:
status_code = int(error_info.code) if error_info.code is not None else 500
except (ValueError, TypeError):
status_code = 500
anthropic_error = AnthropicErrorResponse(
error=AnthropicError(
type=error_info.type or "internal_error",
message=error_info.message or "An unknown error occurred.",
)
)
return JSONResponse(
status_code=status_code,
content=anthropic_error.model_dump(exclude_none=True),
)| if isinstance(messages_response, AsyncGenerator): | ||
| async for response in messages_response: | ||
| if not isinstance(response, str): | ||
| raise ValueError( | ||
| "Expected create_messages to return a stream of strings, " | ||
| f"got an item with type {type(response)}" | ||
| ) | ||
| yield response |
There was a problem hiding this comment.
Any exception raised during the iteration of the messages_response async generator (e.g., token generation errors, connection drops, or engine failures) will propagate uncaught because the async for loop is outside the try...except block. We should wrap the iteration in a try...except block to ensure all stream-time exceptions are properly caught and translated into error responses.
if isinstance(messages_response, AsyncGenerator):
try:
async for response in messages_response:
if not isinstance(response, str):
raise ValueError(
"Expected create_messages to return a stream of strings, "
f"got an item with type {type(response)}"
)
yield response
except Exception as e:
yield self._make_error_response(self._anthropic_serving_messages, e)| initial_response, gen = await _peek_at_generator(gen) | ||
|
|
||
| if isinstance(initial_response, ErrorResponse): | ||
| return translate_error_response(initial_response) | ||
|
|
||
| if isinstance(initial_response, str): | ||
|
|
||
| async def stream(): | ||
| yield initial_response | ||
| async for item in gen: | ||
| yield item | ||
|
|
||
| return anthropic_messages_http_response(stream()) | ||
|
|
||
| return anthropic_messages_http_response(initial_response) |
There was a problem hiding this comment.
If the generator gen is empty, initial_response will be None. Passing None to anthropic_messages_http_response will result in an invalid StreamingResponse with None content, leading to a broken or hanging connection. We should explicitly handle the None case by raising an HTTPException.
initial_response, gen = await _peek_at_generator(gen)
if initial_response is None:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Engine returned an empty response.",
)
if isinstance(initial_response, ErrorResponse):
return translate_error_response(initial_response)
if isinstance(initial_response, str):
async def stream():
yield initial_response
async for item in gen:
yield item
return anthropic_messages_http_response(stream())
return anthropic_messages_http_response(initial_response)| results = model_handle.count_tokens.remote(body, raw_request_info) | ||
| result = await results.__anext__() |
There was a problem hiding this comment.
await results.__anext__() can raise StopAsyncIteration if the generator is empty. This uncaught exception will bubble up and cause a 500 Internal Server Error. We should wrap it in a try...except StopAsyncIteration block to handle it gracefully.
| results = model_handle.count_tokens.remote(body, raw_request_info) | |
| result = await results.__anext__() | |
| raw_request_info = RawRequestInfo.from_starlette_request(request) | |
| results = model_handle.count_tokens.remote(body, raw_request_info) | |
| try: | |
| result = await results.__anext__() | |
| except StopAsyncIteration: | |
| raise HTTPException( | |
| status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, | |
| detail="Engine did not return any response for token counting.", | |
| ) |
| ingress_cls_config = builder_config.ingress_cls_config | ||
| if ingress_cls_config.ingress_cls == OpenAiIngress: | ||
| ingress_cls_config = IngressClsConfig( | ||
| ingress_cls=AnthropicIngress, | ||
| ingress_extra_kwargs=ingress_cls_config.ingress_extra_kwargs, | ||
| ) |
There was a problem hiding this comment.
Direct comparison ingress_cls_config.ingress_cls == OpenAiIngress will fail if ingress_cls is configured as a string (e.g., "OpenAiIngress" or "ray.serve.llm.OpenAiIngress" in a YAML configuration file). This will prevent the application from automatically switching to AnthropicIngress, leading to 404 errors on the Anthropic endpoints. We should handle both class and string comparisons.
| ingress_cls_config = builder_config.ingress_cls_config | |
| if ingress_cls_config.ingress_cls == OpenAiIngress: | |
| ingress_cls_config = IngressClsConfig( | |
| ingress_cls=AnthropicIngress, | |
| ingress_extra_kwargs=ingress_cls_config.ingress_extra_kwargs, | |
| ) | |
| ingress_cls = ingress_cls_config.ingress_cls | |
| if ingress_cls == OpenAiIngress or (isinstance(ingress_cls, str) and ingress_cls.endswith("OpenAiIngress")): | |
| ingress_cls_config = IngressClsConfig( | |
| ingress_cls=AnthropicIngress, | |
| ingress_extra_kwargs=ingress_cls_config.ingress_extra_kwargs, | |
| ) |
| ingress_cls_config = builder_config.ingress_cls_config | ||
| if ingress_cls_config.ingress_cls == OpenAiIngress: | ||
| ingress_cls_config = IngressClsConfig( | ||
| ingress_cls=AnthropicIngress, | ||
| ingress_extra_kwargs=ingress_cls_config.ingress_extra_kwargs, | ||
| ) |
There was a problem hiding this comment.
Direct comparison ingress_cls_config.ingress_cls == OpenAiIngress will fail if ingress_cls is configured as a string (e.g., "OpenAiIngress" or "ray.serve.llm.OpenAiIngress" in a YAML configuration file). This will prevent the application from automatically switching to AnthropicIngress, leading to 404 errors on the Anthropic endpoints. We should handle both class and string comparisons.
| ingress_cls_config = builder_config.ingress_cls_config | |
| if ingress_cls_config.ingress_cls == OpenAiIngress: | |
| ingress_cls_config = IngressClsConfig( | |
| ingress_cls=AnthropicIngress, | |
| ingress_extra_kwargs=ingress_cls_config.ingress_extra_kwargs, | |
| ) | |
| ingress_cls_config = builder_config.ingress_cls_config | |
| ingress_cls = ingress_cls_config.ingress_cls | |
| if ingress_cls == OpenAiIngress or (isinstance(ingress_cls, str) and ingress_cls.endswith("OpenAiIngress")): | |
| ingress_cls_config = IngressClsConfig( | |
| ingress_cls=AnthropicIngress, | |
| ingress_extra_kwargs=ingress_cls_config.ingress_extra_kwargs, | |
| ) |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
Reviewed by Cursor Bugbot for commit ea9fe11. Configure here.
| raise HTTPException( | ||
| status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, | ||
| detail="Unexpected response type from count_tokens", | ||
| ) |
There was a problem hiding this comment.
Ingress errors not Anthropic-shaped
Medium Severity
Engine ErrorResponse values are translated to Anthropic payloads, but missing-model lookups, FastAPI HTTPExceptions, and the shared exception middleware still emit OpenAI-shaped JSON (error.message/error.code, or detail). Anthropic SDKs and Claude Code expect {type: "error", error: {type, message}}, so common failures such as an unknown model will not parse as API errors.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit ea9fe11. Configure here.
ray-gardener
Bot
added
serve
eicherseiji
left a comment
eicherseiji
left a comment
There was a problem hiding this comment.
Hi @wanadzhar913, this is an exciting contribution! Please reach out on Ray Slack so we can land this
|
|
||
| if isinstance(initial_response, str): | ||
|
|
||
| async def stream(): |
There was a problem hiding this comment.
Could we stream gen directly here? _peek_at_generator() returns a replacement generator that already replays initial_response, so yielding it again duplicates the first SSE event—usually message_start—and produces an invalid event sequence for strict Anthropic clients.
| DEFAULT_MAX_ONGOING_REQUESTS, | ||
| DEFAULT_MAX_TARGET_ONGOING_REQUESTS, | ||
| ) | ||
| from ray.llm._internal.serve.core.configs.anthropic_api_models import ( |
There was a problem hiding this comment.
Could we keep these vLLM-only imports off the SGLang import path, for example by loading them lazily on the Anthropic path? The supported SGLang image uninstalls vLLM, but ray.serve.llm imports the builder and then this module, so this raises before the existing OpenAI/SGLang code can load. This breaks TestSGLangProtocolDecoupling.test_modules_importable_without_vllm.
| default_ingress_options, builder_config.ingress_deployment_config | ||
| ) | ||
|
|
||
| ingress_cls = make_fastapi_ingress( |
There was a problem hiding this comment.
Could we install Anthropic-specific validation and HTTPException handlers on this app? make_fastapi_ingress() currently uses the shared init() app, whose validation handler emits an OpenAI error envelope and whose default HTTP error response uses {"detail": ...}. As a result, malformed bodies and unknown-model errors on both Anthropic endpoints return incompatible response shapes instead of Anthropic error envelopes.
2 participants
Description
Adds first-class Anthropic Messages API support to Ray Serve LLM for vLLM-based deployments. This allows Anthropic-compatible clients, including Claude Code, to communicate directly with Ray Serve LLM without an additional translation proxy such as LiteLLM or Claude Code Router. The initial implementation supports:
POST /v1/messagesPOST /v1/messages/count_tokensRelated issues
Implementation
LLMEngine,LLMServerProtocol, andLLMServerwithmessagesandcount_tokensoperations.AnthropicServingMessages.AnthropicIngress(which inherits fromOpenaiIngress), reusing the existing ingress model resolution, LoRA routing, handle caching, session affinity, and deployment configuration.build_anthropic_appwith support for both standard ingress and direct streaming.Testing
pytest
API
I tested this using the
anthropic==0.106.0&vllm==0.26.0python library, and have not used Claude Code (yet). Testing infra had the following specs:Ray LLMServer Setup Script
Ray LLMServer Setup Logs
This sets up the
ANTHROPIC_BASE_URLat: http://localhost:8000/v1/messages.Anthropic API Tests Script
Anthropic API Tests Logs
AI assistance
AI assistance was used during implementation. I reviewed the changed code and am responsible for understanding, testing, and maintaining it. Used Cursor: GPT-5.6 Sol, Cursor Grok 4.6