From cc451671b5b69c64c25e22a0b9f02f80e56bb960 Mon Sep 17 00:00:00 2001 From: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com> Date: Sat, 21 Feb 2026 06:37:38 -0500 Subject: [PATCH] [FEAT] Add Anthropic compatible API endpoint (#18630) Signed-off-by: Xinyuan Tong --- .../srt/entrypoints/anthropic/__init__.py | 0 .../srt/entrypoints/anthropic/protocol.py | 178 +++++ .../srt/entrypoints/anthropic/serving.py | 706 ++++++++++++++++++ python/sglang/srt/entrypoints/http_server.py | 33 + test/manual/vlm/test_anthropic_vision.py | 433 +++++++++++ .../basic/test_anthropic_server.py | 493 ++++++++++++ .../function_call/test_anthropic_tool_use.py | 555 ++++++++++++++ 7 files changed, 2398 insertions(+) create mode 100644 python/sglang/srt/entrypoints/anthropic/__init__.py create mode 100644 python/sglang/srt/entrypoints/anthropic/protocol.py create mode 100644 python/sglang/srt/entrypoints/anthropic/serving.py create mode 100644 test/manual/vlm/test_anthropic_vision.py create mode 100644 test/registered/openai_server/basic/test_anthropic_server.py create mode 100644 test/registered/openai_server/function_call/test_anthropic_tool_use.py diff --git a/python/sglang/srt/entrypoints/anthropic/__init__.py b/python/sglang/srt/entrypoints/anthropic/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/python/sglang/srt/entrypoints/anthropic/protocol.py b/python/sglang/srt/entrypoints/anthropic/protocol.py new file mode 100644 index 000000000..6e4b2d7d7 --- /dev/null +++ b/python/sglang/srt/entrypoints/anthropic/protocol.py @@ -0,0 +1,178 @@ +"""Pydantic models for Anthropic Messages API protocol""" + +import uuid +from typing import Any, Literal, Optional + +from pydantic import BaseModel, Field, field_validator + + +class AnthropicError(BaseModel): + """Error structure for Anthropic API""" + + type: str + message: str + + +class AnthropicErrorResponse(BaseModel): + """Error response structure for Anthropic API""" + + type: Literal["error"] = "error" + error: AnthropicError + + +class AnthropicUsage(BaseModel): + """Token usage information""" + + input_tokens: int + output_tokens: int + cache_creation_input_tokens: Optional[int] = None + cache_read_input_tokens: Optional[int] = None + + +class AnthropicContentBlock(BaseModel): + """Content block in message""" + + type: Literal[ + "text", "image", "tool_use", "tool_result", "thinking", "redacted_thinking" + ] + text: Optional[str] = None + # For image content + source: Optional[dict[str, Any]] = None + # For tool use/result + id: Optional[str] = None + tool_use_id: Optional[str] = None + name: Optional[str] = None + input: Optional[dict[str, Any]] = None + content: Optional[str | list[dict[str, Any]]] = None + is_error: Optional[bool] = None + # For thinking content + thinking: Optional[str] = None + signature: Optional[str] = None + + +class AnthropicMessage(BaseModel): + """Message structure""" + + role: Literal["user", "assistant"] + content: str | list[AnthropicContentBlock] + + +class AnthropicTool(BaseModel): + """Tool definition""" + + name: str + description: Optional[str] = None + input_schema: dict[str, Any] + + @field_validator("input_schema") + @classmethod + def validate_input_schema(cls, v): + if not isinstance(v, dict): + raise ValueError("input_schema must be a dictionary") + if "type" not in v: + v["type"] = "object" + return v + + +class AnthropicToolChoice(BaseModel): + """Tool Choice definition""" + + type: Literal["auto", "any", "tool", "none"] + name: Optional[str] = None + + +class AnthropicCountTokensRequest(BaseModel): + """Anthropic Count Tokens API request""" + + model: str + messages: list[AnthropicMessage] + system: Optional[str | list[AnthropicContentBlock]] = None + tool_choice: Optional[AnthropicToolChoice] = None + tools: Optional[list[AnthropicTool]] = None + + +class AnthropicCountTokensResponse(BaseModel): + """Anthropic Count Tokens API response""" + + input_tokens: int + + +class AnthropicMessagesRequest(BaseModel): + """Anthropic Messages API request""" + + model: str + messages: list[AnthropicMessage] + max_tokens: int + metadata: Optional[dict[str, Any]] = None + stop_sequences: Optional[list[str]] = None + stream: Optional[bool] = False + system: Optional[str | list[AnthropicContentBlock]] = None + temperature: Optional[float] = None + tool_choice: Optional[AnthropicToolChoice] = None + tools: Optional[list[AnthropicTool]] = None + top_k: Optional[int] = None + top_p: Optional[float] = None + + @field_validator("model") + @classmethod + def validate_model(cls, v): + if not v: + raise ValueError("Model is required") + return v + + @field_validator("max_tokens") + @classmethod + def validate_max_tokens(cls, v): + if v <= 0: + raise ValueError("max_tokens must be positive") + return v + + +class AnthropicDelta(BaseModel): + """Delta for streaming responses""" + + type: Optional[Literal["text_delta", "input_json_delta"]] = None + text: Optional[str] = None + partial_json: Optional[str] = None + + # Message delta fields + stop_reason: Optional[ + Literal["end_turn", "max_tokens", "stop_sequence", "tool_use"] + ] = None + stop_sequence: Optional[str] = None + + +class AnthropicStreamEvent(BaseModel): + """Streaming event""" + + type: Literal[ + "message_start", + "message_delta", + "message_stop", + "content_block_start", + "content_block_delta", + "content_block_stop", + "ping", + "error", + ] + message: Optional["AnthropicMessagesResponse"] = None + delta: Optional[AnthropicDelta] = None + content_block: Optional[AnthropicContentBlock] = None + index: Optional[int] = None + error: Optional[AnthropicError] = None + usage: Optional[AnthropicUsage] = None + + +class AnthropicMessagesResponse(BaseModel): + """Anthropic Messages API response""" + + id: str = Field(default_factory=lambda: f"msg_{uuid.uuid4().hex}") + type: Literal["message"] = "message" + role: Literal["assistant"] = "assistant" + content: list[AnthropicContentBlock] + model: str + stop_reason: Optional[ + Literal["end_turn", "max_tokens", "stop_sequence", "tool_use"] + ] = None + stop_sequence: Optional[str] = None + usage: Optional[AnthropicUsage] = None diff --git a/python/sglang/srt/entrypoints/anthropic/serving.py b/python/sglang/srt/entrypoints/anthropic/serving.py new file mode 100644 index 000000000..e1de35dcb --- /dev/null +++ b/python/sglang/srt/entrypoints/anthropic/serving.py @@ -0,0 +1,706 @@ +"""Handler for Anthropic Messages API requests. + +Converts Anthropic requests to OpenAI ChatCompletion format, delegates to +OpenAIServingChat for processing, and converts responses back to Anthropic format. +""" + +from __future__ import annotations + +import json +import logging +import time +import uuid +from typing import TYPE_CHECKING, AsyncGenerator, Optional, Union + +from fastapi import Request +from fastapi.responses import JSONResponse, StreamingResponse + +from sglang.srt.entrypoints.anthropic.protocol import ( + AnthropicContentBlock, + AnthropicCountTokensRequest, + AnthropicCountTokensResponse, + AnthropicDelta, + AnthropicError, + AnthropicErrorResponse, + AnthropicMessagesRequest, + AnthropicMessagesResponse, + AnthropicStreamEvent, + AnthropicUsage, +) +from sglang.srt.entrypoints.openai.protocol import ( + ChatCompletionRequest, + ChatCompletionResponse, + ChatCompletionStreamResponse, + StreamOptions, + Tool, + ToolChoice, + ToolChoiceFuncName, +) + +if TYPE_CHECKING: + from sglang.srt.entrypoints.openai.serving_chat import OpenAIServingChat + +logger = logging.getLogger(__name__) + +# Map OpenAI finish reasons to Anthropic stop reasons +STOP_REASON_MAP = { + "stop": "end_turn", + "length": "max_tokens", + "tool_calls": "tool_use", +} + + +def _wrap_sse_event(data: str, event_type: str) -> str: + """Format an Anthropic SSE event with event type and data lines.""" + return f"event: {event_type}\ndata: {data}\n\n" + + +class AnthropicServing: + """Handler for Anthropic Messages API requests. + + Acts as a translation layer between Anthropic's Messages API and SGLang's + OpenAI-compatible chat completion infrastructure. + """ + + def __init__(self, openai_serving_chat: OpenAIServingChat): + self.openai_serving_chat = openai_serving_chat + + async def handle_messages( + self, + request: AnthropicMessagesRequest, + raw_request: Request, + ) -> Union[JSONResponse, StreamingResponse]: + """Main entry point for /v1/messages endpoint.""" + try: + chat_request = self._convert_to_chat_completion_request(request) + except Exception as e: + logger.exception("Error converting Anthropic request: %s", e) + return self._error_response( + status_code=400, + error_type="invalid_request_error", + message=str(e), + ) + + if request.stream: + return await self._handle_streaming(chat_request, request, raw_request) + else: + return await self._handle_non_streaming(chat_request, request, raw_request) + + def _convert_to_chat_completion_request( + self, anthropic_request: AnthropicMessagesRequest + ) -> ChatCompletionRequest: + """Convert an Anthropic Messages request to an OpenAI ChatCompletion request.""" + openai_messages = [] + + # Add system message if provided + if anthropic_request.system: + if isinstance(anthropic_request.system, str): + openai_messages.append( + {"role": "system", "content": anthropic_request.system} + ) + else: + system_parts = [] + for block in anthropic_request.system: + if block.type == "text" and block.text: + system_parts.append(block.text) + system_text = "\n".join(system_parts) + openai_messages.append({"role": "system", "content": system_text}) + + # Convert messages + for msg in anthropic_request.messages: + if isinstance(msg.content, str): + openai_messages.append({"role": msg.role, "content": msg.content}) + continue + + # Complex content with blocks + openai_msg = {"role": msg.role} + content_parts = [] + tool_calls = [] + + for block in msg.content: + if block.type == "text" and block.text: + content_parts.append({"type": "text", "text": block.text}) + + elif block.type == "image" and block.source: + media_type = block.source.get("media_type", "image/png") + data = block.source.get("data", "") + content_parts.append( + { + "type": "image_url", + "image_url": { + "url": f"data:{media_type};base64,{data}", + }, + } + ) + + elif block.type == "tool_use": + tool_call = { + "id": block.id or f"call_{uuid.uuid4().hex}", + "type": "function", + "function": { + "name": block.name or "", + "arguments": json.dumps(block.input or {}), + }, + } + tool_calls.append(tool_call) + + elif block.type == "tool_result": + # Extract text content from list or string + if isinstance(block.content, list): + tool_content = "\n".join( + item.get("text", "") + for item in block.content + if isinstance(item, dict) and item.get("type") == "text" + ) + else: + tool_content = str(block.content) if block.content else "" + + # Use tool_use_id (per spec) with fallback to id + tool_call_id = block.tool_use_id or block.id or "" + + # Tool results from user become separate tool messages + if msg.role == "user": + openai_messages.append( + { + "role": "tool", + "tool_call_id": tool_call_id, + "content": tool_content, + } + ) + else: + content_parts.append( + { + "type": "text", + "text": f"Tool result: {tool_content}", + } + ) + + # Attach tool calls to assistant messages + if tool_calls: + openai_msg["tool_calls"] = tool_calls + + # Attach content + if content_parts: + if len(content_parts) == 1 and content_parts[0]["type"] == "text": + openai_msg["content"] = content_parts[0]["text"] + else: + openai_msg["content"] = content_parts + elif not tool_calls: + continue + + openai_messages.append(openai_msg) + + # Build ChatCompletionRequest + request_data = { + "messages": openai_messages, + "model": anthropic_request.model, + "max_tokens": anthropic_request.max_tokens, + "stream": anthropic_request.stream or False, + } + + if anthropic_request.temperature is not None: + request_data["temperature"] = anthropic_request.temperature + if anthropic_request.top_p is not None: + request_data["top_p"] = anthropic_request.top_p + if anthropic_request.top_k is not None: + request_data["top_k"] = anthropic_request.top_k + if anthropic_request.stop_sequences is not None: + request_data["stop"] = anthropic_request.stop_sequences + + # Enable usage in stream so we can report it + if anthropic_request.stream: + request_data["stream_options"] = StreamOptions(include_usage=True) + + chat_request = ChatCompletionRequest(**request_data) + + # Convert tools + if anthropic_request.tools: + tools = [] + for tool in anthropic_request.tools: + tools.append( + Tool( + type="function", + function={ + "name": tool.name, + "description": tool.description or "", + "parameters": tool.input_schema, + }, + ) + ) + chat_request.tools = tools + + # Convert tool choice + if anthropic_request.tool_choice is not None: + if anthropic_request.tool_choice.type == "none": + chat_request.tool_choice = "none" + elif anthropic_request.tool_choice.type == "auto": + chat_request.tool_choice = "auto" + elif anthropic_request.tool_choice.type == "any": + chat_request.tool_choice = "required" + elif anthropic_request.tool_choice.type == "tool": + chat_request.tool_choice = ToolChoice( + type="function", + function=ToolChoiceFuncName( + name=anthropic_request.tool_choice.name + ), + ) + elif anthropic_request.tools: + # Default to auto when tools are provided + chat_request.tool_choice = "auto" + + return chat_request + + async def _handle_non_streaming( + self, + chat_request: ChatCompletionRequest, + anthropic_request: AnthropicMessagesRequest, + raw_request: Request, + ) -> JSONResponse: + """Handle non-streaming Anthropic request by delegating to OpenAI handler.""" + received_time = time.time() + received_time_perf = time.perf_counter() + + # Validate + error_msg = self.openai_serving_chat._validate_request(chat_request) + if error_msg: + return self._error_response( + status_code=400, + error_type="invalid_request_error", + message=error_msg, + ) + + try: + # Convert to internal request + validation_time = time.perf_counter() - received_time_perf + adapted_request, processed_request = ( + self.openai_serving_chat._convert_to_internal_request( + chat_request, raw_request + ) + ) + adapted_request.validation_time = validation_time + adapted_request.received_time = received_time + adapted_request.received_time_perf = received_time_perf + + # Get response from OpenAI handler + response = await self.openai_serving_chat._handle_non_streaming_request( + adapted_request, processed_request, raw_request + ) + except Exception as e: + logger.exception("Error processing Anthropic request: %s", e) + return self._error_response( + status_code=500, + error_type="internal_error", + message="Internal server error", + ) + + # Check for error responses from OpenAI handler + if not isinstance(response, ChatCompletionResponse): + # It's an error response (ORJSONResponse) + return self._error_response( + status_code=500, + error_type="internal_error", + message="Internal processing error", + ) + + # Convert to Anthropic response + anthropic_response = self._convert_response(response) + return JSONResponse(content=anthropic_response.model_dump(exclude_none=True)) + + async def _handle_streaming( + self, + chat_request: ChatCompletionRequest, + anthropic_request: AnthropicMessagesRequest, + raw_request: Request, + ) -> Union[StreamingResponse, JSONResponse]: + """Handle streaming Anthropic request.""" + received_time = time.time() + received_time_perf = time.perf_counter() + + # Validate + error_msg = self.openai_serving_chat._validate_request(chat_request) + if error_msg: + return self._error_response( + status_code=400, + error_type="invalid_request_error", + message=error_msg, + ) + + try: + validation_time = time.perf_counter() - received_time_perf + adapted_request, processed_request = ( + self.openai_serving_chat._convert_to_internal_request( + chat_request, raw_request + ) + ) + adapted_request.validation_time = validation_time + adapted_request.received_time = received_time + adapted_request.received_time_perf = received_time_perf + except Exception as e: + logger.exception("Error converting streaming request: %s", e) + return self._error_response( + status_code=500, + error_type="internal_error", + message="Internal server error", + ) + + return StreamingResponse( + self._generate_anthropic_stream( + adapted_request, + processed_request, + anthropic_request, + raw_request, + ), + media_type="text/event-stream", + background=self.openai_serving_chat.tokenizer_manager.create_abort_task( + adapted_request + ), + ) + + async def _generate_anthropic_stream( + self, + adapted_request, + processed_request: ChatCompletionRequest, + anthropic_request: AnthropicMessagesRequest, + raw_request: Request, + ) -> AsyncGenerator[str, None]: + """Convert OpenAI chat stream to Anthropic event stream.""" + openai_stream = self.openai_serving_chat._generate_chat_stream( + adapted_request, processed_request, raw_request + ) + + # State tracking + first_chunk = True + content_block_index = 0 + content_block_open = False + finish_reason: Optional[str] = None + usage_info: Optional[dict] = None + message_id = f"msg_{uuid.uuid4().hex}" + model = anthropic_request.model + + async for sse_line in openai_stream: + if not sse_line.startswith("data: "): + continue + + data_str = sse_line[6:].strip() + + if data_str == "[DONE]": + # Close any open content block + if content_block_open: + stop_event = AnthropicStreamEvent( + type="content_block_stop", + index=content_block_index, + ) + yield _wrap_sse_event( + stop_event.model_dump_json(exclude_none=True), + "content_block_stop", + ) + + # Emit message_delta with stop_reason and usage + stop_reason = STOP_REASON_MAP.get(finish_reason or "stop", "end_turn") + delta_event = AnthropicStreamEvent( + type="message_delta", + delta=AnthropicDelta(stop_reason=stop_reason), + usage=AnthropicUsage( + input_tokens=( + usage_info.get("input_tokens", 0) if usage_info else 0 + ), + output_tokens=( + usage_info.get("output_tokens", 0) if usage_info else 0 + ), + ), + ) + yield _wrap_sse_event( + delta_event.model_dump_json(exclude_none=True), + "message_delta", + ) + + # Emit message_stop + stop_msg = AnthropicStreamEvent(type="message_stop") + yield _wrap_sse_event( + stop_msg.model_dump_json(exclude_none=True), + "message_stop", + ) + continue + + # Parse the OpenAI chunk + try: + chunk = ChatCompletionStreamResponse.model_validate_json(data_str) + except Exception: + logger.debug("Failed to parse stream chunk: %s", data_str) + error_event = AnthropicStreamEvent( + type="error", + error=AnthropicError( + type="api_error", message="Stream processing error" + ), + ) + yield _wrap_sse_event( + error_event.model_dump_json(exclude_none=True), "error" + ) + continue + + # First chunk: emit message_start + if first_chunk: + first_chunk = False + + start_event = AnthropicStreamEvent( + type="message_start", + message=AnthropicMessagesResponse( + id=message_id, + content=[], + model=model, + usage=AnthropicUsage( + input_tokens=( + chunk.usage.prompt_tokens if chunk.usage else 0 + ), + output_tokens=0, + ), + ), + ) + yield _wrap_sse_event( + start_event.model_dump_json(exclude_none=True), + "message_start", + ) + # Skip if this was just the role chunk with empty content + if chunk.choices and chunk.choices[0].delta.content == "": + continue + + # Usage-only chunk (empty choices with usage info) + if not chunk.choices and chunk.usage: + usage_info = { + "input_tokens": chunk.usage.prompt_tokens, + "output_tokens": chunk.usage.completion_tokens or 0, + } + continue + + if not chunk.choices: + continue + + choice = chunk.choices[0] + + # Capture finish reason + if choice.finish_reason is not None: + finish_reason = choice.finish_reason + continue + + delta = choice.delta + + # Handle tool call deltas + if delta.tool_calls: + for tc in delta.tool_calls: + tc_id = tc.id + tc_func = tc.function + + # New tool call: close previous block, start new one + if tc_func and tc_func.name: + # Close previous content block if open + if content_block_open: + stop_event = AnthropicStreamEvent( + type="content_block_stop", + index=content_block_index, + ) + yield _wrap_sse_event( + stop_event.model_dump_json(exclude_none=True), + "content_block_stop", + ) + content_block_index += 1 + + # Start tool_use content block + start_event = AnthropicStreamEvent( + type="content_block_start", + index=content_block_index, + content_block=AnthropicContentBlock( + type="tool_use", + id=tc_id or f"toolu_{uuid.uuid4().hex}", + name=tc_func.name, + input={}, + ), + ) + yield _wrap_sse_event( + start_event.model_dump_json(exclude_none=True), + "content_block_start", + ) + content_block_open = True + + # Stream initial arguments if present + if tc_func.arguments: + delta_event = AnthropicStreamEvent( + type="content_block_delta", + index=content_block_index, + delta=AnthropicDelta( + type="input_json_delta", + partial_json=tc_func.arguments, + ), + ) + yield _wrap_sse_event( + delta_event.model_dump_json(exclude_none=True), + "content_block_delta", + ) + + elif tc_func and tc_func.arguments: + # Continuing arguments for current tool call + delta_event = AnthropicStreamEvent( + type="content_block_delta", + index=content_block_index, + delta=AnthropicDelta( + type="input_json_delta", + partial_json=tc_func.arguments, + ), + ) + yield _wrap_sse_event( + delta_event.model_dump_json(exclude_none=True), + "content_block_delta", + ) + continue + + # Handle text content deltas + if delta.content is not None and delta.content != "": + # Start a text content block if needed + if not content_block_open: + start_event = AnthropicStreamEvent( + type="content_block_start", + index=content_block_index, + content_block=AnthropicContentBlock(type="text", text=""), + ) + yield _wrap_sse_event( + start_event.model_dump_json(exclude_none=True), + "content_block_start", + ) + content_block_open = True + + # Emit text delta + delta_event = AnthropicStreamEvent( + type="content_block_delta", + index=content_block_index, + delta=AnthropicDelta( + type="text_delta", + text=delta.content, + ), + ) + yield _wrap_sse_event( + delta_event.model_dump_json(exclude_none=True), + "content_block_delta", + ) + + def _convert_response( + self, response: ChatCompletionResponse + ) -> AnthropicMessagesResponse: + """Convert an OpenAI ChatCompletionResponse to an Anthropic Messages response.""" + if not response.choices: + return AnthropicMessagesResponse( + content=[AnthropicContentBlock(type="text", text="")], + model=response.model, + stop_reason="end_turn", + usage=AnthropicUsage(input_tokens=0, output_tokens=0), + ) + + choice = response.choices[0] + content: list[AnthropicContentBlock] = [] + + # Add text content + if choice.message.content: + content.append( + AnthropicContentBlock(type="text", text=choice.message.content) + ) + + # Add tool calls + if choice.message.tool_calls: + for tool_call in choice.message.tool_calls: + try: + tool_input = json.loads(tool_call.function.arguments) + except (json.JSONDecodeError, TypeError): + tool_input = {} + + content.append( + AnthropicContentBlock( + type="tool_use", + id=tool_call.id, + name=tool_call.function.name, + input=tool_input, + ) + ) + + # Map stop reason + stop_reason = STOP_REASON_MAP.get(choice.finish_reason or "stop", "end_turn") + + return AnthropicMessagesResponse( + id=f"msg_{uuid.uuid4().hex}", + content=content, + model=response.model, + stop_reason=stop_reason, + usage=AnthropicUsage( + input_tokens=response.usage.prompt_tokens if response.usage else 0, + output_tokens=response.usage.completion_tokens if response.usage else 0, + ), + ) + + def _error_response( + self, + status_code: int, + error_type: str, + message: str, + ) -> JSONResponse: + """Create an Anthropic-format error response.""" + error_resp = AnthropicErrorResponse( + error=AnthropicError(type=error_type, message=message) + ) + return JSONResponse( + status_code=status_code, + content=error_resp.model_dump(), + ) + + async def handle_count_tokens( + self, + request: AnthropicCountTokensRequest, + raw_request: Request, + ) -> JSONResponse: + """Handle /v1/messages/count_tokens endpoint. + + Converts the request to a ChatCompletionRequest, applies the chat + template via the OpenAI handler to tokenize, and returns the count. + """ + try: + # Build a minimal AnthropicMessagesRequest so we can reuse conversion + messages_request = AnthropicMessagesRequest( + model=request.model, + messages=request.messages, + max_tokens=1, # dummy, not used for counting + system=request.system, + tools=request.tools, + tool_choice=request.tool_choice, + ) + chat_request = self._convert_to_chat_completion_request(messages_request) + except Exception as e: + logger.exception("Error converting count_tokens request: %s", e) + return self._error_response( + status_code=400, + error_type="invalid_request_error", + message=str(e), + ) + + try: + is_multimodal = ( + self.openai_serving_chat.tokenizer_manager.model_config.is_multimodal + ) + processed = self.openai_serving_chat._process_messages( + chat_request, is_multimodal + ) + + if isinstance(processed.prompt_ids, list): + input_tokens = len(processed.prompt_ids) + else: + # prompt_ids is a string (multimodal case) — tokenize it + tokenizer = self.openai_serving_chat.tokenizer_manager.tokenizer + input_tokens = len(tokenizer.encode(processed.prompt_ids)) + + return JSONResponse( + content=AnthropicCountTokensResponse( + input_tokens=input_tokens + ).model_dump() + ) + except Exception as e: + logger.exception("Error counting tokens: %s", e) + return self._error_response( + status_code=500, + error_type="internal_error", + message="Internal server error", + ) diff --git a/python/sglang/srt/entrypoints/http_server.py b/python/sglang/srt/entrypoints/http_server.py index 2cc45e63b..1d6816c01 100644 --- a/python/sglang/srt/entrypoints/http_server.py +++ b/python/sglang/srt/entrypoints/http_server.py @@ -52,6 +52,11 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import ORJSONResponse, Response, StreamingResponse from sglang.srt.disaggregation.utils import FAKE_BOOTSTRAP_HOST, DisaggregationMode +from sglang.srt.entrypoints.anthropic.protocol import ( + AnthropicCountTokensRequest, + AnthropicMessagesRequest, +) +from sglang.srt.entrypoints.anthropic.serving import AnthropicServing from sglang.srt.entrypoints.engine import ( _launch_subprocesses, init_tokenizer_manager, @@ -297,6 +302,11 @@ async def lifespan(fast_api_app: FastAPI): # Initialize Ollama-compatible serving handler fast_api_app.state.ollama_serving = OllamaServing(_global_state.tokenizer_manager) + # Initialize Anthropic-compatible serving handler + fast_api_app.state.anthropic_serving = AnthropicServing( + fast_api_app.state.openai_serving_chat + ) + # Launch tool server tool_server = None if server_args.tool_server == "demo": @@ -1555,6 +1565,29 @@ async def ollama_show(request: OllamaShowRequest, raw_request: Request): return raw_request.app.state.ollama_serving.get_show(request.model) +##### Anthropic-compatible API endpoints ##### + + +@app.post("/v1/messages", dependencies=[Depends(validate_json_request)]) +async def anthropic_v1_messages( + request: AnthropicMessagesRequest, raw_request: Request +): + """Anthropic-compatible Messages API endpoint.""" + return await raw_request.app.state.anthropic_serving.handle_messages( + request, raw_request + ) + + +@app.post("/v1/messages/count_tokens", dependencies=[Depends(validate_json_request)]) +async def anthropic_v1_count_tokens( + request: AnthropicCountTokensRequest, raw_request: Request +): + """Anthropic-compatible token counting endpoint.""" + return await raw_request.app.state.anthropic_serving.handle_count_tokens( + request, raw_request + ) + + ## SageMaker API @app.get("/ping") async def sagemaker_health() -> Response: diff --git a/test/manual/vlm/test_anthropic_vision.py b/test/manual/vlm/test_anthropic_vision.py new file mode 100644 index 000000000..3cf0d3a86 --- /dev/null +++ b/test/manual/vlm/test_anthropic_vision.py @@ -0,0 +1,433 @@ +""" +Tests for Anthropic-compatible image input via the /v1/messages endpoint. + +python3 anthorpic_api/test/manual/vlm/test_anthropic_vision.py +""" + +import json +import unittest + +import pybase64 +import requests + +from sglang.srt.utils import kill_process_tree +from sglang.test.test_utils import ( + DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST, + DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + DEFAULT_URL_FOR_TEST, + CustomTestCase, + popen_launch_server, +) + +IMAGE_MAN_IRONING_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/images/man_ironing_on_back_of_suv.png" +IMAGE_SGL_LOGO_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/images/sgl_logo.png" + + +def _fetch_image_base64(url: str) -> str: + """Download an image and return its base64-encoded content.""" + resp = requests.get(url, timeout=30) + resp.raise_for_status() + return pybase64.b64encode(resp.content).decode("utf-8") + + +class TestAnthropicVision(CustomTestCase): + @classmethod + def setUpClass(cls): + cls.model = DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST + cls.base_url = DEFAULT_URL_FOR_TEST + cls.api_key = "sk-123456" + cls.process = popen_launch_server( + cls.model, + cls.base_url, + timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + api_key=cls.api_key, + other_args=[ + "--trust-remote-code", + "--enable-multimodal", + "--cuda-graph-max-bs=4", + ], + ) + cls.messages_url = cls.base_url + "/v1/messages" + # Pre-fetch the image as base64 once for all tests + cls.image_base64 = _fetch_image_base64(IMAGE_MAN_IRONING_URL) + + @classmethod + def tearDownClass(cls): + kill_process_tree(cls.process.pid) + + def _make_request(self, payload, stream=False): + """Send a request to the /v1/messages endpoint.""" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}", + } + return requests.post( + self.messages_url, + headers=headers, + json=payload, + stream=stream, + ) + + def _parse_sse_events(self, response): + """Parse SSE events from a streaming response.""" + events = [] + for line in response.iter_lines(decode_unicode=True): + if not line: + continue + if line.startswith("data: "): + data_str = line[6:].strip() + if data_str == "[DONE]": + continue + try: + events.append(json.loads(data_str)) + except json.JSONDecodeError: + pass + return events + + def _verify_ironing_image_content(self, text): + """Verify the response text describes the man-ironing-on-SUV image.""" + text_lower = text.lower() + self.assertTrue( + any(w in text_lower for w in ["man", "person", "driver", "someone"]), + f"Expected mention of a person, got: {text}", + ) + self.assertTrue( + any( + w in text_lower + for w in ["cab", "taxi", "suv", "vehicle", "car", "trunk", "back"] + ), + f"Expected mention of a vehicle, got: {text}", + ) + self.assertTrue( + any( + w in text_lower + for w in ["iron", "hang", "cloth", "holding", "laundry", "shirt"] + ), + f"Expected mention of ironing/clothes, got: {text}", + ) + + # ---- Base64 image tests ---- + + def test_single_image_base64(self): + """Test sending a single base64 image in Anthropic format.""" + payload = { + "model": self.model, + "max_tokens": 128, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": self.image_base64, + }, + }, + { + "type": "text", + "text": "Describe this image in a sentence.", + }, + ], + } + ], + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertEqual(body["role"], "assistant") + self.assertTrue(len(body["content"]) > 0) + self.assertEqual(body["content"][0]["type"], "text") + text = body["content"][0]["text"] + self.assertIsInstance(text, str) + self.assertTrue(len(text) > 0, "Response text should not be empty") + + # Verify response describes the image content + self._verify_ironing_image_content(text) + + # Verify usage + self.assertIn("usage", body) + self.assertGreater(body["usage"]["input_tokens"], 0) + self.assertGreater(body["usage"]["output_tokens"], 0) + + # Verify id format + self.assertTrue( + body["id"].startswith("msg_"), + f"ID should start with 'msg_', got: {body['id']}", + ) + + def test_single_image_url(self): + """Test sending an image via URL (converted to data URI internally).""" + # Anthropic format uses source.type="base64", but we test the data URI path + # by pre-encoding the URL image as base64 + payload = { + "model": self.model, + "max_tokens": 128, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": self.image_base64, + }, + }, + { + "type": "text", + "text": "What objects do you see in this image?", + }, + ], + } + ], + "temperature": 0, + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + text = body["content"][0]["text"] + self.assertIsInstance(text, str) + self.assertTrue(len(text) > 0) + + # Verify response describes the image content + self._verify_ironing_image_content(text) + + def test_image_with_text_blocks(self): + """Test image combined with multiple text content blocks.""" + payload = { + "model": self.model, + "max_tokens": 128, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "text", + "text": "Look at this image carefully.", + }, + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": self.image_base64, + }, + }, + { + "type": "text", + "text": "Describe what you see in one sentence.", + }, + ], + } + ], + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + self.assertEqual(body["content"][0]["type"], "text") + text = body["content"][0]["text"] + self.assertTrue(len(text) > 0) + + # Verify response describes the image content + self._verify_ironing_image_content(text) + + # ---- Streaming with image ---- + + def test_image_stream(self): + """Test streaming response with image input.""" + payload = { + "model": self.model, + "max_tokens": 128, + "stream": True, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": self.image_base64, + }, + }, + { + "type": "text", + "text": "Describe this image briefly.", + }, + ], + } + ], + } + resp = self._make_request(payload, stream=True) + self.assertEqual(resp.status_code, 200) + self.assertIn("text/event-stream", resp.headers.get("content-type", "")) + + events = self._parse_sse_events(resp) + event_types = [e["type"] for e in events] + + # Verify event sequence + self.assertIn("message_start", event_types) + self.assertIn("message_stop", event_types) + self.assertEqual(events[0]["type"], "message_start") + + # Verify we got content + content_deltas = [e for e in events if e["type"] == "content_block_delta"] + self.assertTrue(len(content_deltas) > 0, "Expected content_block_delta events") + + # Reconstruct text + full_text = "".join( + e["delta"]["text"] + for e in content_deltas + if e["delta"].get("type") == "text_delta" + ) + self.assertTrue(len(full_text) > 0, "Streamed text should not be empty") + + # Verify streamed response describes the image content + self._verify_ironing_image_content(full_text) + + # Verify message_delta has stop_reason + message_deltas = [e for e in events if e["type"] == "message_delta"] + self.assertTrue(len(message_deltas) > 0) + self.assertIn("stop_reason", message_deltas[-1]["delta"]) + + # ---- Multi-image tests ---- + + def test_multi_image(self): + """Test sending multiple images in a single message.""" + logo_base64 = _fetch_image_base64(IMAGE_SGL_LOGO_URL) + + payload = { + "model": self.model, + "max_tokens": 128, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": self.image_base64, + }, + }, + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": logo_base64, + }, + }, + { + "type": "text", + "text": "How many images do you see? Describe each briefly.", + }, + ], + } + ], + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + text = body["content"][0]["text"] + self.assertIsInstance(text, str) + self.assertTrue(len(text) > 0) + + # ---- Multi-turn with image ---- + + def test_multi_turn_with_image(self): + """Test multi-turn conversation with image context.""" + # First turn: send image + payload = { + "model": self.model, + "max_tokens": 128, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": self.image_base64, + }, + }, + { + "type": "text", + "text": "What is in this image?", + }, + ], + }, + ], + "temperature": 0, + } + resp1 = self._make_request(payload) + self.assertEqual(resp1.status_code, 200, f"Response: {resp1.text}") + body1 = resp1.json() + first_response_text = body1["content"][0]["text"] + + # Verify first turn describes the image + self._verify_ironing_image_content(first_response_text) + + # Second turn: ask follow-up without re-sending image + payload2 = { + "model": self.model, + "max_tokens": 128, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": self.image_base64, + }, + }, + { + "type": "text", + "text": "What is in this image?", + }, + ], + }, + { + "role": "assistant", + "content": first_response_text, + }, + { + "role": "user", + "content": "Can you describe the colors you see?", + }, + ], + "temperature": 0, + } + resp2 = self._make_request(payload2) + self.assertEqual(resp2.status_code, 200, f"Response: {resp2.text}") + + body2 = resp2.json() + self.assertEqual(body2["type"], "message") + self.assertTrue(len(body2["content"]) > 0) + self.assertEqual(body2["content"][0]["type"], "text") + self.assertTrue(len(body2["content"][0]["text"]) > 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/registered/openai_server/basic/test_anthropic_server.py b/test/registered/openai_server/basic/test_anthropic_server.py new file mode 100644 index 000000000..902fabccd --- /dev/null +++ b/test/registered/openai_server/basic/test_anthropic_server.py @@ -0,0 +1,493 @@ +""" +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_simple_messages +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_simple_messages_stream +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_multi_turn_messages +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_system_message_string +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_system_message_blocks +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_max_tokens +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_temperature +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_stop_sequences +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_error_invalid_max_tokens +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_error_empty_messages +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_raw_http_non_streaming +python3 -m unittest openai_server.basic.test_anthropic_server.TestAnthropicServer.test_raw_http_streaming +""" + +import json +import unittest + +import requests + +from sglang.srt.utils import kill_process_tree +from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci +from sglang.test.test_utils import ( + DEFAULT_SMALL_MODEL_NAME_FOR_TEST, + DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + DEFAULT_URL_FOR_TEST, + CustomTestCase, + popen_launch_server, +) + +register_cuda_ci(est_time=120, suite="stage-b-test-small-1-gpu") +register_amd_ci(est_time=140, suite="stage-b-test-small-1-gpu-amd") + + +class TestAnthropicServer(CustomTestCase): + @classmethod + def setUpClass(cls): + cls.model = DEFAULT_SMALL_MODEL_NAME_FOR_TEST + cls.base_url = DEFAULT_URL_FOR_TEST + cls.api_key = "sk-123456" + cls.process = popen_launch_server( + cls.model, + cls.base_url, + timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + api_key=cls.api_key, + ) + cls.messages_url = cls.base_url + "/v1/messages" + + @classmethod + def tearDownClass(cls): + kill_process_tree(cls.process.pid) + + def _make_request(self, payload, stream=False): + """Send a request to the /v1/messages endpoint.""" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}", + } + return requests.post( + self.messages_url, + headers=headers, + json=payload, + stream=stream, + ) + + def _default_payload(self, **overrides): + """Build a default Anthropic Messages request payload.""" + payload = { + "model": self.model, + "max_tokens": 64, + "messages": [ + { + "role": "user", + "content": "What is the capital of France? Answer in a few words.", + } + ], + } + payload.update(overrides) + return payload + + # ---- Non-streaming tests ---- + + def test_simple_messages(self): + """Test basic non-streaming message request.""" + payload = self._default_payload() + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertEqual(body["role"], "assistant") + self.assertIn("content", body) + self.assertIsInstance(body["content"], list) + self.assertTrue(len(body["content"]) > 0) + self.assertEqual(body["content"][0]["type"], "text") + self.assertIsInstance(body["content"][0]["text"], str) + self.assertTrue(len(body["content"][0]["text"]) > 0) + + # Verify stop reason + self.assertIn(body["stop_reason"], ["end_turn", "max_tokens", "stop_sequence"]) + + # Verify usage + self.assertIn("usage", body) + self.assertIsInstance(body["usage"]["input_tokens"], int) + self.assertIsInstance(body["usage"]["output_tokens"], int) + self.assertGreater(body["usage"]["input_tokens"], 0) + self.assertGreater(body["usage"]["output_tokens"], 0) + + # Verify id format (must be msg_*) and model + self.assertIn("id", body) + self.assertIsInstance(body["id"], str) + self.assertTrue( + body["id"].startswith("msg_"), + f"ID should start with 'msg_', got: {body['id']}", + ) + self.assertIn("model", body) + + def test_multi_turn_messages(self): + """Test multi-turn conversation.""" + payload = self._default_payload( + messages=[ + {"role": "user", "content": "My name is Alice."}, + {"role": "assistant", "content": "Hello Alice! Nice to meet you."}, + {"role": "user", "content": "What is my name?"}, + ] + ) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + self.assertEqual(body["content"][0]["type"], "text") + self.assertIsInstance(body["content"][0]["text"], str) + + def test_system_message_string(self): + """Test system message as a string.""" + payload = self._default_payload( + system="You are a helpful assistant. Always respond in French.", + ) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + + def test_system_message_blocks(self): + """Test system message as content blocks.""" + payload = self._default_payload( + system=[ + {"type": "text", "text": "You are a helpful assistant."}, + {"type": "text", "text": "Always be concise."}, + ], + ) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + + def test_max_tokens(self): + """Test max_tokens limits output length.""" + payload = self._default_payload( + max_tokens=5, + messages=[ + {"role": "user", "content": "Tell me a long story about a dragon."} + ], + ) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + # With very small max_tokens the model should hit the limit + self.assertIn(body["stop_reason"], ["max_tokens", "end_turn"]) + self.assertGreater(body["usage"]["output_tokens"], 0) + + def test_temperature(self): + """Test temperature parameter is accepted.""" + payload = self._default_payload(temperature=0.0) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + + def test_stop_sequences(self): + """Test stop_sequences parameter is accepted.""" + payload = self._default_payload( + stop_sequences=["\n"], + max_tokens=128, + ) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + + def test_top_p_and_top_k(self): + """Test top_p and top_k parameters.""" + payload = self._default_payload(top_p=0.9, top_k=40) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + + # ---- Streaming tests ---- + + def test_simple_messages_stream(self): + """Test basic streaming message request.""" + payload = self._default_payload(stream=True) + resp = self._make_request(payload, stream=True) + self.assertEqual(resp.status_code, 200, f"Status: {resp.status_code}") + + events = self._parse_sse_events(resp) + + # Verify event sequence + event_types = [e["type"] for e in events] + self.assertIn("message_start", event_types) + self.assertIn("message_stop", event_types) + + # Verify message_start + message_start = next(e for e in events if e["type"] == "message_start") + self.assertIn("message", message_start) + self.assertEqual(message_start["message"]["type"], "message") + self.assertEqual(message_start["message"]["role"], "assistant") + self.assertIn("usage", message_start["message"]) + + # Verify we got content deltas + content_deltas = [e for e in events if e["type"] == "content_block_delta"] + self.assertTrue( + len(content_deltas) > 0, "Expected at least one content_block_delta event" + ) + + # Verify all text deltas have correct structure + for delta_event in content_deltas: + self.assertIn("delta", delta_event) + self.assertEqual(delta_event["delta"]["type"], "text_delta") + self.assertIn("text", delta_event["delta"]) + + # Reconstruct the full text + full_text = "".join( + e["delta"]["text"] + for e in content_deltas + if e["delta"].get("type") == "text_delta" + ) + self.assertTrue(len(full_text) > 0, "Reconstructed text should not be empty") + + # Verify content_block_start/stop + block_starts = [e for e in events if e["type"] == "content_block_start"] + block_stops = [e for e in events if e["type"] == "content_block_stop"] + self.assertTrue(len(block_starts) > 0, "Expected content_block_start") + self.assertTrue(len(block_stops) > 0, "Expected content_block_stop") + self.assertEqual(block_starts[0]["content_block"]["type"], "text") + + # Verify message_delta with stop_reason + message_deltas = [e for e in events if e["type"] == "message_delta"] + self.assertTrue(len(message_deltas) > 0, "Expected message_delta event") + last_delta = message_deltas[-1] + self.assertIn("delta", last_delta) + self.assertIn("stop_reason", last_delta["delta"]) + self.assertIn( + last_delta["delta"]["stop_reason"], + ["end_turn", "max_tokens", "stop_sequence", "tool_use"], + ) + + # Verify usage in message_delta + self.assertIn("usage", last_delta) + self.assertIsInstance(last_delta["usage"]["output_tokens"], int) + + def test_stream_multi_turn(self): + """Test streaming with multi-turn conversation.""" + payload = self._default_payload( + stream=True, + messages=[ + {"role": "user", "content": "Say hello."}, + {"role": "assistant", "content": "Hello!"}, + {"role": "user", "content": "Say goodbye."}, + ], + ) + resp = self._make_request(payload, stream=True) + self.assertEqual(resp.status_code, 200) + + events = self._parse_sse_events(resp) + event_types = [e["type"] for e in events] + self.assertIn("message_start", event_types) + self.assertIn("message_stop", event_types) + + def test_stream_with_system(self): + """Test streaming with system message.""" + payload = self._default_payload( + stream=True, + system="You are a pirate. Respond in pirate speak.", + ) + resp = self._make_request(payload, stream=True) + self.assertEqual(resp.status_code, 200) + + events = self._parse_sse_events(resp) + event_types = [e["type"] for e in events] + self.assertIn("message_start", event_types) + self.assertIn("message_stop", event_types) + + # ---- Error handling tests ---- + + def test_error_invalid_max_tokens(self): + """Test error response for invalid max_tokens.""" + payload = self._default_payload(max_tokens=-1) + resp = self._make_request(payload) + self.assertIn(resp.status_code, [400, 422]) + + def test_error_empty_messages(self): + """Test error response for empty messages list.""" + payload = self._default_payload(messages=[]) + resp = self._make_request(payload) + self.assertIn(resp.status_code, [400, 422]) + + def test_error_missing_content_type(self): + """Test error when Content-Type is not application/json.""" + headers = { + "Authorization": f"Bearer {self.api_key}", + } + resp = requests.post( + self.messages_url, + headers=headers, + data="not json", + ) + self.assertIn(resp.status_code, [400, 415, 422]) + + # ---- Raw HTTP tests ---- + + def test_raw_http_non_streaming(self): + """Test raw HTTP request/response format for non-streaming.""" + payload = self._default_payload(temperature=0) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200) + + # Verify response content type + self.assertIn("application/json", resp.headers.get("content-type", "")) + + body = resp.json() + # Verify all required fields per Anthropic spec + required_fields = ["id", "type", "role", "content", "model", "usage"] + for field in required_fields: + self.assertIn(field, body, f"Missing required field: {field}") + + self.assertEqual(body["type"], "message") + self.assertEqual(body["role"], "assistant") + + def test_raw_http_streaming(self): + """Test raw HTTP request/response format for streaming.""" + payload = self._default_payload(stream=True, temperature=0) + resp = self._make_request(payload, stream=True) + self.assertEqual(resp.status_code, 200) + + # Verify streaming content type + self.assertIn("text/event-stream", resp.headers.get("content-type", "")) + + # Verify we get proper SSE events + events = self._parse_sse_events(resp) + self.assertTrue(len(events) > 0, "Expected at least some SSE events") + + # Verify event ordering: message_start should be first + self.assertEqual( + events[0]["type"], "message_start", "First event should be message_start" + ) + + # Verify message_stop is last data event + data_events = [e for e in events if e["type"] != "ping"] + self.assertEqual( + data_events[-1]["type"], + "message_stop", + "Last data event should be message_stop", + ) + + # ---- Content block tests ---- + + def test_content_blocks_message(self): + """Test sending messages with explicit content blocks.""" + payload = self._default_payload( + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "What is 2+2?"}, + ], + } + ], + ) + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertTrue(len(body["content"]) > 0) + self.assertEqual(body["content"][0]["type"], "text") + + # ---- Count tokens tests ---- + + def test_count_tokens(self): + """Test /v1/messages/count_tokens endpoint.""" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}", + } + payload = { + "model": self.model, + "messages": [ + {"role": "user", "content": "Hello, how are you?"}, + ], + } + resp = requests.post( + self.base_url + "/v1/messages/count_tokens", + headers=headers, + json=payload, + ) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertIn("input_tokens", body) + self.assertIsInstance(body["input_tokens"], int) + self.assertGreater(body["input_tokens"], 0) + + def test_count_tokens_with_system(self): + """Test count_tokens with system message.""" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}", + } + payload_no_system = { + "model": self.model, + "messages": [ + {"role": "user", "content": "Hello"}, + ], + } + payload_with_system = { + "model": self.model, + "messages": [ + {"role": "user", "content": "Hello"}, + ], + "system": "You are a helpful assistant with a very long system prompt that adds tokens.", + } + resp1 = requests.post( + self.base_url + "/v1/messages/count_tokens", + headers=headers, + json=payload_no_system, + ) + resp2 = requests.post( + self.base_url + "/v1/messages/count_tokens", + headers=headers, + json=payload_with_system, + ) + self.assertEqual(resp1.status_code, 200) + self.assertEqual(resp2.status_code, 200) + + # System message should increase the token count + tokens_no_system = resp1.json()["input_tokens"] + tokens_with_system = resp2.json()["input_tokens"] + self.assertGreater( + tokens_with_system, + tokens_no_system, + "Adding system message should increase token count", + ) + + # ---- Helpers ---- + + def _parse_sse_events(self, response): + """Parse SSE events from a streaming response.""" + events = [] + + for line in response.iter_lines(decode_unicode=True): + if not line: + continue + + if line.startswith("data: "): + data_str = line[6:].strip() + if data_str == "[DONE]": + continue + try: + data = json.loads(data_str) + events.append(data) + except json.JSONDecodeError: + pass + + return events + + +if __name__ == "__main__": + unittest.main() diff --git a/test/registered/openai_server/function_call/test_anthropic_tool_use.py b/test/registered/openai_server/function_call/test_anthropic_tool_use.py new file mode 100644 index 000000000..086f1b036 --- /dev/null +++ b/test/registered/openai_server/function_call/test_anthropic_tool_use.py @@ -0,0 +1,555 @@ +""" +Tests for Anthropic-compatible tool use via the /v1/messages endpoint. + +python3 -m unittest openai_server.function_call.test_anthropic_tool_use.TestAnthropicToolUse.test_tool_use_format +python3 -m unittest openai_server.function_call.test_anthropic_tool_use.TestAnthropicToolUse.test_tool_use_streaming +python3 -m unittest openai_server.function_call.test_anthropic_tool_use.TestAnthropicToolUse.test_tool_use_streaming_args_parsing +python3 -m unittest openai_server.function_call.test_anthropic_tool_use.TestAnthropicToolUse.test_tool_choice_auto +python3 -m unittest openai_server.function_call.test_anthropic_tool_use.TestAnthropicToolUse.test_tool_choice_any +python3 -m unittest openai_server.function_call.test_anthropic_tool_use.TestAnthropicToolUse.test_tool_choice_specific +python3 -m unittest openai_server.function_call.test_anthropic_tool_use.TestAnthropicToolUse.test_tool_result_multi_turn +""" + +import json +import unittest + +import requests + +from sglang.srt.utils import kill_process_tree +from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci +from sglang.test.test_utils import ( + DEFAULT_SMALL_MODEL_NAME_FOR_TEST, + DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + DEFAULT_URL_FOR_TEST, + CustomTestCase, + popen_launch_server, +) + +register_cuda_ci(est_time=120, suite="stage-b-test-large-1-gpu") +register_amd_ci(est_time=140, suite="stage-b-test-small-1-gpu-amd") + +# System message to guide Llama3.2 to produce proper tool call format +SYSTEM_MESSAGE = ( + "You are a helpful assistant with tool calling capabilities. " + "Only reply with a tool call if the function exists in the library provided by the user. " + "If it doesn't exist, just reply directly in natural language. " + "When you receive a tool call response, use the output to format an answer to the original user question. " + "You have access to the following functions. " + "To call a function, please respond with JSON for a function call. " + 'Respond in the format {"name": function name, "parameters": dictionary of argument name and its value}. ' + "Do not use variables.\n\n" +) + +ADD_TOOL = { + "name": "add", + "description": "Compute the sum of two integers", + "input_schema": { + "type": "object", + "properties": { + "a": {"type": "integer", "description": "First integer"}, + "b": {"type": "integer", "description": "Second integer"}, + }, + "required": ["a", "b"], + }, +} + +WEATHER_TOOL = { + "name": "get_current_weather", + "description": "Get the current weather in a given location", + "input_schema": { + "type": "object", + "properties": { + "city": { + "type": "string", + "description": "The city to find the weather for", + }, + "unit": { + "type": "string", + "description": "Weather unit (celsius or fahrenheit)", + "enum": ["celsius", "fahrenheit"], + }, + }, + "required": ["city", "unit"], + }, +} + + +class TestAnthropicToolUse(CustomTestCase): + @classmethod + def setUpClass(cls): + cls.model = DEFAULT_SMALL_MODEL_NAME_FOR_TEST + cls.base_url = DEFAULT_URL_FOR_TEST + cls.api_key = "sk-123456" + cls.process = popen_launch_server( + cls.model, + cls.base_url, + timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + api_key=cls.api_key, + other_args=[ + "--tool-call-parser", + "llama3", + ], + ) + cls.messages_url = cls.base_url + "/v1/messages" + + @classmethod + def tearDownClass(cls): + kill_process_tree(cls.process.pid) + + def _make_request(self, payload, stream=False): + """Send a request to the /v1/messages endpoint.""" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}", + } + return requests.post( + self.messages_url, + headers=headers, + json=payload, + stream=stream, + ) + + def _parse_sse_events(self, response): + """Parse SSE events from a streaming response.""" + events = [] + for line in response.iter_lines(decode_unicode=True): + if not line: + continue + if line.startswith("data: "): + data_str = line[6:].strip() + if data_str == "[DONE]": + continue + try: + events.append(json.loads(data_str)) + except json.JSONDecodeError: + pass + return events + + # ---- Non-streaming tool use tests ---- + + def test_tool_use_format(self): + """Test that tool use returns proper Anthropic content blocks.""" + payload = { + "model": self.model, + "max_tokens": 2048, + "system": SYSTEM_MESSAGE, + "messages": [ + {"role": "user", "content": "Compute (3+5)"}, + ], + "tools": [ADD_TOOL], + "temperature": 0.8, + "top_p": 0.8, + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + + # Find tool_use content blocks + tool_use_blocks = [b for b in body["content"] if b["type"] == "tool_use"] + self.assertTrue( + len(tool_use_blocks) > 0, + f"Expected tool_use content blocks, got: {body['content']}", + ) + + tool_block = tool_use_blocks[0] + self.assertEqual(tool_block["name"], "add", "Tool name should be 'add'") + self.assertIn("id", tool_block, "Tool use block should have an id") + self.assertIn("input", tool_block, "Tool use block should have input") + self.assertIsInstance(tool_block["input"], dict) + + # Verify stop_reason is tool_use + self.assertEqual( + body["stop_reason"], + "tool_use", + f"Expected stop_reason 'tool_use', got: {body['stop_reason']}", + ) + + def test_tool_choice_auto(self): + """Test tool_choice type=auto (default when tools provided).""" + payload = { + "model": self.model, + "max_tokens": 2048, + "system": SYSTEM_MESSAGE, + "messages": [ + {"role": "user", "content": "Compute (3+5)"}, + ], + "tools": [ADD_TOOL], + "tool_choice": {"type": "auto"}, + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + # With auto, model may or may not use tools - just verify valid response + self.assertIsInstance(body["content"], list) + + def test_tool_choice_any(self): + """Test tool_choice type=any (maps to required).""" + payload = { + "model": self.model, + "max_tokens": 2048, + "system": SYSTEM_MESSAGE, + "messages": [ + { + "role": "user", + "content": "What is the weather in Paris in celsius?", + }, + ], + "tools": [WEATHER_TOOL], + "tool_choice": {"type": "any"}, + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + + # With 'any', the model must use a tool + tool_use_blocks = [b for b in body["content"] if b["type"] == "tool_use"] + self.assertTrue( + len(tool_use_blocks) > 0, + f"Expected tool_use blocks with tool_choice=any, got: {body['content']}", + ) + + def test_tool_choice_specific(self): + """Test tool_choice type=tool with specific tool name.""" + payload = { + "model": self.model, + "max_tokens": 2048, + "system": SYSTEM_MESSAGE, + "messages": [ + {"role": "user", "content": "What is the capital of France?"}, + ], + "tools": [ADD_TOOL, WEATHER_TOOL], + "tool_choice": {"type": "tool", "name": "get_current_weather"}, + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + + # With specific tool choice, the model should call that specific tool + tool_use_blocks = [b for b in body["content"] if b["type"] == "tool_use"] + self.assertTrue( + len(tool_use_blocks) > 0, + f"Expected tool_use blocks with specific tool_choice, got: {body['content']}", + ) + for block in tool_use_blocks: + self.assertEqual( + block["name"], + "get_current_weather", + f"Expected tool name 'get_current_weather', got: {block['name']}", + ) + + def test_tool_result_multi_turn(self): + """Test multi-turn conversation with tool_result messages.""" + # First turn: request a tool call + payload_1 = { + "model": self.model, + "max_tokens": 2048, + "system": SYSTEM_MESSAGE, + "messages": [ + {"role": "user", "content": "Compute (3+5)"}, + ], + "tools": [ADD_TOOL], + "temperature": 0.8, + } + resp_1 = self._make_request(payload_1) + self.assertEqual(resp_1.status_code, 200, f"Response: {resp_1.text}") + body_1 = resp_1.json() + + # Extract tool call info + tool_use_blocks = [b for b in body_1["content"] if b["type"] == "tool_use"] + self.assertTrue(len(tool_use_blocks) > 0, "Expected tool_use in first response") + tool_call_id = tool_use_blocks[0]["id"] + + # Second turn: send tool_result back + payload_2 = { + "model": self.model, + "max_tokens": 2048, + "system": SYSTEM_MESSAGE, + "messages": [ + {"role": "user", "content": "Compute (3+5)"}, + { + "role": "assistant", + "content": body_1["content"], + }, + { + "role": "user", + "content": [ + { + "type": "tool_result", + "id": tool_call_id, + "content": "8", + } + ], + }, + ], + "tools": [ADD_TOOL], + } + resp_2 = self._make_request(payload_2) + self.assertEqual(resp_2.status_code, 200, f"Response: {resp_2.text}") + + body_2 = resp_2.json() + self.assertEqual(body_2["type"], "message") + self.assertTrue( + len(body_2["content"]) > 0, "Second response should have content" + ) + + def test_tool_use_with_text_content(self): + """Test that response can contain both text and tool_use blocks.""" + payload = { + "model": self.model, + "max_tokens": 2048, + "system": SYSTEM_MESSAGE, + "messages": [ + {"role": "user", "content": "Compute (3+5)"}, + ], + "tools": [ADD_TOOL], + "tool_choice": {"type": "auto"}, + "temperature": 0.8, + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + self.assertEqual(body["type"], "message") + self.assertIsInstance(body["content"], list) + # Verify that content has valid block types + for block in body["content"]: + self.assertIn( + block["type"], + ["text", "tool_use"], + f"Unexpected content block type: {block['type']}", + ) + + # ---- Streaming tool use tests ---- + + def test_tool_use_streaming(self): + """Test streaming tool use returns proper Anthropic events.""" + payload = { + "model": self.model, + "max_tokens": 2048, + "stream": True, + "system": SYSTEM_MESSAGE, + "messages": [ + { + "role": "user", + "content": "What is the temperature in Paris in celsius?", + }, + ], + "tools": [WEATHER_TOOL], + "tool_choice": {"type": "any"}, + } + resp = self._make_request(payload, stream=True) + self.assertEqual(resp.status_code, 200) + + events = self._parse_sse_events(resp) + event_types = [e["type"] for e in events] + + # Verify basic event sequence + self.assertIn("message_start", event_types) + self.assertIn("message_stop", event_types) + + # Check for tool use content block events + block_starts = [e for e in events if e["type"] == "content_block_start"] + tool_use_starts = [ + e + for e in block_starts + if e.get("content_block", {}).get("type") == "tool_use" + ] + + self.assertTrue( + len(tool_use_starts) > 0, + "Expected tool_use content_block_start events with tool_choice=any", + ) + + # Verify tool_use content_block_start has proper structure + tool_start = tool_use_starts[0] + self.assertIn("content_block", tool_start) + self.assertEqual(tool_start["content_block"]["type"], "tool_use") + self.assertIn("id", tool_start["content_block"]) + self.assertIn("name", tool_start["content_block"]) + + # Check for input_json_delta events + input_deltas = [ + e + for e in events + if e["type"] == "content_block_delta" + and e.get("delta", {}).get("type") == "input_json_delta" + ] + # Tool calls should have at least some argument deltas + self.assertTrue( + len(input_deltas) > 0, + "Expected input_json_delta events for tool call", + ) + + # Verify message_delta has stop_reason=tool_use + message_deltas = [e for e in events if e["type"] == "message_delta"] + self.assertTrue(len(message_deltas) > 0) + self.assertEqual( + message_deltas[-1]["delta"]["stop_reason"], + "tool_use", + "Expected stop_reason 'tool_use' in streaming", + ) + + def test_tool_use_streaming_args_parsing(self): + """Test that streaming tool call arguments can be concatenated into valid JSON.""" + payload = { + "model": self.model, + "max_tokens": 2048, + "stream": True, + "system": SYSTEM_MESSAGE, + "messages": [ + { + "role": "user", + "content": "Please sum 5 and 7, just call the function.", + }, + ], + "tools": [ + { + "name": "add", + "description": "Compute the sum of two integers", + "input_schema": { + "type": "object", + "properties": { + "a": {"type": "integer", "description": "First integer"}, + "b": {"type": "integer", "description": "Second integer"}, + }, + "required": ["a", "b"], + }, + } + ], + } + resp = self._make_request(payload, stream=True) + self.assertEqual(resp.status_code, 200) + + events = self._parse_sse_events(resp) + + # Collect tool call data from stream + tool_name = None + argument_fragments = [] + + for event in events: + if event["type"] == "content_block_start": + cb = event.get("content_block", {}) + if cb.get("type") == "tool_use": + tool_name = cb.get("name") + elif event["type"] == "content_block_delta": + delta = event.get("delta", {}) + if delta.get("type") == "input_json_delta": + partial = delta.get("partial_json", "") + if partial: + argument_fragments.append(partial) + + if tool_name is not None: + # If we got a tool call, verify arguments are valid JSON + self.assertEqual(tool_name, "add", "Tool name should be 'add'") + joined_args = "".join(argument_fragments) + self.assertTrue( + len(joined_args) > 0, + "No argument fragments returned for tool call", + ) + + try: + args_obj = json.loads(joined_args) + except json.JSONDecodeError: + self.fail( + f"Concatenated tool call arguments are not valid JSON: {joined_args}" + ) + + self.assertIn("a", args_obj, "Missing parameter 'a'") + self.assertIn("b", args_obj, "Missing parameter 'b'") + + def test_tool_use_streaming_event_sequence(self): + """Test that streaming tool use events follow the correct order.""" + payload = { + "model": self.model, + "max_tokens": 2048, + "stream": True, + "system": SYSTEM_MESSAGE, + "messages": [ + {"role": "user", "content": "Compute (3+5)"}, + ], + "tools": [ADD_TOOL], + "tool_choice": {"type": "any"}, + } + resp = self._make_request(payload, stream=True) + self.assertEqual(resp.status_code, 200) + + events = self._parse_sse_events(resp) + event_types = [e["type"] for e in events] + + # message_start must be first + self.assertEqual( + event_types[0], + "message_start", + "First event should be message_start", + ) + + # message_stop must be last + self.assertEqual( + event_types[-1], + "message_stop", + "Last event should be message_stop", + ) + + # message_delta should come before message_stop + self.assertIn("message_delta", event_types) + delta_idx = event_types.index("message_delta") + stop_idx = event_types.index("message_stop") + self.assertLess( + delta_idx, stop_idx, "message_delta should come before message_stop" + ) + + # For each content block, start should come before stop + block_start_indices = [ + i for i, t in enumerate(event_types) if t == "content_block_start" + ] + block_stop_indices = [ + i for i, t in enumerate(event_types) if t == "content_block_stop" + ] + self.assertEqual( + len(block_start_indices), + len(block_stop_indices), + "Number of content_block_start should equal content_block_stop", + ) + for start_i, stop_i in zip(block_start_indices, block_stop_indices): + self.assertLess( + start_i, + stop_i, + "content_block_start should come before content_block_stop", + ) + + def test_no_tools_no_tool_use(self): + """Test that without tools, no tool_use blocks appear.""" + payload = { + "model": self.model, + "max_tokens": 64, + "messages": [ + {"role": "user", "content": "What is the capital of France?"}, + ], + } + resp = self._make_request(payload) + self.assertEqual(resp.status_code, 200, f"Response: {resp.text}") + + body = resp.json() + tool_use_blocks = [b for b in body["content"] if b["type"] == "tool_use"] + self.assertEqual( + len(tool_use_blocks), + 0, + "Should not have tool_use blocks when no tools provided", + ) + self.assertIn( + body["stop_reason"], + ["end_turn", "max_tokens"], + "Stop reason should be end_turn or max_tokens without tools", + ) + + +if __name__ == "__main__": + unittest.main()