From b3270264e4d2417bf051f5660015071aeda40120 Mon Sep 17 00:00:00 2001 From: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com> Date: Fri, 12 Jun 2026 22:46:57 +0100 Subject: [PATCH] Fix Anthropic Messages API compatibility (#25876) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Jairo David Campaña Rosero Co-authored-by: Karan Bansal <3264937+karanb192@users.noreply.github.com> Co-authored-by: eason <85663565+mango766@users.noreply.github.com> Co-authored-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Co-authored-by: qingchanghan <17794466+qingchanghan@users.noreply.github.com> Co-authored-by: Claude Opus 4.7 (1M context) Co-authored-by: Ajay Anubolu <124525760+AjAnubolu@users.noreply.github.com> Co-authored-by: Ravitez Dondeti <13931987+dondetir@users.noreply.github.com> Co-authored-by: Ratish P <114130421+Ratish1@users.noreply.github.com> Co-authored-by: Xiaoshuai Zhang <15795935+jetd1@users.noreply.github.com> Co-authored-by: Ricardo-M-L <69202550+Ricardo-M-L@users.noreply.github.com> Co-authored-by: Xinyuan Tong --- .../srt/entrypoints/anthropic/protocol.py | 473 +++++- .../srt/entrypoints/anthropic/serving.py | 1055 ++++++++++--- python/sglang/srt/entrypoints/http_server.py | 118 +- .../srt/entrypoints/openai/serving_chat.py | 105 ++ .../entrypoints/anthropic/test_serving.py | 1339 +++++++++++++++++ 5 files changed, 2791 insertions(+), 299 deletions(-) create mode 100644 test/registered/unit/entrypoints/anthropic/test_serving.py diff --git a/python/sglang/srt/entrypoints/anthropic/protocol.py b/python/sglang/srt/entrypoints/anthropic/protocol.py index 22a30bb4f..25144c102 100644 --- a/python/sglang/srt/entrypoints/anthropic/protocol.py +++ b/python/sglang/srt/entrypoints/anthropic/protocol.py @@ -1,71 +1,140 @@ -"""Pydantic models for Anthropic Messages API protocol""" +"""Pydantic models for Anthropic Messages API protocol. + +Mirrors the shape of the official Anthropic Python SDK +(``anthropic-sdk-python``): ``ContentBlock``, ``Tool``, ``MessageStreamEvent`` +and ``ContentBlockDelta`` are discriminated unions over the ``type`` field, +so each variant carries only the fields it actually uses. +""" import uuid -from typing import Any, Literal, Optional +from typing import Annotated, Any, Literal, Optional, Union -from pydantic import BaseModel, Field, field_validator +from pydantic import ( + BaseModel, + Discriminator, + Field, + NonNegativeInt, + Tag, + field_validator, + model_validator, +) class AnthropicError(BaseModel): - """Error structure for Anthropic API""" + """Error structure for Anthropic API.""" type: str message: str class AnthropicErrorResponse(BaseModel): - """Error response structure for Anthropic API""" + """Error response structure for Anthropic API.""" type: Literal["error"] = "error" error: AnthropicError class AnthropicUsage(BaseModel): - """Token usage information""" + """Token usage information. - input_tokens: int - output_tokens: int - cache_creation_input_tokens: Optional[int] = None - cache_read_input_tokens: Optional[int] = None + ``input_tokens``/``output_tokens`` are ``Optional`` because Anthropic's + streaming ``message_delta`` event omits ``input_tokens`` (the spec + requires it only on ``message_start``). Non-streaming responses set both. + """ + + input_tokens: Optional[NonNegativeInt] = None + output_tokens: Optional[NonNegativeInt] = None + cache_creation_input_tokens: Optional[NonNegativeInt] = None + cache_read_input_tokens: Optional[NonNegativeInt] = None -class AnthropicContentBlock(BaseModel): - """Content block in message""" +# ---------- Content blocks (discriminated by ``type``) ---------- - type: Literal[ - "text", - "image", - "tool_use", - "tool_result", - "tool_reference", - "thinking", - "redacted_thinking", - ] - text: Optional[str] = None - # For image content - source: Optional[dict[str, Any]] = None - # For tool use/result - id: Optional[str] = None + +class TextBlock(BaseModel): + type: Literal["text"] = "text" + text: str + + +class ImageBlock(BaseModel): + type: Literal["image"] = "image" + # Kept loosely typed for compat with both base64 and URL sources; the + # serving layer normalises to OpenAI ``image_url`` parts. + source: Optional[Union[dict[str, Any], str]] = None + + +class ToolUseBlock(BaseModel): + type: Literal["tool_use"] = "tool_use" + id: str + name: str + input: dict[str, Any] = Field(default_factory=dict) + + +class ToolResultBlock(BaseModel): + type: Literal["tool_result"] = "tool_result" tool_use_id: Optional[str] = None - name: Optional[str] = None - input: Optional[dict[str, Any]] = None - content: Optional[str | list[dict[str, Any]]] = None + # Some legacy payloads use ``id`` instead of ``tool_use_id``. + id: Optional[str] = None + content: Optional[Union[str, list["AnthropicContentBlock"]]] = None is_error: Optional[bool] = None - # For thinking content - thinking: Optional[str] = None + + +class ToolReferenceBlock(BaseModel): + """sglang extension: references a deferred-loaded tool by name.""" + + type: Literal["tool_reference"] = "tool_reference" + name: Optional[str] = None + # Anthropic-style payloads sometimes use ``tool_name``; accept both. + tool_name: Optional[str] = None + id: Optional[str] = None + + +class SearchResultBlock(BaseModel): + type: Literal["search_result"] = "search_result" + # ``source`` here is a URL/identifier string (unlike ImageBlock.source). + source: Optional[Union[str, dict[str, Any]]] = None + title: Optional[str] = None + content: Optional[list[dict[str, Any]]] = None + + +class ThinkingBlock(BaseModel): + type: Literal["thinking"] = "thinking" + thinking: str signature: Optional[str] = None +class RedactedThinkingBlock(BaseModel): + type: Literal["redacted_thinking"] = "redacted_thinking" + data: Optional[str] = None + + +AnthropicContentBlock = Annotated[ + Union[ + TextBlock, + ImageBlock, + ToolUseBlock, + ToolResultBlock, + ToolReferenceBlock, + SearchResultBlock, + ThinkingBlock, + RedactedThinkingBlock, + ], + Field(discriminator="type"), +] + + class AnthropicMessage(BaseModel): - """Message structure""" - role: Literal["user", "assistant"] - content: str | list[AnthropicContentBlock] + content: Union[str, list[AnthropicContentBlock]] -class AnthropicTool(BaseModel): - """Tool definition""" +# ---------- Tools (discriminated by ``type`` family) ---------- + +class AnthropicCustomTool(BaseModel): + """Custom tool defined by the API user — requires ``input_schema``.""" + + type: Optional[Literal["custom"]] = None # absent or explicit "custom" name: str description: Optional[str] = None input_schema: dict[str, Any] @@ -73,7 +142,7 @@ class AnthropicTool(BaseModel): @field_validator("input_schema") @classmethod - def validate_input_schema(cls, v): + def _ensure_object_schema(cls, v): if not isinstance(v, dict): raise ValueError("input_schema must be a dictionary") if "type" not in v: @@ -81,31 +150,216 @@ class AnthropicTool(BaseModel): return v +class AnthropicWebSearchTool(BaseModel): + """Anthropic ``web_search_*`` server tool family. + + No client-side ``input_schema`` — Anthropic provides the backing + search implementation. Tag format is ``web_search_YYYYMMDD``. + """ + + type: str = Field(pattern=r"^web_search_\d{8}$") + name: Literal["web_search"] = "web_search" + description: Optional[str] = None + defer_loading: Optional[bool] = None + max_uses: Optional[int] = None + allowed_domains: Optional[list[str]] = None + blocked_domains: Optional[list[str]] = None + + +class AnthropicComputerTool(BaseModel): + """Anthropic ``computer_*`` server tool family.""" + + type: str = Field(pattern=r"^computer_\d{8}$") + name: Literal["computer"] = "computer" + description: Optional[str] = None + defer_loading: Optional[bool] = None + display_width_px: Optional[int] = None + display_height_px: Optional[int] = None + display_number: Optional[int] = None + + +class AnthropicBashTool(BaseModel): + """Anthropic ``bash_*`` server tool family.""" + + type: str = Field(pattern=r"^bash_\d{8}$") + name: Literal["bash"] = "bash" + description: Optional[str] = None + defer_loading: Optional[bool] = None + + +class AnthropicTextEditorTool(BaseModel): + """Anthropic ``text_editor_*`` server tool family.""" + + type: str = Field(pattern=r"^text_editor_\d{8}$") + name: Literal["str_replace_editor", "str_replace_based_edit_tool"] + description: Optional[str] = None + defer_loading: Optional[bool] = None + + +def _tool_discriminator(v) -> str: + """Pick the right tool variant from a dict or model instance. + + Pydantic discriminators don't accept ``None`` as a tag, and custom + tools allow ``type`` to be absent. Map missing/``custom`` to + ``"custom"`` and prefix-match server-tool families. + """ + if isinstance(v, dict): + t = v.get("type") + else: + t = getattr(v, "type", None) + if not t or t == "custom": + return "custom" + if t.startswith("web_search_"): + return "web_search" + if t.startswith("computer_"): + return "computer" + if t.startswith("bash_"): + return "bash" + if t.startswith("text_editor_"): + return "text_editor" + return "custom" + + +AnthropicTool = Annotated[ + Union[ + Annotated[AnthropicCustomTool, Tag("custom")], + Annotated[AnthropicWebSearchTool, Tag("web_search")], + Annotated[AnthropicComputerTool, Tag("computer")], + Annotated[AnthropicBashTool, Tag("bash")], + Annotated[AnthropicTextEditorTool, Tag("text_editor")], + ], + Discriminator(_tool_discriminator), +] + + +def is_server_tool(tool) -> bool: + """Return True for Anthropic built-in server-side tools.""" + return isinstance( + tool, + ( + AnthropicWebSearchTool, + AnthropicComputerTool, + AnthropicBashTool, + AnthropicTextEditorTool, + ), + ) + + class AnthropicToolChoice(BaseModel): - """Tool Choice definition""" + """Tool choice strategy.""" type: Literal["auto", "any", "tool", "none"] name: Optional[str] = None +class AnthropicThinkingParam(BaseModel): + """Anthropic extended-thinking control on the request. + + Mirrors the Anthropic SDK's ``ThinkingConfigParam`` discriminated + union of three variants — see ``anthropic-sdk-python``'s + ``thinking_config_{enabled,disabled,adaptive}_param.py``: + + * ``enabled`` requires ``budget_tokens`` (≥1024) and accepts + ``display``. + * ``disabled`` accepts no other fields. + * ``adaptive`` (Claude 4.7) accepts ``display`` but not + ``budget_tokens``. + + The serving layer treats ``adaptive`` identically to ``enabled`` + because the local OpenAI-compatible backend has no auto-throttle + equivalent. ``budget_tokens`` is accepted on ``enabled`` for SDK + compatibility but the backend has no hard-cap knob to honor it; the + serving layer logs a WARNING so operators see that the requested + budget is not enforced. ``display="omitted"`` is accepted but + similarly cannot suppress reasoning mid-stream and is logged. + """ + + type: Literal["enabled", "disabled", "adaptive"] + budget_tokens: Optional[int] = None + display: Optional[Literal["summarized", "omitted"]] = None + + @model_validator(mode="after") + def _validate_thinking_shape(self): + # Cross-field rules mirror the SDK's three discriminated variants. + if self.type == "enabled": + if self.budget_tokens is None: + raise ValueError( + "thinking.budget_tokens is required when " + "thinking.type is 'enabled'" + ) + if self.budget_tokens < 1024: + raise ValueError( + "thinking.budget_tokens must be >= 1024 " + "(got {})".format(self.budget_tokens) + ) + elif self.type == "disabled": + if self.budget_tokens is not None: + raise ValueError( + "thinking.budget_tokens is not allowed when " + "thinking.type is 'disabled'" + ) + if self.display is not None: + raise ValueError( + "thinking.display is not allowed when " + "thinking.type is 'disabled'" + ) + elif self.type == "adaptive": + if self.budget_tokens is not None: + raise ValueError( + "thinking.budget_tokens is not allowed when " + "thinking.type is 'adaptive'" + ) + return self + + +class AnthropicTaskBudget(BaseModel): + """Claude 4.7 ``output_config.task_budget`` — soft hint, not a hard cap. + + Mirrors ``BetaTokenTaskBudgetParam`` in the Anthropic SDK: ``total`` + and ``type`` are required; ``remaining`` is the client-tracked + countdown used for compaction. The hard cap on generation is still + ``max_tokens``; we never enforce ``task_budget`` ourselves. + """ + + type: Literal["tokens"] + total: int = Field(gt=0) + remaining: Optional[int] = Field(default=None, ge=0) + + +class AnthropicOutputConfig(BaseModel): + """Claude 4.7 ``output_config`` block. + + ``effort`` maps to the OpenAI ``reasoning_effort`` knob (``xhigh`` → + ``max`` because the OpenAI Literal does not include ``xhigh``). + ``task_budget`` is propagated as a custom-param hint. + """ + + effort: Optional[Literal["low", "medium", "high", "xhigh", "max"]] = None + task_budget: Optional[AnthropicTaskBudget] = None + + class AnthropicCountTokensRequest(BaseModel): - """Anthropic Count Tokens API request""" + """Anthropic count_tokens API request.""" model: str messages: list[AnthropicMessage] - system: Optional[str | list[AnthropicContentBlock]] = None + system: Optional[Union[str, list[AnthropicContentBlock]]] = None + thinking: Optional[AnthropicThinkingParam] = None tool_choice: Optional[AnthropicToolChoice] = None tools: Optional[list[AnthropicTool]] = None + # Claude 4.7 / SDK-compatibility fields. Accepted but no-op on count. + output_config: Optional[AnthropicOutputConfig] = None + betas: Optional[list[str]] = None class AnthropicCountTokensResponse(BaseModel): - """Anthropic Count Tokens API response""" + """Anthropic count_tokens API response.""" input_tokens: int class AnthropicMessagesRequest(BaseModel): - """Anthropic Messages API request""" + """Anthropic Messages API request.""" model: str messages: list[AnthropicMessage] @@ -113,65 +367,139 @@ class AnthropicMessagesRequest(BaseModel): metadata: Optional[dict[str, Any]] = None stop_sequences: Optional[list[str]] = None stream: Optional[bool] = False - system: Optional[str | list[AnthropicContentBlock]] = None + system: Optional[Union[str, list[AnthropicContentBlock]]] = None temperature: Optional[float] = None + thinking: Optional[AnthropicThinkingParam] = None tool_choice: Optional[AnthropicToolChoice] = None tools: Optional[list[AnthropicTool]] = None top_k: Optional[int] = None top_p: Optional[float] = None + # Claude 4.7 fields. The Anthropic SDK / Claude Code attach these even + # when targeting non-Anthropic backends, so the schema must accept them. + output_config: Optional[AnthropicOutputConfig] = None + betas: Optional[list[str]] = None @field_validator("model") @classmethod - def validate_model(cls, v): + 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): + 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""" +# ---------- Stream deltas ---------- +# Content-block deltas (discriminated by ``type``) vs message-end delta +# (separate model; the wire format does not put ``type`` inside its payload). - type: Optional[Literal["text_delta", "input_json_delta"]] = None - text: Optional[str] = None - partial_json: Optional[str] = None - # Message delta fields +class TextDelta(BaseModel): + type: Literal["text_delta"] = "text_delta" + text: str + + +class InputJsonDelta(BaseModel): + type: Literal["input_json_delta"] = "input_json_delta" + partial_json: str + + +class ThinkingDelta(BaseModel): + type: Literal["thinking_delta"] = "thinking_delta" + thinking: str + + +class SignatureDelta(BaseModel): + type: Literal["signature_delta"] = "signature_delta" + signature: str + + +AnthropicContentDelta = Annotated[ + Union[TextDelta, InputJsonDelta, ThinkingDelta, SignatureDelta], + Field(discriminator="type"), +] + + +class AnthropicMessageEndDelta(BaseModel): + """Delta carried on ``message_delta`` events. + + Anthropic's protocol does NOT put a ``type`` field inside this delta + payload — the SSE ``event:`` header already says ``message_delta``. + Stop reason and stop sequence are the only fields. + """ + stop_reason: Optional[ Literal["end_turn", "max_tokens", "stop_sequence", "tool_use"] ] = None stop_sequence: Optional[str] = None -class AnthropicStreamEvent(BaseModel): - """Streaming event""" +# ---------- Stream events (discriminated by ``type``) ---------- - 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 MessageStartEvent(BaseModel): + type: Literal["message_start"] = "message_start" + message: "AnthropicMessagesResponse" + + +class MessageDeltaEvent(BaseModel): + type: Literal["message_delta"] = "message_delta" + delta: AnthropicMessageEndDelta + usage: AnthropicUsage + + +class MessageStopEvent(BaseModel): + type: Literal["message_stop"] = "message_stop" + + +class ContentBlockStartEvent(BaseModel): + type: Literal["content_block_start"] = "content_block_start" + index: int + content_block: AnthropicContentBlock + + +class ContentBlockDeltaEvent(BaseModel): + type: Literal["content_block_delta"] = "content_block_delta" + index: int + delta: AnthropicContentDelta + + +class ContentBlockStopEvent(BaseModel): + type: Literal["content_block_stop"] = "content_block_stop" + index: int + + +class PingEvent(BaseModel): + type: Literal["ping"] = "ping" + + +class ErrorEvent(BaseModel): + type: Literal["error"] = "error" + error: AnthropicError + + +AnthropicStreamEvent = Annotated[ + Union[ + MessageStartEvent, + MessageDeltaEvent, + MessageStopEvent, + ContentBlockStartEvent, + ContentBlockDeltaEvent, + ContentBlockStopEvent, + PingEvent, + ErrorEvent, + ], + Field(discriminator="type"), +] class AnthropicMessagesResponse(BaseModel): - """Anthropic Messages API response""" + """Anthropic Messages API response.""" id: str = Field(default_factory=lambda: f"msg_{uuid.uuid4().hex}") type: Literal["message"] = "message" @@ -183,3 +511,8 @@ class AnthropicMessagesResponse(BaseModel): ] = None stop_sequence: Optional[str] = None usage: Optional[AnthropicUsage] = None + + +# Resolve forward references for nested types. +ToolResultBlock.model_rebuild() +MessageStartEvent.model_rebuild() diff --git a/python/sglang/srt/entrypoints/anthropic/serving.py b/python/sglang/srt/entrypoints/anthropic/serving.py index 77085a2b2..392e09ae2 100644 --- a/python/sglang/srt/entrypoints/anthropic/serving.py +++ b/python/sglang/srt/entrypoints/anthropic/serving.py @@ -6,26 +6,42 @@ OpenAIServingChat for processing, and converts responses back to Anthropic forma from __future__ import annotations +import asyncio import json import logging -import time import uuid -from typing import TYPE_CHECKING, AsyncGenerator, Optional, Union +from typing import TYPE_CHECKING, Any, AsyncGenerator, Optional, Union from fastapi import Request from fastapi.responses import JSONResponse, StreamingResponse +from pydantic import BaseModel, ValidationError from sglang.srt.entrypoints.anthropic.protocol import ( AnthropicContentBlock, AnthropicCountTokensRequest, AnthropicCountTokensResponse, - AnthropicDelta, AnthropicError, AnthropicErrorResponse, + AnthropicMessageEndDelta, AnthropicMessagesRequest, AnthropicMessagesResponse, AnthropicStreamEvent, AnthropicUsage, + ContentBlockDeltaEvent, + ContentBlockStartEvent, + ContentBlockStopEvent, + ErrorEvent, + InputJsonDelta, + MessageDeltaEvent, + MessageStartEvent, + MessageStopEvent, + SignatureDelta, + TextBlock, + TextDelta, + ThinkingBlock, + ThinkingDelta, + ToolUseBlock, + is_server_tool, ) from sglang.srt.entrypoints.openai.protocol import ( ChatCompletionRequest, @@ -43,19 +59,107 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -# Map OpenAI finish reasons to Anthropic stop reasons +# Map OpenAI finish reasons to Anthropic stop reasons. Only the four +# values in ``AnthropicMessagesResponse.stop_reason``'s Literal are valid +# on the wire; ``content_filter`` and ``abort`` have no perfect mapping +# so they fall through to the ``end_turn`` default with a WARNING at the +# call site so operators don't lose the safety/abort signal in logs. STOP_REASON_MAP = { "stop": "end_turn", "length": "max_tokens", "tool_calls": "tool_use", } +ERROR_TYPE_MAP = { + 400: "invalid_request_error", + 401: "authentication_error", + 403: "permission_error", + 404: "not_found_error", + 408: "request_timeout_error", + 429: "rate_limit_error", + 500: "api_error", + 502: "api_error", + 503: "overloaded_error", + 504: "api_error", +} + + +def _cached_prompt_tokens(usage) -> int: + prompt_tokens_details = getattr(usage, "prompt_tokens_details", None) + return getattr(prompt_tokens_details, "cached_tokens", 0) or 0 + + +def _anthropic_input_tokens(usage) -> int: + prompt = getattr(usage, "prompt_tokens", 0) or 0 + cached = _cached_prompt_tokens(usage) + if cached > prompt: + # Upstream telemetry bug: cached cannot exceed the prompt it caches. + # Clamping silently here would hide the discrepancy from billing + # dashboards, so make it visible at WARNING level. + logger.warning( + "Cached tokens (%d) exceed prompt tokens (%d); clamping " + "input_tokens to 0. This usually indicates an upstream " + "telemetry bug.", + cached, + prompt, + ) + return max(prompt - cached, 0) + + +def _anthropic_usage_from_openai( + usage, + *, + include_input: bool, + include_output: bool, + force_zero_output: bool = False, +) -> AnthropicUsage: + if usage is None: + return AnthropicUsage( + input_tokens=0 if include_input else None, + output_tokens=0 if include_output else None, + ) + + usage_fields: dict[str, int] = {} + cached_tokens = _cached_prompt_tokens(usage) + if include_input: + usage_fields["input_tokens"] = _anthropic_input_tokens(usage) + if cached_tokens: + usage_fields["cache_read_input_tokens"] = cached_tokens + if include_output: + usage_fields["output_tokens"] = ( + 0 if force_zero_output else (getattr(usage, "completion_tokens", 0) or 0) + ) + return AnthropicUsage(**usage_fields) + 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" +def _scrub_error_message(message: str, status_code: int) -> str: + """Return a safe outward-facing error message. + + 5xx is always generic — never echo upstream ``str(e)`` payloads, which + may contain stack frames, file paths, or PII. 4xx keeps the original + message (truncated and with obvious traceback lines stripped) so + callers see the real validation failure. + """ + if status_code >= 500: + return "Internal server error" + if not message: + return "Request failed" + safe_lines = [ + ln + for ln in message.splitlines() + if not ln.startswith("Traceback") and 'File "/' not in ln + ] + cleaned = "\n".join(safe_lines).strip() + if len(cleaned) > 500: + cleaned = cleaned[:500] + "…" + return cleaned or "Request failed" + + class AnthropicServing: """Handler for Anthropic Messages API requests. @@ -74,6 +178,8 @@ class AnthropicServing: """Main entry point for /v1/messages endpoint.""" try: chat_request = self._convert_to_chat_completion_request(request) + except asyncio.CancelledError: + raise except Exception as e: logger.exception("Error converting Anthropic request: %s", e) return self._error_response( @@ -94,8 +200,12 @@ class AnthropicServing: openai_messages = [] def _convert_anthropic_image_source_to_openai_part( - source: Optional[dict], + source: Any, ) -> Optional[dict]: + # Source may arrive as a Pydantic model (typed ImageBlock.source) + # or as a raw dict when parsed from a nested tool_result payload. + if isinstance(source, BaseModel): + source = source.model_dump(exclude_none=True) if not isinstance(source, dict): return None @@ -123,15 +233,51 @@ class AnthropicServing: return None + def _text_from_search_result(item: dict[str, Any]) -> str: + search_parts = [] + title = item.get("title") + if title: + search_parts.append(f"Title: {title}") + + source = item.get("source") + if isinstance(source, dict): + source_text = source.get("url") or source.get("text") + if source_text: + search_parts.append(f"Source: {source_text}") + elif source: + search_parts.append(f"Source: {source}") + + content = item.get("content") + content_parts = [] + if isinstance(content, str): + content_parts.append(content) + elif isinstance(content, list): + for part in content: + if not isinstance(part, dict): + continue + if part.get("type") == "text" and part.get("text"): + content_parts.append(part["text"]) + if content_parts: + search_parts.append("Content: " + "\n".join(content_parts)) + + return "\n".join(search_parts) + def _convert_tool_result_content( - content: Optional[str | list[dict]], - ) -> tuple[str | list[dict], str]: + content: Any, + ) -> tuple[Union[str, list[dict]], str]: if isinstance(content, list): tool_content_parts = [] tool_text_parts = [] - for item in content: - if not isinstance(item, dict): + for raw_item in content: + # Items may be typed Pydantic blocks (after request + # validation) or raw dicts (from legacy callers). Coerce + # to dict so the existing key-based logic still works. + if isinstance(raw_item, BaseModel): + item = raw_item.model_dump(exclude_none=True) + elif isinstance(raw_item, dict): + item = raw_item + else: continue item_type = item.get("type") @@ -154,6 +300,13 @@ class AnthropicServing: tool_content_parts.append( {"type": "tool_reference", "name": ref_name} ) + elif item_type == "search_result": + search_text = _text_from_search_result(item) + if search_text: + tool_text_parts.append(search_text) + tool_content_parts.append( + {"type": "text", "text": search_text} + ) tool_text = "\n".join(tool_text_parts) if ( @@ -168,6 +321,41 @@ class AnthropicServing: tool_text = str(content) if content else "" return tool_text, tool_text + def _convert_assistant_thinking_blocks( + blocks: list[AnthropicContentBlock], + ) -> Optional[str]: + """Re-wrap prior-turn thinking blocks in the parser's own tokens. + + ``redacted_thinking`` carries encrypted bytes that no local + parser can interpret, so we raise rather than silently drop it. + On non-reasoning models (no detector configured) the rewrap is + best-effort: we log a warning and drop the thinking text so a + history echo doesn't 400 the whole request — the prior thinking + is opaque context the model didn't need anyway. + """ + if any(block.type == "redacted_thinking" for block in blocks): + raise ValueError("Anthropic redacted_thinking history is not supported") + + thinking_parts = [ + block.thinking + for block in blocks + if block.type == "thinking" and block.thinking + ] + if not thinking_parts: + return None + + try: + return self.openai_serving_chat.wrap_reasoning_history( + "\n".join(thinking_parts) + ) + except ValueError as e: + logger.warning( + "Dropping prior-turn thinking history (%d blocks): %s", + len(thinking_parts), + e, + ) + return None + # Add system message if provided if anthropic_request.system: if isinstance(anthropic_request.system, str): @@ -182,6 +370,22 @@ class AnthropicServing: system_text = "\n".join(system_parts) openai_messages.append({"role": "system", "content": system_text}) + def _emit_user_message(parts: list[dict]) -> None: + """Append accumulated parts as a user message, then clear them. + + Used to flush content collected BEFORE a tool_result so the + wire order stays user(pre) → tool → user(post). Without this + flush, text/image parts that appeared before a tool_result + block would be moved AFTER the tool message at end of loop. + """ + if not parts: + return + if len(parts) == 1 and parts[0]["type"] == "text": + openai_messages.append({"role": "user", "content": parts[0]["text"]}) + else: + openai_messages.append({"role": "user", "content": list(parts)}) + parts.clear() + # Convert messages for msg in anthropic_request.messages: if isinstance(msg.content, str): @@ -190,11 +394,26 @@ class AnthropicServing: # Complex content with blocks openai_msg = {"role": msg.role} - content_parts = [] - tool_calls = [] + content_parts: list[dict] = [] + tool_calls: list[dict] = [] + + if msg.role == "assistant": + reasoning_history = _convert_assistant_thinking_blocks(msg.content) + if reasoning_history is not None: + content_parts.append({"type": "text", "text": reasoning_history}) for block in msg.content: - if block.type == "text" and block.text: + # ``thinking``/``redacted_thinking`` blocks are surfaced via + # the reasoning-history reconstruction above; skip them here + # to avoid double-injecting their text into the prompt. + if block.type in ("thinking", "redacted_thinking"): + continue + + # ``is not None`` (not truthy) so an empty-string text block + # still produces a placeholder text part — without it, an + # assistant turn whose only content is "" vanishes and + # subsequent user→user pairs trip strict chat templates. + if block.type == "text" and block.text is not None: content_parts.append({"type": "text", "text": block.text}) elif block.type == "image" and block.source: @@ -204,6 +423,11 @@ class AnthropicServing: if image_part is not None: content_parts.append(image_part) + elif block.type == "search_result": + search_text = _text_from_search_result(block.model_dump()) + if search_text: + content_parts.append({"type": "text", "text": search_text}) + elif block.type == "tool_use": tool_call = { "id": block.id or f"call_{uuid.uuid4().hex}", @@ -223,8 +447,12 @@ class AnthropicServing: # 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 + # Tool results from user become separate tool messages. + # Flush any pending text/image first so the wire order + # is preserved (a tool_result that arrived AFTER a text + # block must come AFTER that text in OpenAI form too). if msg.role == "user": + _emit_user_message(content_parts) openai_messages.append( { "role": "tool", @@ -250,10 +478,19 @@ class AnthropicServing: openai_msg["content"] = content_parts[0]["text"] else: openai_msg["content"] = content_parts - elif not tool_calls: + openai_messages.append(openai_msg) + elif tool_calls: + openai_messages.append(openai_msg) + elif msg.role == "user": + # User turn that was entirely tool_results — the tool + # messages were already emitted above, nothing left. continue - - openai_messages.append(openai_msg) + else: + # Assistant turn with no content and no tool_calls: emit + # an empty-string placeholder so strict templates still + # see a valid role-alternation sequence. + openai_msg["content"] = "" + openai_messages.append(openai_msg) # Build ChatCompletionRequest request_data = { @@ -274,28 +511,119 @@ class AnthropicServing: # Enable usage in stream so we can report it if anthropic_request.stream: - request_data["stream_options"] = StreamOptions(include_usage=True) + request_data["stream_options"] = StreamOptions( + include_usage=True, + continuous_usage_stats=True, + ) chat_request = ChatCompletionRequest(**request_data) + if anthropic_request.thinking is not None: + # The protocol layer already enforces SDK shape: + # enabled -> budget_tokens required (>=1024), display optional + # disabled -> neither budget_tokens nor display allowed + # adaptive -> budget_tokens forbidden, display optional + # So by the time we get here ``budget_tokens`` can only be + # set on ``enabled``. The local backend has no equivalent + # hard-cap knob, so we log a WARNING instead of rejecting — + # the Anthropic SDK would have accepted the request and we + # mirror that. Operators see the unenforced budget in logs. + if anthropic_request.thinking.budget_tokens is not None: + logger.warning( + "Anthropic thinking.budget_tokens=%d is accepted for " + "SDK compatibility but the local backend has no " + "equivalent hard-cap knob — the budget is not enforced", + anthropic_request.thinking.budget_tokens, + ) + # Claude 4.7's ``adaptive`` is treated identically to ``enabled`` + # because the local backend has no auto-throttle equivalent. + # Anything other than ``disabled`` enables reasoning. + enabled = anthropic_request.thinking.type != "disabled" + if anthropic_request.thinking.display == "omitted": + # Anthropic 4.7 spec: keep reasoning ON but hide reasoning + # text from the client. The OpenAI streaming pipeline has + # no equivalent suppress knob — log so operators can see + # the request, then proceed with normal reasoning emission. + logger.warning( + "Anthropic thinking.display='omitted' is accepted for " + "SDK compatibility but reasoning text will still be " + "emitted to the client" + ) + self.openai_serving_chat.apply_reasoning_enabled(chat_request, enabled) + + # Claude 4.7 ``output_config``: map ``effort`` onto the OpenAI + # ``reasoning_effort`` knob. ``xhigh`` collapses to ``max`` because + # the OpenAI Literal does not include the Anthropic-only ``xhigh``. + # ``task_budget`` is a soft hint forwarded as a custom param so the + # model can see it without it becoming a hard cap (``max_tokens`` + # is still the hard cap). + if anthropic_request.output_config is not None: + oc = anthropic_request.output_config + if oc.effort is not None: + chat_request.reasoning_effort = ( + "max" if oc.effort == "xhigh" else oc.effort + ) + if oc.task_budget is not None: + # Custom params are silently ignored by backends that + # don't recognise them; logging it makes the propagation + # visible. + logger.info( + "Anthropic output_config.task_budget hint: %d %s", + oc.task_budget.total, + oc.task_budget.type, + ) + + # ``betas`` is the Anthropic SDK's opt-in feature list (e.g. + # ``["thinking-2025-08-04"]``). The local backend has no + # equivalent beta system; accept-and-log so requests don't 400. + if anthropic_request.betas: + logger.info( + "Anthropic request opted into betas %s — no-op locally", + anthropic_request.betas, + ) + # Convert tools. Deferred tools stay in the list with defer_loading=True; # the chat template hides them from the initial block and renders # them on demand when a tool_reference block names them. if anthropic_request.tools: - chat_request.tools = [ - Tool( - type="function", - defer_loading=tool.defer_loading, - function={ - "name": tool.name, - "description": tool.description or "", - "parameters": tool.input_schema, - }, - ) - for tool in anthropic_request.tools - ] + converted_tools = [] + for tool in anthropic_request.tools: + if is_server_tool(tool): + # Anthropic server-side tools (web_search_*, computer_*, + # bash_*, text_editor_*) have no client-side input_schema + # because Anthropic provides the implementation. We can't + # forward them to the OpenAI tools array (which requires a + # schema), so skip with a visible log. + logger.info( + "Skipping built-in Anthropic server tool %r (type=%r): " + "no native support in the OpenAI-compatible backend", + tool.name, + tool.type, + ) + continue - # Convert tool choice + # Custom tools always have a validated input_schema + # (enforced at Pydantic parse time). + converted_tools.append( + Tool( + type="function", + defer_loading=tool.defer_loading, + function={ + "name": tool.name, + "description": tool.description or "", + "parameters": tool.input_schema, + }, + ) + ) + + if converted_tools: + chat_request.tools = converted_tools + + # Convert tool choice. ``any``/``tool`` express a hard requirement + # ("the model MUST call a tool"); if every requested tool was a + # server-side Anthropic built-in that we just skipped, there is + # no tool the model could call. Silently downgrading to "no tool" + # would deceive the caller, so raise an explicit 400. if anthropic_request.tool_choice is not None: tc_type = anthropic_request.tool_choice.type if tc_type == "none": @@ -306,12 +634,30 @@ class AnthropicServing: elif tc_type == "any": chat_request.tool_choice = "required" elif tc_type == "tool": + tool_name = anthropic_request.tool_choice.name + # ``Tool.function`` is a ``Function`` Pydantic model, not + # a dict — access by attribute. A dict ``.get`` would + # AttributeError and surface as a 500 instead of the + # intended 400 / happy path. + if not any( + t.function.name == tool_name for t in chat_request.tools + ): + raise ValueError( + f"tool_choice references tool {tool_name!r} but it " + f"is not in the forwarded tools list " + f"(server-side Anthropic tools cannot be selected)" + ) chat_request.tool_choice = ToolChoice( type="function", - function=ToolChoiceFuncName( - name=anthropic_request.tool_choice.name - ), + function=ToolChoiceFuncName(name=tool_name), ) + elif tc_type in ("any", "tool"): + raise ValueError( + f"tool_choice={tc_type!r} requires at least one custom " + f"tool; all supplied tools were server-side Anthropic " + f"built-ins which the OpenAI-compatible backend cannot " + f"invoke" + ) elif chat_request.tools: chat_request.tool_choice = "auto" @@ -324,8 +670,10 @@ class AnthropicServing: raw_request: Request, ) -> JSONResponse: """Handle non-streaming Anthropic request by delegating to OpenAI handler.""" + # ``monotonic_time`` is ``time.perf_counter`` under the hood; the + # downstream stats layer subtracts other ``perf_counter`` samples + # from this, so they must come from the same clock. received_time = monotonic_time() - received_time_perf = time.perf_counter() # Validate error_msg = self.openai_serving_chat._validate_request(chat_request) @@ -338,36 +686,32 @@ class AnthropicServing: 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 asyncio.CancelledError: + raise except Exception as e: logger.exception("Error processing Anthropic request: %s", e) return self._error_response( status_code=500, - error_type="internal_error", + error_type="api_error", message="Internal server error", + exception_name=type(e).__name__, ) # 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", - ) + return self._convert_openai_error_response(response) # Convert to Anthropic response anthropic_response = self._convert_response(response) @@ -381,7 +725,6 @@ class AnthropicServing: ) -> Union[StreamingResponse, JSONResponse]: """Handle streaming Anthropic request.""" received_time = monotonic_time() - received_time_perf = time.perf_counter() # Validate error_msg = self.openai_serving_chat._validate_request(chat_request) @@ -393,21 +736,21 @@ class AnthropicServing: ) 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 asyncio.CancelledError: + raise except Exception as e: logger.exception("Error converting streaming request: %s", e) return self._error_response( status_code=500, - error_type="internal_error", + error_type="api_error", message="Internal server error", + exception_name=type(e).__name__, ) return StreamingResponse( @@ -435,108 +778,283 @@ class AnthropicServing: adapted_request, processed_request, raw_request ) - # State tracking - first_chunk = True content_block_index = 0 content_block_open = False + content_block_type: Optional[str] = None + captured_thinking_signature: str = "" finish_reason: Optional[str] = None - usage_info: Optional[dict] = None + final_usage: Optional[AnthropicUsage] = None + message_started = False + had_content_delta = False message_id = f"msg_{uuid.uuid4().hex}" model = anthropic_request.model - async for sse_line in openai_stream: + def _message_start_event(usage) -> MessageStartEvent: + return MessageStartEvent( + message=AnthropicMessagesResponse( + id=message_id, + content=[], + model=model, + usage=_anthropic_usage_from_openai( + usage, + include_input=True, + include_output=True, + force_zero_output=True, + ), + ), + ) + + def _emit(event: AnthropicStreamEvent) -> str: + return _wrap_sse_event( + event.model_dump_json(exclude_none=True), + event.type, + ) + + def _close_content_block_events() -> list[AnthropicStreamEvent]: + nonlocal content_block_index, content_block_open + nonlocal content_block_type, captured_thinking_signature + + events: list[AnthropicStreamEvent] = [] + if not content_block_open: + return events + + # Only emit signature_delta when a real signature is available. + # Anthropic's spec treats absence as "unsigned thinking"; an + # empty-string signature would fail downstream verifiers. + if content_block_type == "thinking" and captured_thinking_signature: + events.append( + ContentBlockDeltaEvent( + index=content_block_index, + delta=SignatureDelta( + signature=captured_thinking_signature, + ), + ) + ) + + events.append(ContentBlockStopEvent(index=content_block_index)) + content_block_open = False + content_block_type = None + content_block_index += 1 + captured_thinking_signature = "" + return events + + def _ensure_content_block_events( + block_type: str, + content_block: AnthropicContentBlock, + force_new: bool = False, + ) -> list[AnthropicStreamEvent]: + """Open a content_block, closing the prior one if needed. + + ``force_new=True`` closes an existing block even when its type + matches — required when a stream emits two consecutive + ``tool_use`` blocks: each tool needs its own + ``content_block_start``/``stop`` pair and its own + ``content_block_index``, otherwise the second tool's + ``input_json_delta`` chunks would append to the first tool's + JSON arguments and corrupt both tool calls. + """ + nonlocal content_block_open, content_block_type + + events: list[AnthropicStreamEvent] = [] + if content_block_open and (force_new or content_block_type != block_type): + events.extend(_close_content_block_events()) + if not content_block_open: + events.append( + ContentBlockStartEvent( + index=content_block_index, + content_block=content_block, + ) + ) + content_block_open = True + content_block_type = block_type + return events + + def _ensure_message_started(usage) -> list[str]: + """Emit message_start exactly once. Returns SSE frames to yield.""" + nonlocal message_started + if message_started: + return [] + message_started = True + return [_emit(_message_start_event(usage))] + + def _build_error_event(error_type: str, message: str) -> ErrorEvent: + return ErrorEvent( + error=AnthropicError(type=error_type, message=message), + ) + + def _flush_on_error(error_type: str, message: str) -> list[str]: + """Build a self-contained terminal SSE sequence on error. + + Guarantees that whatever events we emit on the failure path + leave the wire in a valid state: message_start (if not yet + sent), close any open content block, then ErrorEvent and + MessageStopEvent. Strict SDK clients reject streams whose + content_block_start has no matching content_block_stop, so + the close step is mandatory even on the error path. + """ + frames: list[str] = [] + frames.extend(_ensure_message_started(None)) + for event in _close_content_block_events(): + frames.append(_emit(event)) + frames.append(_emit(_build_error_event(error_type, message))) + frames.append(_emit(MessageStopEvent())) + return frames + + def _parse_upstream_error(data_str: str) -> Optional[tuple[str, str]]: + """Detect an OpenAI handler streaming-error envelope. + + ``OpenAIServingChat.create_streaming_error_response`` emits + ``data: {"error": {"object":"error","message":"...", + "type":"BadRequestError","code":400}}``; the regular + ChatCompletionStreamResponse validator rejects it. Pull the + type/message out so the Anthropic client sees the real + failure instead of a generic 'Stream processing error'. + """ + try: + payload = json.loads(data_str) + except (json.JSONDecodeError, ValueError): + return None + if not isinstance(payload, dict): + return None + err = payload.get("error") + if not isinstance(err, dict): + return None + upstream_message = err.get("message") or "Upstream error" + code = err.get("code") + error_type = ( + ERROR_TYPE_MAP.get(code, "api_error") + if isinstance(code, int) + else "api_error" + ) + return error_type, str(upstream_message) + + # Pre-first-chunk errors from the OpenAI generator (e.g. tokenization + # failure that raises ValueError before any chunk is yielded) would + # otherwise abort the StreamingResponse with no envelope at all and + # the client would see a half-open SSE / TCP close. Catch them here + # and emit a clean Anthropic error sequence instead. + try: + stream_iter = openai_stream.__aiter__() + except Exception as e: + logger.exception("Failed to open OpenAI stream: %s", e) + for frame in _flush_on_error("api_error", "Internal server error"): + yield frame + return + + while True: + try: + sse_line = await stream_iter.__anext__() + except StopAsyncIteration: + break + except asyncio.CancelledError: + raise + except ValueError as e: + # _generate_chat_stream re-raises ValueError when its own + # ``stream_started`` flag is still False — surface as a + # proper Anthropic error event rather than aborting the + # StreamingResponse generator. + logger.warning("OpenAI stream raised before first chunk: %s", e) + for frame in _flush_on_error( + "invalid_request_error", str(e) or "Request failed" + ): + yield frame + return + except Exception as e: + logger.exception("OpenAI stream raised mid-flight: %s", e) + for frame in _flush_on_error("api_error", "Internal server error"): + yield frame + return + if not sse_line.startswith("data: "): continue data_str = sse_line[6:].strip() if data_str == "[DONE]": + for frame in _ensure_message_started(None): + yield frame + + # No content AND no finish_reason: the backend dropped the + # stream silently. Surface as api_error so clients see the + # failure instead of a fake empty success. If finish_reason + # IS set we trust the backend's signal — a legitimate empty + # completion (max_tokens=1 stop, content filter, etc.) + # deserves a normal message_delta/message_stop pair, not + # an error that triggers SDK retry loops. + if not had_content_delta and finish_reason is None: + logger.warning( + "Stream produced no content and no finish_reason " + "before [DONE]; emitting api_error event" + ) + yield _emit( + _build_error_event("api_error", "Backend produced no content") + ) + yield _emit(MessageStopEvent()) + continue + # 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", - ) + for event in _close_content_block_events(): + yield _emit(event) # 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", + effective_finish = finish_reason or "stop" + if effective_finish not in STOP_REASON_MAP: + logger.warning( + "Unmapped streaming finish_reason %r; defaulting " + "to end_turn", + effective_finish, + ) + stop_reason = STOP_REASON_MAP.get(effective_finish, "end_turn") + yield _emit( + MessageDeltaEvent( + delta=AnthropicMessageEndDelta(stop_reason=stop_reason), + usage=final_usage or AnthropicUsage(output_tokens=0), + ) ) - # Emit message_stop - stop_msg = AnthropicStreamEvent(type="message_stop") - yield _wrap_sse_event( - stop_msg.model_dump_json(exclude_none=True), - "message_stop", - ) + yield _emit(MessageStopEvent()) 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 + except (ValidationError, json.JSONDecodeError, UnicodeDecodeError) as e: + # First check whether this is the OpenAI handler's + # streaming error envelope (validator rejects it because + # it lacks id/choices/created/model). Forwarding the real + # type/message keeps the failure debuggable instead of + # collapsing every backend error into "Stream processing + # error". + upstream = _parse_upstream_error(data_str) + if upstream is not None: + error_type, error_message = upstream + logger.warning( + "Forwarding upstream stream error (%s): %s", + error_type, + error_message, + ) + for frame in _flush_on_error(error_type, error_message): + yield frame + return - # First chunk: emit message_start - if first_chunk: - first_chunk = False + logger.warning( + "Failed to parse Anthropic stream chunk (%s): %s", + type(e).__name__, + data_str[:200], + ) + for frame in _flush_on_error("api_error", "Stream processing error"): + yield frame + return - 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, - ), - ), + if chunk.usage is not None: + final_usage = _anthropic_usage_from_openai( + chunk.usage, + include_input=False, + include_output=True, ) - 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: @@ -544,109 +1062,130 @@ class AnthropicServing: choice = chunk.choices[0] - # Capture finish reason + # Capture finish_reason on this chunk but DO NOT short-circuit: + # some OpenAI-compatible backends pack the final content token + # (or last tool-args fragment) into the same chunk as + # finish_reason. Skipping delta processing would silently drop + # that payload — sometimes the whole completion if it was a + # one-token reply. Fall through to the delta handlers below. if choice.finish_reason is not None: finish_reason = choice.finish_reason - continue delta = choice.delta + # Defer message_start until the first chunk carrying real prompt + # usage or content. OpenAI streams emit a role-only chunk before + # usage is available; emitting message_start there would ship + # input_tokens=0 to the client. + has_delta_payload = bool( + delta.reasoning_content + or delta.tool_calls + or (delta.content is not None and delta.content != "") + or chunk.usage + ) + # The finish_reason chunk should also flip message_started so a + # zero-content completion (the path that previously fired the + # 'Backend produced no content' error) emits the standard + # message_start before [DONE] closes the stream. + if ( + has_delta_payload or choice.finish_reason is not None + ) and not message_started: + yield _emit(_message_start_event(chunk.usage)) + message_started = True + + if ( + not has_delta_payload + and delta.role == "assistant" + and (delta.content is None or delta.content == "") + ): + continue + + # Handle reasoning content deltas + if delta.reasoning_content: + for event in _ensure_content_block_events( + "thinking", + ThinkingBlock(thinking=""), + ): + yield _emit(event) + + yield _emit( + ContentBlockDeltaEvent( + index=content_block_index, + delta=ThinkingDelta(thinking=delta.reasoning_content), + ) + ) + had_content_delta = True + # 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 + # New tool call: always close the previous block (even if + # it was also tool_use — each tool needs its own index) + # and start a fresh 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", + for event in _ensure_content_block_events( + "tool_use", + ToolUseBlock( 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 + force_new=True, + ): + yield _emit(event) + # A zero-argument tool call may never emit an + # input_json_delta; the tool_use start block itself is + # still meaningful content because it carries id/name. + had_content_delta = 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", + yield _emit( + ContentBlockDeltaEvent( + index=content_block_index, + delta=InputJsonDelta( + partial_json=tc_func.arguments, + ), + ) ) + had_content_delta = True 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, - ), + if content_block_type != "tool_use": + logger.warning( + "Dropping tool_call argument delta with no " + "open tool_use block: %r", + (tc_func.arguments or "")[:100], + ) + continue + yield _emit( + ContentBlockDeltaEvent( + index=content_block_index, + delta=InputJsonDelta( + partial_json=tc_func.arguments, + ), + ) ) - yield _wrap_sse_event( - delta_event.model_dump_json(exclude_none=True), - "content_block_delta", - ) - continue + had_content_delta = True # 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 + for event in _ensure_content_block_events( + "text", + TextBlock(text=""), + ): + yield _emit(event) - # 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", + yield _emit( + ContentBlockDeltaEvent( + index=content_block_index, + delta=TextDelta(text=delta.content), + ) ) + had_content_delta = True def _convert_response( self, response: ChatCompletionResponse @@ -654,7 +1193,7 @@ class AnthropicServing: """Convert an OpenAI ChatCompletionResponse to an Anthropic Messages response.""" if not response.choices: return AnthropicMessagesResponse( - content=[AnthropicContentBlock(type="text", text="")], + content=[TextBlock(text="")], model=response.model, stop_reason="end_turn", usage=AnthropicUsage(input_tokens=0, output_tokens=0), @@ -663,23 +1202,36 @@ class AnthropicServing: choice = response.choices[0] content: list[AnthropicContentBlock] = [] + # Add reasoning content as a thinking block. signature is omitted + # entirely when the backend doesn't provide one — empty strings + # would fail downstream Anthropic signature verifiers. + if choice.message.reasoning_content: + content.append(ThinkingBlock(thinking=choice.message.reasoning_content)) + # Add text content if choice.message.content: - content.append( - AnthropicContentBlock(type="text", text=choice.message.content) - ) + content.append(TextBlock(text=choice.message.content)) # Add tool calls if choice.message.tool_calls: for tool_call in choice.message.tool_calls: + raw_args = tool_call.function.arguments try: - tool_input = json.loads(tool_call.function.arguments) + tool_input = json.loads(raw_args) except (json.JSONDecodeError, TypeError): + # Surface invalid tool arguments so an empty-dict + # tool call is never indistinguishable from a real + # one when something downstream goes wrong. + logger.warning( + "Tool %r emitted invalid JSON arguments: %r — " + "defaulting to empty input", + tool_call.function.name, + (raw_args or "")[:200], + ) tool_input = {} content.append( - AnthropicContentBlock( - type="tool_use", + ToolUseBlock( id=tool_call.id, name=tool_call.function.name, input=tool_input, @@ -687,26 +1239,103 @@ class AnthropicServing: ) # Map stop reason - stop_reason = STOP_REASON_MAP.get(choice.finish_reason or "stop", "end_turn") + finish_reason = choice.finish_reason or "stop" + if finish_reason not in STOP_REASON_MAP: + logger.warning( + "Unmapped OpenAI finish_reason %r; defaulting to end_turn", + finish_reason, + ) + stop_reason = STOP_REASON_MAP.get(finish_reason, "end_turn") + + # Anthropic requires ``content`` to contain at least one block. + # Empty string completions (max_tokens=1 stop, content filter, etc.) + # would otherwise ship ``content=[]`` and break strict SDK parsers. + if not content: + content.append(TextBlock(text="")) 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, + usage=_anthropic_usage_from_openai( + response.usage, + include_input=True, + include_output=True, ), ) + def _convert_openai_error_response(self, response) -> JSONResponse: + """Forward an upstream OpenAI-handler error as an Anthropic error. + + The original error message is preserved for 4xx (after light + sanitization) so callers see the real validation failure. For 5xx + we always return a generic ``"Internal server error"`` to avoid + leaking ``str(e)`` payloads that the OpenAI handler builds from + raw exceptions (paths, tracebacks, prompt fragments, etc.). + """ + status_code = getattr(response, "status_code", 500) + body = getattr(response, "body", b"") or b"" + error_type = ERROR_TYPE_MAP.get(status_code, "api_error") + + upstream_message: Optional[str] = None + try: + payload = json.loads(body.decode("utf-8")) if body else None + except (json.JSONDecodeError, UnicodeDecodeError): + # Non-JSON body (HTML gateway error, plain text, ...). Use a + # bounded slice of the raw body so the client still has a + # useful hint instead of a generic placeholder. + try: + upstream_message = body.decode("utf-8", errors="replace")[:500] + except Exception: + upstream_message = None + else: + if isinstance(payload, dict): + error_payload = payload.get("error", payload) + if isinstance(error_payload, dict): + upstream_message = error_payload.get("message") or payload.get( + "message" + ) + # Honor the upstream error.type only for 4xx; 5xx is + # normalized below. + if status_code < 500: + upstream_type = error_payload.get("type") + if isinstance(upstream_type, str) and upstream_type: + error_type = upstream_type + elif isinstance(error_payload, str): + upstream_message = error_payload + elif isinstance(payload.get("message"), str): + upstream_message = payload["message"] + + message = _scrub_error_message(upstream_message or "", status_code) + return self._error_response( + status_code=status_code, + error_type=error_type, + message=message, + ) + def _error_response( self, status_code: int, error_type: str, message: str, + exception_name: Optional[str] = None, ) -> JSONResponse: - """Create an Anthropic-format error response.""" + """Create an Anthropic-format error response. + + ``error.type`` is restricted to Anthropic's documented enum so strict + SDK clients (anthropic-sdk-python / -typescript) keep parsing the + response into their typed error classes. ``exception_name`` — when + provided — is logged at WARNING level so operators can still grep + server-side, but it never reaches the wire. + """ + if exception_name: + logger.warning( + "Anthropic error response %s (exception=%s): %s", + error_type, + exception_name, + message, + ) error_resp = AnthropicErrorResponse( error=AnthropicError(type=error_type, message=message) ) @@ -732,10 +1361,13 @@ class AnthropicServing: messages=request.messages, max_tokens=1, # dummy, not used for counting system=request.system, + thinking=request.thinking, tools=request.tools, tool_choice=request.tool_choice, ) chat_request = self._convert_to_chat_completion_request(messages_request) + except asyncio.CancelledError: + raise except Exception as e: logger.exception("Error converting count_tokens request: %s", e) return self._error_response( @@ -764,10 +1396,13 @@ class AnthropicServing: input_tokens=input_tokens ).model_dump() ) + except asyncio.CancelledError: + raise except Exception as e: logger.exception("Error counting tokens: %s", e) return self._error_response( status_code=500, - error_type="internal_error", + error_type="api_error", message="Internal server error", + exception_name=type(e).__name__, ) diff --git a/python/sglang/srt/entrypoints/http_server.py b/python/sglang/srt/entrypoints/http_server.py index d7368383d..cc1002084 100644 --- a/python/sglang/srt/entrypoints/http_server.py +++ b/python/sglang/srt/entrypoints/http_server.py @@ -425,13 +425,77 @@ from sglang.srt.entrypoints.v1_loads import router as v1_loads_router app.include_router(v1_loads_router) +def _anthropic_validation_message(raw_errors) -> str: + """Render Pydantic-style errors for an Anthropic /v1/messages route. + + Builds a short ``loc: msg`` digest that names the offending fields without + leaking file paths or Python internals (the default ``str(exc)`` includes + the dispatcher's ``File "/.../http_server.py"`` line). + """ + parts: list[str] = [] + for err in raw_errors or []: + loc = err.get("loc") or () + if loc: + loc_str = ".".join(str(p) for p in loc if p not in ("body",)) + else: + loc_str = "" + msg = (err.get("msg") or "").strip() + if loc_str and msg: + parts.append(f"{loc_str}: {msg}") + elif msg: + parts.append(msg) + text = "; ".join(parts) or "Invalid request" + if len(text) > 500: + text = text[:500] + "…" + return text + + +def _anthropic_error_response(*, status_code: int, error_type: str, message: str): + """Anthropic-format error envelope: {"type":"error","error":{"type":...,"message":...}}.""" + return ORJSONResponse( + status_code=status_code, + content={ + "type": "error", + "error": {"type": error_type, "message": message}, + }, + ) + + @app.exception_handler(HTTPException) async def validation_exception_handler(request: Request, exc: HTTPException): """Enrich HTTP exception with status code and other details. For /v1/responses, emit OpenAI-style nested error envelope: {"error": {"message": "...", "type": "...", "param": null, "code": }} + For /v1/messages, emit Anthropic-style envelope so SDK clients can parse it. """ + if request.url.path.startswith("/v1/messages"): + # Map HTTP status to Anthropic error.type; fall back to api_error. + anthropic_type = { + 400: "invalid_request_error", + 401: "authentication_error", + 403: "permission_error", + 404: "not_found_error", + 413: "request_too_large", + 422: "invalid_request_error", + 429: "rate_limit_error", + 500: "api_error", + 502: "api_error", + 503: "overloaded_error", + 504: "api_error", + }.get(exc.status_code, "api_error") + # 5xx must never echo upstream detail (may contain stack/PII). + message = ( + "Internal server error" + if exc.status_code >= 500 + else (str(exc.detail) if exc.detail else "Request failed") + ) + return _anthropic_error_response( + status_code=exc.status_code, + error_type=anthropic_type, + message=message, + ) + # adjust fmt for responses api if request.url.path.startswith("/v1/responses"): nested_error = { @@ -458,8 +522,18 @@ async def validation_exception_handler(request: Request, exc: HTTPException): async def validation_exception_handler(request: Request, exc: RequestValidationError): """Override FastAPI's default 422 validation error with 400. - For /v1/responses, emit OpenAI-style nested error envelope; for other endpoints keep legacy format. + For /v1/messages, emit Anthropic-style envelope and scrub the message so + file paths or Python internals from the default ``str(exc)`` representation + never reach the client. For /v1/responses, keep OpenAI-style. Otherwise + use the legacy ErrorResponse shape. """ + if request.url.path.startswith("/v1/messages"): + return _anthropic_error_response( + status_code=HTTPStatus.BAD_REQUEST.value, + error_type="invalid_request_error", + message=_anthropic_validation_message(exc.errors()), + ) + exc_str = str(exc) errors_str = str(exc.errors()) @@ -1067,9 +1141,11 @@ async def dump_expert_distribution_record_async(): @auth_level(AuthLevel.ADMIN_OPTIONAL) async def update_weights_from_disk(obj: UpdateWeightFromDiskReqInput, request: Request): """Update the weights from disk inplace without re-launching the server.""" - success, message, num_paused_requests = ( - await _global_state.tokenizer_manager.update_weights_from_disk(obj, request) - ) + ( + success, + message, + num_paused_requests, + ) = await _global_state.tokenizer_manager.update_weights_from_disk(obj, request) content = { "success": success, @@ -1093,10 +1169,11 @@ async def update_weights_from_disk(obj: UpdateWeightFromDiskReqInput, request: R async def init_weights_send_group_for_remote_instance( obj: InitWeightsSendGroupForRemoteInstanceReqInput, request: Request ): - success, message = ( - await _global_state.tokenizer_manager.init_weights_send_group_for_remote_instance( - obj, request - ) + ( + success, + message, + ) = await _global_state.tokenizer_manager.init_weights_send_group_for_remote_instance( + obj, request ) content = {"success": success, "message": message} if success: @@ -1110,10 +1187,11 @@ async def init_weights_send_group_for_remote_instance( async def send_weights_to_remote_instance( obj: SendWeightsToRemoteInstanceReqInput, request: Request ): - success, message = ( - await _global_state.tokenizer_manager.send_weights_to_remote_instance( - obj, request - ) + ( + success, + message, + ) = await _global_state.tokenizer_manager.send_weights_to_remote_instance( + obj, request ) content = {"success": success, "message": message} if success: @@ -1182,9 +1260,10 @@ async def destroy_weights_update_group( obj: DestroyWeightsUpdateGroupReqInput, request: Request ): """Destroy the parameter update group.""" - success, message = ( - await _global_state.tokenizer_manager.destroy_weights_update_group(obj, request) - ) + ( + success, + message, + ) = await _global_state.tokenizer_manager.destroy_weights_update_group(obj, request) content = {"success": success, "message": message} return ORJSONResponse( content, status_code=200 if success else HTTPStatus.BAD_REQUEST @@ -1219,10 +1298,11 @@ async def update_weights_from_distributed( obj: UpdateWeightsFromDistributedReqInput, request: Request ): """Update model parameter from distributed online.""" - success, message = ( - await _global_state.tokenizer_manager.update_weights_from_distributed( - obj, request - ) + ( + success, + message, + ) = await _global_state.tokenizer_manager.update_weights_from_distributed( + obj, request ) content = {"success": success, "message": message} diff --git a/python/sglang/srt/entrypoints/openai/serving_chat.py b/python/sglang/srt/entrypoints/openai/serving_chat.py index 2ea4c4909..7cd9d3645 100644 --- a/python/sglang/srt/entrypoints/openai/serving_chat.py +++ b/python/sglang/srt/entrypoints/openai/serving_chat.py @@ -1564,6 +1564,111 @@ class OpenAIServingChat(OpenAIServingBase): ): request.skip_special_tokens = False + def wrap_reasoning_history(self, reasoning_text: str) -> str: + """Wrap prior-turn reasoning in the detector's own start/end tokens. + + Pulling the delimiters from the detector keeps adapters in lockstep + with any future parser that ships non-```` markers — Mistral's + ``[THINK]``, Gemma4's ``think_start_self_label = "thought\\n"``, etc. + Falling back to a plain string is unsafe: it would let prior + thinking text reach a non-reasoning model as ordinary assistant + content, so the caller must surface this state, not paper over it. + """ + if self._reasoning_detector is None: + raise ValueError( + "Cannot rewrap thinking history: no reasoning detector is " + "configured for this model" + ) + d = self._reasoning_detector + return ( + f"{d.think_start_token}{d.think_start_self_label}" + f"{reasoning_text}\n{d.think_end_token}" + ) + + def _reasoning_default_mode(self) -> Optional[str]: + if self._reasoning_detector is None: + return None + return self._reasoning_detector.reasoning_default + + def _get_reasoning_toggle_param(self) -> Optional[str]: + """Resolve the chat-template kwarg that toggles reasoning, if any.""" + config = self.template_manager.reasoning_config + if config is not None: + return config.toggle_param + + mode = self._reasoning_default_mode() + if mode in ("thinking", "enable_thinking"): + return mode + if mode in ("explicit_thinking", "explicit_enable_thinking"): + return mode.replace("explicit_", "") + return None + + def apply_reasoning_enabled( + self, request: ChatCompletionRequest, enabled: bool + ) -> None: + """Force the request into the requested reasoning-on/off mode. + + Mirrors the read-side logic in ``_get_reasoning_from_request``; + the two must stay in sync. Always-on models cannot be disabled, + so explicit ``enabled=False`` raises rather than silently leaving + reasoning on. + """ + if not self.reasoning_parser: + if enabled: + raise ValueError( + "Anthropic thinking is not supported for models without " + "a reasoning parser" + ) + return + + if self.reasoning_parser == "hunyuan": + request.reasoning_effort = "medium" if enabled else "no_think" + return + + config = self.template_manager.reasoning_config + is_mistral = (config is not None and config.special_case == "mistral") or ( + config is None and self._reasoning_default_mode() == "mistral" + ) + if is_mistral: + request.reasoning_effort = "medium" if enabled else "none" + return + + is_always_on = (config is not None and config.special_case == "always") or ( + config is None and self._reasoning_default_mode() == "always" + ) + if is_always_on: + if not enabled: + raise ValueError( + f"Reasoning parser '{self.reasoning_parser}' is always-on " + f"and cannot be disabled via Anthropic thinking" + ) + return + + toggle_param = self._get_reasoning_toggle_param() + # The read side (``_get_reasoning_from_request``) returns False + # whenever ``config.toggle_param is None`` OR + # ``config.default_enabled is None``. The write side must mirror + # both conditions: if ``default_enabled`` is unset we cannot + # actually honor an ``enabled=True`` request even when the toggle + # name itself is resolvable, so writing the kwarg would set up the + # template to emit reasoning tokens while the parser ignores them + # (literal ```` markers leak into the assistant text). + config = self.template_manager.reasoning_config + read_side_supported = toggle_param is not None and ( + config is None or config.default_enabled is not None + ) + if not read_side_supported: + if not enabled: + return + raise ValueError( + f"Anthropic thinking is not supported for reasoning parser " + f"'{self.reasoning_parser}'" + ) + + chat_template_kwargs = dict(request.chat_template_kwargs or {}) + chat_template_kwargs[toggle_param] = enabled + request.chat_template_kwargs = chat_template_kwargs + def _get_reasoning_from_request(self, request: ChatCompletionRequest) -> bool: """Determine whether reasoning mode should be enabled for this request. diff --git a/test/registered/unit/entrypoints/anthropic/test_serving.py b/test/registered/unit/entrypoints/anthropic/test_serving.py new file mode 100644 index 000000000..c94d03dd2 --- /dev/null +++ b/test/registered/unit/entrypoints/anthropic/test_serving.py @@ -0,0 +1,1339 @@ +import asyncio +import json +import unittest +from types import SimpleNamespace + +from sglang.test.test_utils import maybe_stub_sgl_kernel + +maybe_stub_sgl_kernel() # must precede imports that may pull in sgl_kernel + +from fastapi.responses import JSONResponse # noqa: E402 + +from sglang.srt.entrypoints.anthropic.protocol import ( # noqa: E402 + AnthropicMessagesRequest, +) +from sglang.srt.entrypoints.anthropic.serving import AnthropicServing # noqa: E402 +from sglang.srt.entrypoints.openai.protocol import ( # noqa: E402 + ChatCompletionRequest, + ChatCompletionResponse, +) +from sglang.test.ci.ci_register import register_cpu_ci # noqa: E402 + +register_cpu_ci(est_time=1, suite="base-a-test-cpu") + + +class _FakeOpenAIServingChat: + def __init__(self, stream_lines=None): + self.stream_lines = stream_lines or [] + self.apply_reasoning_calls: list[bool] = [] + + def _generate_chat_stream(self, adapted_request, processed_request, raw_request): + async def _gen(): + for line in self.stream_lines: + yield line + + return _gen() + + def apply_reasoning_enabled(self, chat_request, enabled): + self.apply_reasoning_calls.append(enabled) + + def wrap_reasoning_history(self, text): + return f"\n{text}\n" + + +class _FakeNonStreamingErrorOpenAI: + """Returns a configurable error response from the OpenAI handler.""" + + def __init__(self, status_code=400, body=None, content=None): + self._status_code = status_code + self._body = body + self._content = content + + def _validate_request(self, chat_request): + return None + + def _convert_to_internal_request(self, chat_request, raw_request): + return SimpleNamespace(), chat_request + + async def _handle_non_streaming_request( + self, adapted_request, processed_request, raw_request + ): + if self._body is not None: + # Build a response object exposing raw bytes via `.body`. + return SimpleNamespace(status_code=self._status_code, body=self._body) + return JSONResponse( + status_code=self._status_code, + content=self._content + or { + "error": { + "type": "invalid_request_error", + "message": "context length exceeded", + } + }, + ) + + +class _FakeNonStreamingOpenAI: + """Returns a configurable ChatCompletionResponse from the OpenAI handler.""" + + def __init__(self, response): + self._response = response + + def _validate_request(self, chat_request): + return None + + def _convert_to_internal_request(self, chat_request, raw_request): + return SimpleNamespace(), chat_request + + async def _handle_non_streaming_request( + self, adapted_request, processed_request, raw_request + ): + return self._response + + +def _chunk(choices=None, usage=None): + data = { + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "created": 0, + "model": "test-model", + "choices": choices or [], + } + if usage is not None: + data["usage"] = usage + return f"data: {json.dumps(data)}\n\n" + + +def _choice(delta, finish_reason=None): + return { + "index": 0, + "delta": delta, + "finish_reason": finish_reason, + } + + +async def _collect_anthropic_events(serving, anthropic_request): + events = [] + async for sse in serving._generate_anthropic_stream( + adapted_request=object(), + processed_request=object(), + anthropic_request=anthropic_request, + raw_request=object(), + ): + for line in sse.splitlines(): + if line.startswith("data: "): + events.append(json.loads(line[6:])) + return events + + +class TestAnthropicServing(unittest.TestCase): + def _serving(self, stream_lines=None): + return AnthropicServing(_FakeOpenAIServingChat(stream_lines)) + + def _anthropic_request(self, **overrides): + data = { + "model": "test-model", + "max_tokens": 16, + "messages": [{"role": "user", "content": "hello"}], + "stream": True, + } + data.update(overrides) + return AnthropicMessagesRequest.model_validate(data) + + def test_stream_closes_tool_block_before_text_delta(self): + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": ""})]), + _chunk( + [ + _choice( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": { + "name": "lookup", + "arguments": '{"query"', + }, + } + ] + } + ) + ] + ), + _chunk( + [ + _choice( + { + "tool_calls": [ + { + "index": 0, + "type": "function", + "function": {"arguments": ': "sglang"}'}, + } + ] + } + ) + ] + ), + _chunk([_choice({"content": "done"})]), + _chunk([_choice({}, finish_reason="stop")]), + "data: [DONE]\n\n", + ] + ) + + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + block_events = [ + (event["type"], event.get("content_block", {}).get("type")) + for event in events + if event["type"].startswith("content_block") + ] + + self.assertEqual( + block_events, + [ + ("content_block_start", "tool_use"), + ("content_block_delta", None), + ("content_block_delta", None), + ("content_block_stop", None), + ("content_block_start", "text"), + ("content_block_delta", None), + ("content_block_stop", None), + ], + ) + + text_delta = [ + event + for event in events + if event["type"] == "content_block_delta" + and event["delta"].get("type") == "text_delta" + ][0] + self.assertEqual(text_delta["index"], 1) + + def test_stream_reasoning_content_uses_thinking_block(self): + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": ""})]), + _chunk([_choice({"reasoning_content": "think first"})]), + _chunk([_choice({"content": "answer"})]), + _chunk([_choice({}, finish_reason="stop")]), + "data: [DONE]\n\n", + ] + ) + + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + content_events = [ + event for event in events if event["type"].startswith("content_block") + ] + + self.assertEqual(content_events[0]["content_block"]["type"], "thinking") + # Signature is absent (None and excluded) — never emit empty + # string, which would fail downstream Anthropic signature verifiers. + self.assertNotIn("signature", content_events[0]["content_block"]) + self.assertEqual(content_events[1]["delta"]["type"], "thinking_delta") + self.assertEqual(content_events[1]["delta"]["thinking"], "think first") + # No empty signature_delta event between thinking_delta and content_block_stop. + self.assertEqual(content_events[2]["type"], "content_block_stop") + self.assertEqual(content_events[3]["content_block"]["type"], "text") + # Confirm no signature_delta event was emitted in the entire stream. + sig_deltas = [ + event + for event in events + if event["type"] == "content_block_delta" + and event.get("delta", {}).get("type") == "signature_delta" + ] + self.assertEqual(sig_deltas, []) + + def test_stream_usage_subtracts_cache_read_and_omits_final_input_tokens(self): + usage = { + "prompt_tokens": 10, + "completion_tokens": 0, + "total_tokens": 10, + "prompt_tokens_details": {"cached_tokens": 4}, + } + final_usage = { + "prompt_tokens": 10, + "completion_tokens": 2, + "total_tokens": 12, + "prompt_tokens_details": {"cached_tokens": 4}, + } + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": ""})]), + _chunk([_choice({"content": "hi"})], usage=usage), + _chunk([], usage=final_usage), + "data: [DONE]\n\n", + ] + ) + + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + message_start = [event for event in events if event["type"] == "message_start"][ + 0 + ] + message_delta = [event for event in events if event["type"] == "message_delta"][ + 0 + ] + + self.assertEqual(message_start["message"]["usage"]["input_tokens"], 6) + self.assertEqual( + message_start["message"]["usage"]["cache_read_input_tokens"], 4 + ) + self.assertNotIn("input_tokens", message_delta["usage"]) + self.assertEqual(message_delta["usage"]["output_tokens"], 2) + + def test_non_streaming_usage_subtracts_cache_read_tokens(self): + response = ChatCompletionResponse.model_validate( + { + "id": "chatcmpl-test", + "model": "test-model", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "ok"}, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 2, + "total_tokens": 12, + "prompt_tokens_details": {"cached_tokens": 4}, + }, + } + ) + + anthropic_response = self._serving()._convert_response(response) + + self.assertEqual(anthropic_response.usage.input_tokens, 6) + self.assertEqual(anthropic_response.usage.output_tokens, 2) + self.assertEqual(anthropic_response.usage.cache_read_input_tokens, 4) + + def test_tool_result_search_result_content_is_flattened(self): + request = AnthropicMessagesRequest.model_validate( + { + "model": "test-model", + "max_tokens": 16, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "call_1", + "content": [ + { + "type": "search_result", + "title": "SGLang docs", + "source": "https://docs.sglang.ai", + "content": [ + { + "type": "text", + "text": "Anthropic API notes", + } + ], + } + ], + } + ], + } + ], + } + ) + + chat_request = self._serving()._convert_to_chat_completion_request(request) + tool_message = [ + msg + for msg in chat_request.model_dump()["messages"] + if msg["role"] == "tool" + ][0] + + self.assertIn("SGLang docs", tool_message["content"]) + self.assertIn("https://docs.sglang.ai", tool_message["content"]) + self.assertIn("Anthropic API notes", tool_message["content"]) + + def test_builtin_web_search_tool_without_schema_is_skipped(self): + request = AnthropicMessagesRequest.model_validate( + { + "model": "test-model", + "max_tokens": 16, + "messages": [{"role": "user", "content": "search sglang"}], + "tools": [{"name": "web_search", "type": "web_search_20250305"}], + "tool_choice": {"type": "auto"}, + } + ) + + chat_request = self._serving()._convert_to_chat_completion_request(request) + + self.assertIsNone(chat_request.tools) + self.assertEqual(chat_request.tool_choice, "none") + + def test_custom_tool_without_schema_is_rejected(self): + # With the discriminated union, an AnthropicCustomTool variant must + # carry an input_schema. The check fires at request-parse time + # (Pydantic raises ValidationError, a subclass of ValueError). + with self.assertRaisesRegex(ValueError, "input_schema"): + AnthropicMessagesRequest.model_validate( + { + "model": "test-model", + "max_tokens": 16, + "messages": [{"role": "user", "content": "call a tool"}], + "tools": [{"name": "custom_without_schema"}], + } + ) + + def test_non_streaming_openai_error_response_is_forwarded(self): + serving = AnthropicServing(_FakeNonStreamingErrorOpenAI()) + chat_request = ChatCompletionRequest( + model="test-model", + max_tokens=16, + messages=[{"role": "user", "content": "hello"}], + ) + anthropic_request = self._anthropic_request(stream=False) + + response = asyncio.run( + serving._handle_non_streaming(chat_request, anthropic_request, object()) + ) + payload = json.loads(response.body) + + self.assertEqual(response.status_code, 400) + self.assertEqual(payload["error"]["type"], "invalid_request_error") + self.assertEqual(payload["error"]["message"], "context length exceeded") + + # ------------------------------------------------------------------ + # Edge-case coverage added in the review-fix pass + # ------------------------------------------------------------------ + + def test_stream_text_then_tool_use_closes_text_block(self): + """Text deltas followed by tool_use must close the text block before opening tool_use index.""" + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": "Hello"})]), + _chunk([_choice({"content": " world"})]), + _chunk( + [ + _choice( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": { + "name": "search", + "arguments": '{"q":"hello"}', + }, + } + ] + } + ) + ] + ), + _chunk([_choice({}, finish_reason="tool_calls")]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + block_events = [ + (event["type"], event.get("content_block", {}).get("type")) + for event in events + if event["type"].startswith("content_block") + ] + # text block (start, delta, delta, stop) then tool_use (start, delta, stop) + self.assertEqual(block_events[0], ("content_block_start", "text")) + text_stop_idx = next( + i for i, ev in enumerate(block_events) if ev == ("content_block_stop", None) + ) + tool_start_idx = next( + i + for i, ev in enumerate(block_events) + if ev == ("content_block_start", "tool_use") + ) + self.assertLess(text_stop_idx, tool_start_idx) + + def test_stream_tool_use_without_arguments_is_not_empty_completion(self): + """A zero-argument tool call is valid content even without input_json_delta.""" + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": ""})]), + _chunk( + [ + _choice( + { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "ping", "arguments": ""}, + } + ] + } + ) + ] + ), + _chunk([_choice({}, finish_reason="tool_calls")]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + + self.assertFalse(any(event["type"] == "error" for event in events)) + tool_start = next( + event + for event in events + if event["type"] == "content_block_start" + and event["content_block"]["type"] == "tool_use" + ) + self.assertEqual(tool_start["content_block"]["name"], "ping") + message_delta = next( + event for event in events if event["type"] == "message_delta" + ) + self.assertEqual(message_delta["delta"]["stop_reason"], "tool_use") + + def test_stream_no_usage_chunk_emits_error_event(self): + """Stream that yields only [DONE] (no content delta) must surface as an error event.""" + serving = self._serving(["data: [DONE]\n\n"]) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + types = [event["type"] for event in events] + # Sequence: message_start, error, message_stop + self.assertEqual(types, ["message_start", "error", "message_stop"]) + error_event = events[1] + self.assertEqual(error_event["error"]["type"], "api_error") + self.assertIn("no content", error_event["error"]["message"].lower()) + + def test_cache_read_exceeds_prompt_tokens_clamps_to_zero(self): + """When cached_tokens > prompt_tokens, input_tokens clamps to 0 instead of going negative.""" + usage = { + "prompt_tokens": 4, + "completion_tokens": 0, + "total_tokens": 4, + "prompt_tokens_details": {"cached_tokens": 10}, + } + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": ""})]), + _chunk([_choice({"content": "ok"})], usage=usage), + _chunk([_choice({}, finish_reason="stop")]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + message_start = next(e for e in events if e["type"] == "message_start") + usage_out = message_start["message"]["usage"] + self.assertEqual(usage_out["input_tokens"], 0) + self.assertEqual(usage_out["cache_read_input_tokens"], 10) + + def test_usage_without_prompt_tokens_details(self): + """Usage object without prompt_tokens_details must omit cache_read_input_tokens cleanly.""" + usage = {"prompt_tokens": 5, "completion_tokens": 0, "total_tokens": 5} + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": ""})]), + _chunk([_choice({"content": "ok"})], usage=usage), + _chunk([_choice({}, finish_reason="stop")]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + message_start = next(e for e in events if e["type"] == "message_start") + usage_out = message_start["message"]["usage"] + self.assertEqual(usage_out["input_tokens"], 5) + self.assertNotIn("cache_read_input_tokens", usage_out) + + def test_non_streaming_error_with_non_json_body(self): + """Non-JSON upstream error body falls back to body[:500] as the message (for 4xx).""" + serving = AnthropicServing( + _FakeNonStreamingErrorOpenAI( + status_code=400, + body=b"upstream gateway rejected: bad payload", + ) + ) + chat_request = ChatCompletionRequest( + model="test-model", + max_tokens=16, + messages=[{"role": "user", "content": "hello"}], + ) + anthropic_request = self._anthropic_request(stream=False) + response = asyncio.run( + serving._handle_non_streaming(chat_request, anthropic_request, object()) + ) + payload = json.loads(response.body) + self.assertEqual(response.status_code, 400) + self.assertEqual(payload["error"]["type"], "invalid_request_error") + self.assertIn("upstream gateway rejected", payload["error"]["message"]) + + def test_non_streaming_error_5xx_scrubs_message(self): + """5xx errors always return a generic message regardless of upstream payload.""" + for status_code, expected_type in [ + (500, "api_error"), + (502, "api_error"), + (503, "overloaded_error"), + (504, "api_error"), + ]: + serving = AnthropicServing( + _FakeNonStreamingErrorOpenAI( + status_code=status_code, + body=b'{"error":{"message":"sensitive internals: /opt/secret","type":"internal"}}', + ) + ) + chat_request = ChatCompletionRequest( + model="test-model", + max_tokens=16, + messages=[{"role": "user", "content": "hello"}], + ) + anthropic_request = self._anthropic_request(stream=False) + response = asyncio.run( + serving._handle_non_streaming(chat_request, anthropic_request, object()) + ) + payload = json.loads(response.body) + self.assertEqual( + response.status_code, + status_code, + f"status {status_code} should be preserved", + ) + self.assertEqual( + payload["error"]["type"], + expected_type, + f"status {status_code} should map to {expected_type}", + ) + self.assertEqual( + payload["error"]["message"], + "Internal server error", + f"status {status_code} must scrub the message; got {payload['error']['message']!r}", + ) + + def test_non_streaming_response_includes_thinking_block(self): + """When the OpenAI response carries reasoning_content, the Anthropic response has a thinking block first.""" + response = ChatCompletionResponse.model_validate( + { + "id": "chatcmpl-test", + "model": "test-model", + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": "the answer is 4", + "reasoning_content": "2 + 2 = 4", + }, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 3, + "completion_tokens": 5, + "total_tokens": 8, + }, + } + ) + anthropic_response = self._serving()._convert_response(response) + # thinking block first, then text block + self.assertEqual(anthropic_response.content[0].type, "thinking") + self.assertEqual(anthropic_response.content[0].thinking, "2 + 2 = 4") + self.assertEqual(anthropic_response.content[1].type, "text") + self.assertEqual(anthropic_response.content[1].text, "the answer is 4") + + def test_request_thinking_enabled_invokes_apply_reasoning_enabled(self): + """``thinking={"type":"enabled", "budget_tokens":N}`` flips reasoning on. + + ``budget_tokens`` is required by the SDK shape on ``enabled``; the + local backend does not enforce it but accepts the value. + """ + serving = self._serving() + request = self._anthropic_request( + thinking={"type": "enabled", "budget_tokens": 1024}, stream=False + ) + serving._convert_to_chat_completion_request(request) + self.assertEqual(serving.openai_serving_chat.apply_reasoning_calls, [True]) + + def test_request_thinking_disabled_invokes_apply_reasoning_enabled(self): + """``thinking={"type": "disabled"}`` must flip the reasoning toggle off.""" + serving = self._serving() + request = self._anthropic_request(thinking={"type": "disabled"}, stream=False) + serving._convert_to_chat_completion_request(request) + self.assertEqual(serving.openai_serving_chat.apply_reasoning_calls, [False]) + + def test_request_thinking_enabled_with_budget_tokens_logs_warning(self): + """SDK shape: ``enabled`` requires ``budget_tokens``. We accept it + (the SDK would), but log a WARNING because the local backend has + no equivalent hard-cap knob — the budget is not enforced.""" + import logging + + serving = self._serving() + request = self._anthropic_request( + thinking={"type": "enabled", "budget_tokens": 2048}, stream=False + ) + with self.assertLogs( + "sglang.srt.entrypoints.anthropic.serving", level=logging.WARNING + ) as log: + serving._convert_to_chat_completion_request(request) + self.assertEqual(serving.openai_serving_chat.apply_reasoning_calls, [True]) + self.assertTrue( + any("budget_tokens=2048" in r and "not enforced" in r for r in log.output), + f"expected unenforced-budget warning: {log.output}", + ) + + def test_request_thinking_enabled_requires_budget_tokens(self): + """SDK requires ``budget_tokens`` for ``type=enabled`` — Pydantic 400.""" + from pydantic import ValidationError + + with self.assertRaises(ValidationError) as ctx: + self._anthropic_request(thinking={"type": "enabled"}, stream=False) + self.assertIn("budget_tokens", str(ctx.exception)) + + def test_request_thinking_enabled_budget_below_min_is_rejected(self): + """SDK doc: ``budget_tokens`` must be >= 1024.""" + from pydantic import ValidationError + + with self.assertRaises(ValidationError) as ctx: + self._anthropic_request( + thinking={"type": "enabled", "budget_tokens": 512}, stream=False + ) + self.assertIn("1024", str(ctx.exception)) + + def test_request_thinking_disabled_with_display_is_rejected(self): + """SDK ``ThinkingConfigDisabledParam`` has no ``display`` field.""" + from pydantic import ValidationError + + with self.assertRaises(ValidationError) as ctx: + self._anthropic_request( + thinking={"type": "disabled", "display": "omitted"}, stream=False + ) + self.assertIn("display", str(ctx.exception)) + + def test_request_thinking_disabled_with_budget_is_rejected(self): + """SDK ``ThinkingConfigDisabledParam`` has no ``budget_tokens`` field.""" + from pydantic import ValidationError + + with self.assertRaises(ValidationError) as ctx: + self._anthropic_request( + thinking={"type": "disabled", "budget_tokens": 2048}, stream=False + ) + self.assertIn("budget_tokens", str(ctx.exception)) + + def test_request_thinking_adaptive_with_budget_is_rejected(self): + """SDK ``ThinkingConfigAdaptiveParam`` has no ``budget_tokens`` field.""" + from pydantic import ValidationError + + with self.assertRaises(ValidationError) as ctx: + self._anthropic_request( + thinking={"type": "adaptive", "budget_tokens": 2048}, stream=False + ) + self.assertIn("budget_tokens", str(ctx.exception)) + + def test_request_thinking_adaptive_is_treated_as_enabled(self): + """Claude 4.7 ``thinking.type='adaptive'`` (the SDK default for + unknown models) must be accepted and routed to ``apply_reasoning_enabled(True)``. + """ + serving = self._serving() + request = self._anthropic_request(thinking={"type": "adaptive"}, stream=False) + serving._convert_to_chat_completion_request(request) + self.assertEqual(serving.openai_serving_chat.apply_reasoning_calls, [True]) + + def test_request_thinking_display_omitted_logs_warning_but_still_enables(self): + """``thinking.display='omitted'`` is accepted; reasoning stays on + because we cannot suppress reasoning text from the OpenAI stream. + ``enabled`` requires ``budget_tokens`` per SDK shape.""" + import logging + + serving = self._serving() + request = self._anthropic_request( + thinking={ + "type": "enabled", + "budget_tokens": 1024, + "display": "omitted", + }, + stream=False, + ) + with self.assertLogs( + "sglang.srt.entrypoints.anthropic.serving", level=logging.WARNING + ) as log: + serving._convert_to_chat_completion_request(request) + self.assertEqual(serving.openai_serving_chat.apply_reasoning_calls, [True]) + self.assertTrue(any("omitted" in r for r in log.output)) + + def test_request_output_config_effort_maps_to_reasoning_effort(self): + """``output_config.effort`` rows map onto ``reasoning_effort``.""" + for anthropic_effort, openai_effort in [ + ("low", "low"), + ("medium", "medium"), + ("high", "high"), + ("xhigh", "max"), # OpenAI Literal has no xhigh + ("max", "max"), + ]: + with self.subTest(anthropic_effort=anthropic_effort): + serving = self._serving() + request = self._anthropic_request( + output_config={"effort": anthropic_effort}, stream=False + ) + chat_request = serving._convert_to_chat_completion_request(request) + self.assertEqual(chat_request.reasoning_effort, openai_effort) + + def test_request_output_config_task_budget_is_logged_not_enforced(self): + """``task_budget`` is a soft hint; ``max_tokens`` is the hard cap.""" + import logging + + serving = self._serving() + request = self._anthropic_request( + output_config={"task_budget": {"type": "tokens", "total": 32768}}, + stream=False, + ) + with self.assertLogs( + "sglang.srt.entrypoints.anthropic.serving", level=logging.INFO + ) as log: + chat_request = serving._convert_to_chat_completion_request(request) + # max_tokens is untouched + self.assertEqual(chat_request.max_tokens, 16) + self.assertTrue(any("task_budget" in r and "32768" in r for r in log.output)) + + def test_request_task_budget_with_remaining_is_accepted(self): + """SDK's ``BetaTokenTaskBudgetParam`` has a ``remaining`` field + used for client-side compaction. Must round-trip cleanly.""" + serving = self._serving() + request = self._anthropic_request( + output_config={ + "task_budget": {"type": "tokens", "total": 32768, "remaining": 12000} + }, + stream=False, + ) + # Must not raise; pre-existing logging still works. + serving._convert_to_chat_completion_request(request) + self.assertEqual(request.output_config.task_budget.remaining, 12000) + + def test_request_betas_is_accepted_and_logged(self): + """The Anthropic SDK attaches ``betas`` to many requests; must not 400.""" + import logging + + serving = self._serving() + request = self._anthropic_request( + betas=["thinking-2025-08-04", "computer-use-2025-01-24"], + stream=False, + ) + with self.assertLogs( + "sglang.srt.entrypoints.anthropic.serving", level=logging.INFO + ) as log: + serving._convert_to_chat_completion_request(request) + self.assertTrue(any("betas" in r for r in log.output)) + + def test_assistant_thinking_history_is_rewrapped_for_chat_template(self): + """Past-turn thinking blocks get re-emitted via wrap_reasoning_history.""" + serving = self._serving() + request = self._anthropic_request( + stream=False, + messages=[ + {"role": "user", "content": "hi"}, + { + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "ponder"}, + {"type": "text", "text": "hello"}, + ], + }, + {"role": "user", "content": "again"}, + ], + ) + chat_request = serving._convert_to_chat_completion_request(request) + # ``ChatCompletionRequest.messages`` is a list of Pydantic + # ChatCompletionMessage*Param instances; access via attributes. + assistant_msg = next(m for m in chat_request.messages if m.role == "assistant") + content = assistant_msg.content + # Reasoning history sits in front; the thinking block itself is dropped + # from the prompt so its text is not duplicated. + if isinstance(content, list): + texts = [] + for part in content: + if isinstance(part, dict): + texts.append(part.get("text", "")) + else: + texts.append(getattr(part, "text", "") or "") + else: + texts = [content] + joined = "\n".join(texts) + self.assertIn("", joined) + self.assertIn("ponder", joined) + self.assertNotIn("\nponder\n\nponder", joined) + + def test_redacted_thinking_history_is_rejected(self): + """``redacted_thinking`` cannot be rendered by local parsers.""" + serving = self._serving() + request = self._anthropic_request( + stream=False, + messages=[ + { + "role": "assistant", + "content": [ + {"type": "redacted_thinking", "data": "opaque"}, + ], + }, + ], + ) + with self.assertRaises(ValueError): + serving._convert_to_chat_completion_request(request) + + def test_stream_text_then_thinking_closes_text_block(self): + """Text deltas followed by reasoning_content must close the text block before opening thinking.""" + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": "Direct"})]), + _chunk([_choice({"reasoning_content": "but wait, let me think"})]), + _chunk([_choice({}, finish_reason="stop")]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + block_events = [ + (event["type"], event.get("content_block", {}).get("type")) + for event in events + if event["type"].startswith("content_block") + ] + # text (start, delta, stop) then thinking (start, delta, stop) + self.assertEqual(block_events[0], ("content_block_start", "text")) + text_stop_idx = next( + i for i, ev in enumerate(block_events) if ev == ("content_block_stop", None) + ) + thinking_start_idx = next( + i + for i, ev in enumerate(block_events) + if ev == ("content_block_start", "thinking") + ) + self.assertLess(text_stop_idx, thinking_start_idx) + + def test_stream_consecutive_tool_calls_get_separate_blocks(self): + """Two tool_use calls in sequence must occupy distinct content_block indices.""" + serving = self._serving( + [ + _chunk( + [ + _choice( + { + "tool_calls": [ + { + "index": 0, + "id": "call_a", + "function": { + "name": "alpha", + "arguments": '{"x":1}', + }, + } + ] + } + ) + ] + ), + _chunk( + [ + _choice( + { + "tool_calls": [ + { + "index": 1, + "id": "call_b", + "function": { + "name": "beta", + "arguments": '{"y":2}', + }, + } + ] + } + ) + ] + ), + _chunk([_choice({}, finish_reason="tool_calls")]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + starts = [ + (e["index"], e["content_block"]["name"]) + for e in events + if e["type"] == "content_block_start" + ] + stops = [e["index"] for e in events if e["type"] == "content_block_stop"] + deltas = [ + (e["index"], e["delta"].get("partial_json")) + for e in events + if e["type"] == "content_block_delta" + and e["delta"].get("type") == "input_json_delta" + ] + # Each tool gets its own start, its own stop, and its own + # argument delta — without the fix, beta's args were appended + # to alpha's index 0 block. + self.assertEqual(starts, [(0, "alpha"), (1, "beta")]) + self.assertEqual(stops, [0, 1]) + self.assertEqual(deltas, [(0, '{"x":1}'), (1, '{"y":2}')]) + + def test_stream_finish_chunk_with_payload_emits_delta(self): + """A chunk carrying both finish_reason and content must not drop the content.""" + serving = self._serving( + [ + _chunk([_choice({"role": "assistant"})]), + _chunk([_choice({"content": "last token"}, finish_reason="stop")]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + text_deltas = [ + e["delta"]["text"] + for e in events + if e["type"] == "content_block_delta" + and e["delta"].get("type") == "text_delta" + ] + self.assertEqual(text_deltas, ["last token"]) + # And stop_reason still travels via message_delta + message_delta = next(e for e in events if e["type"] == "message_delta") + self.assertEqual(message_delta["delta"]["stop_reason"], "end_turn") + + def test_stream_empty_completion_with_finish_reason_emits_message_delta(self): + """An empty stream with a finish_reason is a legitimate stop, not api_error.""" + serving = self._serving( + [ + _chunk([_choice({"role": "assistant"})]), + _chunk([_choice({}, finish_reason="length")]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + types = [e["type"] for e in events] + self.assertIn("message_start", types) + self.assertIn("message_delta", types) + self.assertIn("message_stop", types) + self.assertNotIn("error", types) + message_delta = next(e for e in events if e["type"] == "message_delta") + self.assertEqual(message_delta["delta"]["stop_reason"], "max_tokens") + + def test_stream_no_finish_no_content_still_emits_api_error(self): + """Backend that drops both content and finish_reason is genuinely broken.""" + serving = self._serving( + [ + _chunk([_choice({"role": "assistant"})]), + "data: [DONE]\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + types = [e["type"] for e in events] + self.assertIn("error", types) + err = next(e for e in events if e["type"] == "error") + self.assertEqual(err["error"]["type"], "api_error") + + def test_stream_upstream_error_envelope_is_forwarded(self): + """OpenAI handler streaming-error JSON must surface real type/message.""" + upstream_error = { + "error": { + "object": "error", + "message": "context length exceeded", + "type": "BadRequestError", + "code": 400, + } + } + serving = self._serving( + [ + _chunk([_choice({"role": "assistant"})]), + f"data: {json.dumps(upstream_error)}\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + err = next(e for e in events if e["type"] == "error") + self.assertEqual(err["error"]["type"], "invalid_request_error") + self.assertEqual(err["error"]["message"], "context length exceeded") + # message_stop must still close the stream + self.assertEqual(events[-1]["type"], "message_stop") + + def test_stream_parse_failure_closes_open_content_block(self): + """Unparsable mid-stream chunk must still close any open content_block.""" + serving = self._serving( + [ + _chunk([_choice({"role": "assistant", "content": "first"})]), + "data: {not-json\n\n", + ] + ) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + types = [e["type"] for e in events] + # Sequence: message_start, content_block_start, content_block_delta, + # content_block_stop, error, message_stop + self.assertIn("content_block_start", types) + self.assertEqual( + types.count("content_block_stop"), + types.count("content_block_start"), + f"unbalanced block events: {types}", + ) + self.assertIn("error", types) + self.assertEqual(types[-1], "message_stop") + + def test_stream_pre_first_chunk_value_error_emits_envelope(self): + """ValueError before any chunk must yield a clean Anthropic error sequence.""" + + class _RaisingOpenAI(_FakeOpenAIServingChat): + def _generate_chat_stream( + self, adapted_request, processed_request, raw_request + ): + async def _gen(): + raise ValueError("tokenization failed") + yield # pragma: no cover + + return _gen() + + serving = AnthropicServing(_RaisingOpenAI()) + events = asyncio.run( + _collect_anthropic_events(serving, self._anthropic_request()) + ) + types = [e["type"] for e in events] + self.assertEqual(types[0], "message_start") + self.assertIn("error", types) + err = next(e for e in events if e["type"] == "error") + self.assertEqual(err["error"]["type"], "invalid_request_error") + self.assertIn("tokenization failed", err["error"]["message"]) + self.assertEqual(types[-1], "message_stop") + + def test_server_tool_only_with_tool_choice_any_raises_400(self): + """A request with only server-side tools cannot honor tool_choice=any.""" + serving = self._serving() + request = self._anthropic_request( + stream=False, + tools=[{"type": "web_search_20250305", "name": "web_search"}], + tool_choice={"type": "any"}, + ) + with self.assertRaises(ValueError) as ctx: + serving._convert_to_chat_completion_request(request) + self.assertIn("tool_choice", str(ctx.exception)) + + def test_server_tool_only_with_tool_choice_auto_is_allowed(self): + """tool_choice=auto over server-only tools is a no-op (model decides).""" + serving = self._serving() + request = self._anthropic_request( + stream=False, + tools=[{"type": "web_search_20250305", "name": "web_search"}], + tool_choice={"type": "auto"}, + ) + # Must not raise; the request just runs with no client-side tools. + chat_request = serving._convert_to_chat_completion_request(request) + self.assertIsNone(chat_request.tools) + + def test_tool_choice_named_custom_tool_is_resolved(self): + """tool_choice={type:'tool', name:'X'} where X is a custom tool wires through.""" + serving = self._serving() + request = self._anthropic_request( + stream=False, + tools=[ + { + "type": "custom", + "name": "lookup", + "input_schema": { + "type": "object", + "properties": {"q": {"type": "string"}}, + }, + } + ], + tool_choice={"type": "tool", "name": "lookup"}, + ) + # Must not AttributeError: Tool.function is a Pydantic model, not a + # dict — access must be via .name, never .get("name"). + chat_request = serving._convert_to_chat_completion_request(request) + self.assertEqual(chat_request.tool_choice.type, "function") + self.assertEqual(chat_request.tool_choice.function.name, "lookup") + + def test_tool_choice_named_unknown_tool_raises_400(self): + """tool_choice={type:'tool', name:'X'} where X is missing must raise.""" + serving = self._serving() + request = self._anthropic_request( + stream=False, + tools=[ + { + "type": "custom", + "name": "lookup", + "input_schema": {"type": "object", "properties": {}}, + } + ], + tool_choice={"type": "tool", "name": "nonexistent"}, + ) + with self.assertRaises(ValueError) as ctx: + serving._convert_to_chat_completion_request(request) + self.assertIn("nonexistent", str(ctx.exception)) + + def test_convert_response_non_streaming_empty_content_keeps_block(self): + """Empty-string completion must still produce a content list of len 1.""" + response = ChatCompletionResponse.model_validate( + { + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 0, + "model": "test-model", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": ""}, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 5, + "completion_tokens": 0, + "total_tokens": 5, + }, + } + ) + serving = self._serving() + anthropic_response = serving._convert_response(response) + self.assertEqual(len(anthropic_response.content), 1) + self.assertEqual(anthropic_response.content[0].type, "text") + self.assertEqual(anthropic_response.content[0].text, "") + + def test_error_response_does_not_leak_exception_name(self): + """``error.type`` must stay in Anthropic's documented literal set.""" + serving = self._serving() + response = serving._error_response( + status_code=500, + error_type="api_error", + message="Internal server error", + exception_name="KeyError", + ) + body = json.loads(bytes(response.body).decode()) + self.assertEqual(body["error"]["type"], "api_error") + + def test_user_message_text_tool_text_preserves_order(self): + """User message [text, tool_result, text] must stay user→tool→user on the wire.""" + serving = self._serving() + request = self._anthropic_request( + stream=False, + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "first"}, + { + "type": "tool_result", + "tool_use_id": "call_x", + "content": "ok", + }, + {"type": "text", "text": "second"}, + ], + } + ], + ) + chat_request = serving._convert_to_chat_completion_request(request) + # chat_request.messages items are Pydantic ChatCompletionMessage*Param + # variants — use attribute access, not subscripts. + roles = [m.role for m in chat_request.messages] + self.assertEqual(roles, ["user", "tool", "user"]) + self.assertEqual(chat_request.messages[0].content, "first") + self.assertEqual(chat_request.messages[2].content, "second") + + def test_empty_text_assistant_turn_preserves_role_alternation(self): + """Assistant turn with only empty text must NOT vanish from the wire.""" + serving = self._serving() + request = self._anthropic_request( + stream=False, + messages=[ + {"role": "user", "content": "u1"}, + {"role": "assistant", "content": [{"type": "text", "text": ""}]}, + {"role": "user", "content": "u2"}, + ], + ) + chat_request = serving._convert_to_chat_completion_request(request) + roles = [m.role for m in chat_request.messages] + # Without the fix this collapses to ['user', 'user'] and breaks + # strict role-alternation chat templates (qwen, llama, mistral). + self.assertEqual(roles, ["user", "assistant", "user"]) + + def test_thinking_history_drop_on_missing_detector(self): + """Replaying a thinking block on a non-reasoning model should not 400.""" + + class _NoDetectorOpenAI(_FakeOpenAIServingChat): + def wrap_reasoning_history(self, text): + raise ValueError("no reasoning detector is configured") + + serving = AnthropicServing(_NoDetectorOpenAI()) + request = self._anthropic_request( + stream=False, + messages=[ + { + "role": "assistant", + "content": [{"type": "thinking", "thinking": "I think..."}], + }, + {"role": "user", "content": "follow-up"}, + ], + ) + # Must convert successfully; the thinking block is silently dropped. + chat_request = serving._convert_to_chat_completion_request(request) + roles = [m.role for m in chat_request.messages] + self.assertIn("user", roles) + # The assistant turn was rendered (as empty placeholder) so + # alternation is preserved. + self.assertIn("assistant", roles) + + def test_stop_reason_content_filter_falls_back_with_warning(self): + """Unmapped OpenAI finish_reasons default to 'end_turn' + log a warning. + + ``content_filter`` and ``abort`` have no entry in STOP_REASON_MAP + because Anthropic's ``stop_reason`` Literal (end_turn/max_tokens/ + stop_sequence/tool_use) has no perfect target. The fallback must + produce a spec-valid stop_reason and a WARNING so operators don't + silently lose the safety/abort signal. + """ + import logging + + response = ChatCompletionResponse.model_validate( + { + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 0, + "model": "test-model", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "hi"}, + "finish_reason": "content_filter", + } + ], + "usage": { + "prompt_tokens": 1, + "completion_tokens": 1, + "total_tokens": 2, + }, + } + ) + serving = self._serving() + with self.assertLogs( + "sglang.srt.entrypoints.anthropic.serving", level=logging.WARNING + ) as log: + anthropic_response = serving._convert_response(response) + self.assertEqual(anthropic_response.stop_reason, "end_turn") + self.assertTrue( + any("content_filter" in rec for rec in log.output), + f"expected a warning mentioning the unmapped finish_reason: {log.output}", + ) + + +if __name__ == "__main__": + unittest.main()