diff --git a/backends/exllamav3/model.py b/backends/exllamav3/model.py index a0902327..a0be9ebf 100644 --- a/backends/exllamav3/model.py +++ b/backends/exllamav3/model.py @@ -846,7 +846,9 @@ def validate_context_length( prompt: str, params: BaseSamplerRequest, mm_embeddings: Optional[MultimodalEmbeddingWrapper] = None, - ): + ) -> int: + """Validate the prompt against the context limit and return its length.""" + context_len = len( self.encode_tokens( prompt, @@ -869,6 +871,8 @@ def validate_context_length( allocation_boundary, ) + return context_len + async def generate( self, request_id: str, diff --git a/common/config_models.py b/common/config_models.py index fb0db10e..7640b234 100644 --- a/common/config_models.py +++ b/common/config_models.py @@ -66,10 +66,11 @@ class NetworkConfig(BaseConfigModel): "NOTE: Only enable this for debug purposes." ), ) - api_servers: Optional[List[Literal["oai", "kobold"]]] = Field( + api_servers: Optional[List[Literal["oai", "kobold", "anthropic"]]] = Field( ["OAI"], description=( - 'Select API servers to enable (default: ["OAI"]).\nPossible values: OAI, Kobold.' + 'Select API servers to enable (default: ["OAI"]).\n' + "Possible values: OAI, Kobold, Anthropic." ), ) sse_ping_interval: Optional[int] = Field( diff --git a/common/model.py b/common/model.py index 871a52a7..539c4a9c 100644 --- a/common/model.py +++ b/common/model.py @@ -309,15 +309,24 @@ def check_context_length( prompts: str | list[str], params: BaseSamplerRequest, mm_embeddings: Optional[MultimodalEmbeddingWrapper] = None, -): - """Reject oversized prompts before a streaming response commits HTTP 200.""" +) -> int: + """ + Reject oversized prompts before a streaming response commits HTTP 200. + + Returns the longest prompt's length in tokens, so a caller that needs the + prompt token count up front doesn't have to tokenize a second time. + """ if isinstance(prompts, str): prompts = [prompts] + lengths = [] + try: for prompt in prompts: - container.validate_context_length(prompt, params, mm_embeddings) + lengths.append(container.validate_context_length(prompt, params, mm_embeddings)) except ContextLengthExceededError as exc: error_message = handle_request_error(str(exc), exc_info=False).error.message raise ContextLengthHTTPException(error_message) from exc + + return max(lengths, default=0) diff --git a/config_sample.yml b/config_sample.yml index 7b16b06b..9400b689 100644 --- a/config_sample.yml +++ b/config_sample.yml @@ -28,7 +28,7 @@ network: send_tracebacks: false # Select API servers to enable (default: ["OAI"]). - # Possible values: OAI, Kobold. + # Possible values: OAI, Kobold, Anthropic. api_servers: ["OAI"] # Seconds between SSE keep-alive pings on streaming responses (default: 15). diff --git a/docs/02.-Server-options.md b/docs/02.-Server-options.md index 0bccad0e..00472c33 100644 --- a/docs/02.-Server-options.md +++ b/docs/02.-Server-options.md @@ -21,7 +21,7 @@ All of these options have descriptive comments above them. You should not need t | disable_auth | Bool (False) | Disables API authentication | | disable_fetch_requests | Bool (False) | Disables fetching external content when responding to requests (ex. fetching images from URLs) | | send_tracebacks | Bool (False) | Send server tracebacks to client.

Note: It's not recommended to enable this if sharing the instance with others. | -| api_servers | List[String] (["OAI"]) | API servers to enable. Possible values `"OAI", "Kobold"` | +| api_servers | List[String] (["OAI"]) | API servers to enable. Possible values `"OAI", "Kobold", "Anthropic"` | | sse_ping_interval | Int (15) | Interval in seconds between SSE keep-alive pings on streaming responses. Set to 0 to disable pings. | ### Logging Options diff --git a/docs/03.-Usage.md b/docs/03.-Usage.md index 59b8f9a4..5f9842e1 100644 --- a/docs/03.-Usage.md +++ b/docs/03.-Usage.md @@ -4,7 +4,7 @@ TabbyAPI's main use-case is to be an API server for running ExllamaV3 models. ### API Server -Currently TabbyAPI supports clients that use the [OpenAI](https://platform.openai.com/docs/api-reference) standard and [KoboldAI](https://lite.koboldai.net/koboldcpp_api)'s API. +Currently TabbyAPI supports clients that use the [OpenAI](https://platform.openai.com/docs/api-reference) standard, [KoboldAI](https://lite.koboldai.net/koboldcpp_api)'s API, and [Anthropic](https://docs.claude.com/en/api/messages)'s Messages API. In addition, there are expanded parameters to generation endpoints along with administrative endpoints for loading, unloading, loras, sampling overrides, etc. @@ -26,6 +26,120 @@ curl http://localhost:5000/v1/completions \ }' ``` +### Anthropic API + +The Anthropic Messages API is off by default. Enable it in `config.yml`: + +```yml +network: + api_servers: ["OAI", "Anthropic"] +``` + +This exposes two endpoints: +- `/v1/messages`: Generates a message from a conversation. +- `/v1/messages/count_tokens`: Counts the tokens an equivalent request would consume. + +Requests are translated into a chat completion internally, so the loaded model's +prompt template, reasoning tags, and sampler settings apply exactly as they do on +`/v1/chat/completions`. A prompt template is required. + +Anthropic clients authenticate with an `x-api-key` header, which TabbyAPI already +accepts, so pointing an Anthropic SDK at the server only takes a base URL change. +Errors are returned in the Anthropic envelope rather than TabbyAPI's usual one. + +Streaming is supported and emits the Anthropic event sequence (`message_start`, +the `content_block_*` lifecycle, `message_delta`, `message_stop`). Reasoning +models stream a `thinking` block ahead of the `text` block. There is no `[DONE]` +sentinel; `message_stop` ends the stream. + +Tool use is supported and works with every tool call format the model's +`tool_format` setting supports, since tools are translated into the same shape +`/v1/chat/completions` uses. `tool_choice` maps as `auto`/`none` unchanged, `any` +onto "required", and `{"type": "tool", "name": ...}` onto a forced call. +Because tool calls are parsed once generation finishes, a streamed `tool_use` +block arrives as one `input_json_delta` carrying the whole argument object +rather than incrementally. + +A `system` role is accepted inside `messages`, which is how recent Anthropic +models take operator instructions mid-conversation and how Claude Code sends +them. Chat templates generally only allow a system turn in first position, so a +later one is carried in a user turn wrapped in `` rather than +being folded into the leading system prompt, which would rewrite the front of +the prefix and invalidate the prompt cache on every turn. + +Two details are worth knowing. A `tool_result` marked `is_error` has its text +prefixed with `Error: `, because chat templates have no concept of a failed call +and the model can only react to the failure if it can read it. And Anthropic +packs every tool result for a turn into a single user message, which is fanned +out into one `tool` message per result, since that is what chat templates +expect. + +Images are supported on vision models, from either a `base64` or a `url` source. +Sending one to a model without vision is rejected rather than dropped, since the +model would otherwise be asked about a picture it was never shown. Images are +also accepted inside a `tool_result`, which is how screenshot-returning tools +report back. + +`GET /v1/models` serves the Anthropic field names (`type`, `display_name`, +`created_at`) alongside the OpenAI ones, so an Anthropic SDK can list models +without a separate endpoint. The loaded model's card reports its context window +as `max_input_tokens`, the name the Anthropic model schema uses, with +`max_tokens` beside it; both come from `max_seq_len`, which is also served under +its own name inside `parameters` for OpenAI-side clients. TabbyAPI has no output +limit separate from the context, so the two are equal. The prompt template +content is left out of a listing because of its size, so fetch `/v1/model` for +that. + +Support is currently partial. These are not implemented yet and are rejected with +a descriptive error rather than being silently dropped: +- Documents (PDF input) +- Image `file` sources, which reference the Anthropic Files API +- Anthropic's server-side tools (web search, code execution) + +Fields with no local equivalent are accepted and ignored: `cache_control`, +`service_tier`, `container`, `mcp_servers`, and `thinking.budget_tokens`. The +`thinking` type maps onto the model's `enable_thinking` template variable. + +As a TabbyAPI extension, a request may carry `template_vars` (alias +`chat_template_kwargs`), exactly as on `/v1/chat/completions`. The Anthropic +SDKs won't send an unknown field, so pass it through their escape hatch — +`extra_body={"template_vars": {...}}` in the Python SDK, or plain HTTP: + +```bash +curl http://localhost:5000/v1/messages \ +-H "Content-Type: application/json" \ +-H "x-api-key: " \ +-d '{ + "model": "Meta-Llama-3-8B-exl3", + "max_tokens": 400, + "template_vars": {"reasoning_effort": "high"}, + "messages": [{"role": "user", "content": "Plan a birthday party."}] +}' +``` + +Precedence matches the chat completion path: the model's `template_vars_default` +first, then anything derived from `thinking`, then the request's +`template_vars`, and finally `template_vars_force`, which always wins. So a +model configured with `template_vars_force: {enable_thinking: true}` keeps +thinking on no matter what the request asks for. + +Below is an example CURL request using the messages endpoint: + +```bash +curl http://localhost:5000/v1/messages \ +-H "Content-Type: application/json" \ +-H "x-api-key: " \ +-H "anthropic-version: 2023-06-01" \ +-d '{ + "model": "Meta-Llama-3-8B-exl3", + "max_tokens": 400, + "system": "You are a concise assistant.", + "messages": [ + {"role": "user", "content": "Name three primary colors."} + ] +}' +``` + ### Authentication Every call to a TabbyAPI endpoint requires some form of authentication. Keys have two types of permissions: diff --git a/endpoints/Anthropic/errors.py b/endpoints/Anthropic/errors.py new file mode 100644 index 00000000..1cbfd49e --- /dev/null +++ b/endpoints/Anthropic/errors.py @@ -0,0 +1,105 @@ +"""Error handling for the Anthropic API. + +Anthropic clients parse a different error envelope than the one TabbyAPI +returns elsewhere, so responses on these routes are reshaped into: + + {"type": "error", "error": {"type": , "message": }} + +The reshaping lives in a route class rather than an app-level exception +handler so it also covers errors raised while solving dependencies (a failed +API key check) and request validation, neither of which reach the endpoint. +""" + +from fastapi import HTTPException, Request, Response +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from fastapi.routing import APIRoute +from typing import Callable, Optional, Tuple + +from common.networking import handle_request_error + + +# Anthropic error types by status code. Codes TabbyAPI raises that Anthropic +# does not define (422 from a template failure, 503 from an unloaded model) +# map onto the closest documented type. +ERROR_TYPES = { + 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", + 503: "api_error", + 529: "overloaded_error", +} + + +def error_type_for_status(status_code: int) -> str: + """Get the Anthropic error type for a status code.""" + + if status_code in ERROR_TYPES: + return ERROR_TYPES[status_code] + + return "invalid_request_error" if status_code < 500 else "api_error" + + +def error_content(message: str, error_type: str) -> dict: + """Build the error envelope Anthropic clients expect.""" + + return {"type": "error", "error": {"type": error_type, "message": message}} + + +class AnthropicHTTPException(HTTPException): + """An HTTP error carrying an explicit Anthropic error type.""" + + def __init__(self, status_code: int, message: str, error_type: Optional[str] = None): + super().__init__(status_code=status_code, detail=message) + + self.error_type = error_type or error_type_for_status(status_code) + + +def request_error( + status_code: int, message: str, error_type: Optional[str] = None, exc_info: bool = False +) -> AnthropicHTTPException: + """Log a request error and return the exception to raise.""" + + error_message = handle_request_error(message, exc_info=exc_info).error.message + + return AnthropicHTTPException(status_code, error_message, error_type) + + +def exception_to_response(exc: Exception) -> Tuple[int, dict]: + """Map an exception raised while serving a route onto an error response.""" + + if isinstance(exc, AnthropicHTTPException): + return exc.status_code, error_content(exc.detail, exc.error_type) + + if isinstance(exc, HTTPException): + # Raised by shared TabbyAPI code: auth, inline model loading, chat + # template rendering, context length + detail = exc.detail if isinstance(exc.detail, str) else str(exc.detail) + return exc.status_code, error_content(detail, error_type_for_status(exc.status_code)) + + if isinstance(exc, RequestValidationError): + return 422, error_content(str(exc.errors()), "invalid_request_error") + + raise exc + + +class AnthropicRoute(APIRoute): + """Route class that returns errors in the Anthropic envelope.""" + + def get_route_handler(self) -> Callable: + original_route_handler = super().get_route_handler() + + async def anthropic_route_handler(request: Request) -> Response: + try: + return await original_route_handler(request) + except (HTTPException, RequestValidationError) as exc: + status_code, content = exception_to_response(exc) + + return JSONResponse(status_code=status_code, content=content) + + return anthropic_route_handler diff --git a/endpoints/Anthropic/router.py b/endpoints/Anthropic/router.py new file mode 100644 index 00000000..1779ff9e --- /dev/null +++ b/endpoints/Anthropic/router.py @@ -0,0 +1,134 @@ +import asyncio +from asyncio import CancelledError, InvalidStateError + +from fastapi import APIRouter, Depends, Request +from sse_starlette import EventSourceResponse + +from common import model +from common.auth import check_api_key +from common.logger import xlogger +from common.model import check_model_container +from common.networking import DisconnectHandler, get_sse_ping_interval +from common.tabby_config import config +from endpoints.Anthropic.errors import AnthropicRoute, request_error +from endpoints.Anthropic.types.messages import ( + CountTokensRequest, + CountTokensResponse, + MessagesRequest, + MessagesResponse, +) +from endpoints.Anthropic.utils.convert import convert_messages_request +from endpoints.Anthropic.utils.messages import convert_response, count_tokens +from endpoints.Anthropic.utils.stream import stream_generate_message +from endpoints.OAI.utils.chat_completion import apply_chat_template, generate_chat_completion +from endpoints.OAI.utils.common_ import load_inline_model + + +api_name = "Anthropic" +router = APIRouter(route_class=AnthropicRoute) +urls = { + "Messages": "http://{host}:{port}/v1/messages", + "Token counting": "http://{host}:{port}/v1/messages/count_tokens", +} + +# Block when model is still loading while second inline load request comes in +load_lock: asyncio.Lock = asyncio.Lock() + + +def setup(): + return router + + +async def _resolve_model(model_name: str | None, request: Request): + """Load an inline model if one was named and return the model directory.""" + + async with load_lock: + if model_name: + await load_inline_model(model_name, request) + else: + await check_model_container() + + return model.container.model_dir + + +def _check_prompt_template(): + """Reject the request if the loaded model has no prompt template.""" + + if model.container.prompt_template is None: + raise request_error( + 422, "The Anthropic API is disabled because a prompt template is not set." + ) + + +# Messages endpoint +@router.post( + "/v1/messages", + dependencies=[Depends(check_api_key)], +) +async def messages_request(request: Request, data: MessagesRequest) -> MessagesResponse: + """Generates a message from a conversation.""" + + raw_json = await request.json() + xlogger.debug("[ENDPOINT] /v1/messages", {"raw": raw_json}) + + if data.stream and config.developer.disable_request_streaming: + # Returning a non-streaming body to a client that asked for SSE would + # fail in the client's parser rather than say what went wrong + raise request_error( + 400, "Streaming is disabled on this server (developer.disable_request_streaming)." + ) + + model_path = await _resolve_model(data.model, request) + _check_prompt_template() + + converted = convert_messages_request(data) + prompt, mm_embeddings = await apply_chat_template(converted) + + try: + disconnect_handler = DisconnectHandler(request, "/v1/messages") + await disconnect_handler.poll() + + if data.stream: + # Checked before the response commits HTTP 200, and reused as the + # prompt token count message_start has to carry up front + input_tokens = model.check_context_length(prompt, converted, mm_embeddings) + + return EventSourceResponse( + stream_generate_message( + prompt, + mm_embeddings, + data, + converted, + request, + model_path, + disconnect_handler, + input_tokens, + ), + ping=get_sse_ping_interval(), + ) + + completion = await generate_chat_completion( + prompt, mm_embeddings, converted, request, model_path, disconnect_handler + ) + + return convert_response(completion, data, model_path.name) + + except (CancelledError, InvalidStateError) as ex: + raise request_error(422, "/v1/messages request cancelled by user.") from ex + + +# Token counting endpoint +@router.post( + "/v1/messages/count_tokens", + dependencies=[Depends(check_api_key)], +) +async def count_tokens_request(request: Request, data: CountTokensRequest) -> CountTokensResponse: + """Counts the tokens an equivalent Messages request would consume.""" + + raw_json = await request.json() + xlogger.debug("[ENDPOINT] /v1/messages/count_tokens", {"raw": raw_json}) + + await _resolve_model(data.model, request) + _check_prompt_template() + + return await count_tokens(data) diff --git a/endpoints/Anthropic/types/messages.py b/endpoints/Anthropic/types/messages.py new file mode 100644 index 00000000..8229115c --- /dev/null +++ b/endpoints/Anthropic/types/messages.py @@ -0,0 +1,289 @@ +"""Types for the Anthropic Messages API.""" + +from pydantic import AliasChoices, BaseModel, ConfigDict, Field +from typing import Annotated, Any, Dict, List, Literal, Optional, Union +from uuid import uuid4 + + +# Request content blocks +# +# Blocks carry per-type fields, so they're modelled as a discriminated union +# with a permissive fallback. The fallback lets the converter report an +# unsupported block by name instead of returning a pydantic validation dump. +# Fields the server ignores (cache_control, citations) are dropped silently, +# which is what pydantic does with extra keys by default. + + +class TextBlock(BaseModel): + """A text block.""" + + type: Literal["text"] + text: str + + +class ThinkingBlock(BaseModel): + """A thinking block replayed by the client from a previous turn.""" + + type: Literal["thinking"] + thinking: str = "" + signature: Optional[str] = None + + +class RedactedThinkingBlock(BaseModel): + """An encrypted thinking block. Carries nothing this server can replay.""" + + type: Literal["redacted_thinking"] + data: Optional[str] = None + + +class ImageSource(BaseModel): + """Where an image block's data comes from.""" + + model_config = ConfigDict(extra="allow") + + # Not a Literal so an unsupported source is reported by name + type: str + media_type: Optional[str] = None + data: Optional[str] = None + url: Optional[str] = None + file_id: Optional[str] = None + + +class ImageBlock(BaseModel): + """An image, either inline base64 or a URL to fetch.""" + + type: Literal["image"] + source: ImageSource + + +class ToolUseBlock(BaseModel): + """A tool call the model made on a previous turn, replayed by the client.""" + + type: Literal["tool_use"] + id: str + name: str + input: Dict[str, Any] = Field(default_factory=dict) + + +class ToolResultBlock(BaseModel): + """The result of a tool call, sent back in a user turn.""" + + type: Literal["tool_result"] + tool_use_id: str + + # Anthropic allows a bare string or a list of blocks here + content: Optional[Union[str, List["RequestContentBlock"]]] = None + is_error: Optional[bool] = False + + +class UnsupportedBlock(BaseModel): + """Any block type this server does not handle yet.""" + + model_config = ConfigDict(extra="allow") + + type: str + + +KnownRequestBlock = Annotated[ + Union[ + TextBlock, + ImageBlock, + ThinkingBlock, + RedactedThinkingBlock, + ToolUseBlock, + ToolResultBlock, + ], + Field(discriminator="type"), +] +RequestContentBlock = Union[KnownRequestBlock, UnsupportedBlock] + +ToolResultBlock.model_rebuild() + + +class ToolDefinition(BaseModel): + """ + A tool the model may call. + + Anthropic's server-side tools (web search, code execution) arrive in the + same list distinguished by a `type`, so the field is modelled here in + order to reject them by name rather than fail schema validation. + """ + + model_config = ConfigDict(extra="allow") + + name: Optional[str] = None + description: Optional[str] = None + input_schema: Optional[Dict[str, Any]] = None + type: Optional[str] = None + + +class ToolChoice(BaseModel): + """How the model should choose among the available tools.""" + + model_config = ConfigDict(extra="allow") + + # Not a Literal so an unrecognized mode is reported by name + type: str + name: Optional[str] = None + disable_parallel_tool_use: Optional[bool] = None + + +class AnthropicMessage(BaseModel): + """ + A single turn of the conversation. + + A system role is accepted mid-list: recent Anthropic models take operator + instructions that way rather than by editing the top-level system prompt, + and Claude Code sends them on every request. + """ + + role: Literal["user", "assistant", "system"] + content: Union[str, List[RequestContentBlock]] + + +class ThinkingConfig(BaseModel): + """ + Reasoning configuration. Only the on/off distinction is used; a token + budget has no equivalent in a local chat template. + """ + + model_config = ConfigDict(extra="allow") + + # Not a Literal: the set of thinking types grows over time and an + # unrecognized one should not fail the request + type: str = "enabled" + budget_tokens: Optional[int] = None + + +class Metadata(BaseModel): + """Request metadata. Only user_id is carried over.""" + + model_config = ConfigDict(extra="allow") + + user_id: Optional[str] = None + + +class MessagesRequest(BaseModel): + """Represents an Anthropic Messages request.""" + + messages: List[AnthropicMessage] + + # Required by the Anthropic API, unlike OAI where it's a sampler default + max_tokens: int = Field(..., ge=1) + + # Optional here, unlike the Anthropic API: TabbyAPI serves the loaded + # model when a request doesn't name one + model: Optional[str] = None + + system: Optional[Union[str, List[TextBlock]]] = None + stop_sequences: Optional[List[str]] = None + stream: Optional[bool] = False + + # Bounds are validated against the sampler request these map onto, so a + # local model can be driven outside the ranges the Anthropic API accepts + temperature: Optional[float] = Field(default=None, ge=0) + top_p: Optional[float] = Field(default=None, ge=0, le=1) + top_k: Optional[int] = Field(default=None, ge=0) + + thinking: Optional[ThinkingConfig] = None + metadata: Optional[Metadata] = None + + tools: Optional[List[ToolDefinition]] = None + tool_choice: Optional[ToolChoice] = None + + template_vars: Optional[dict] = Field( + default=None, + validation_alias=AliasChoices("template_vars", "chat_template_kwargs"), + description=( + "TabbyAPI extension, not part of the Messages API. Variables passed " + "to the chat template, as on /v1/chat/completions. Aliases: " + "chat_template_kwargs. Takes precedence over the value derived from " + "the thinking field, and is still overridden by template_vars_force." + ), + ) + + +class CountTokensRequest(BaseModel): + """Represents an Anthropic token counting request.""" + + messages: List[AnthropicMessage] + model: Optional[str] = None + system: Optional[Union[str, List[TextBlock]]] = None + tools: Optional[List[ToolDefinition]] = None + tool_choice: Optional[ToolChoice] = None + + template_vars: Optional[dict] = Field( + default=None, + validation_alias=AliasChoices("template_vars", "chat_template_kwargs"), + description=( + "TabbyAPI extension. Counting has to render the same prompt " + "generation would, so the same variables apply." + ), + ) + + +# Response types + + +class ResponseTextBlock(BaseModel): + """A text block in a response.""" + + type: Literal["text"] = "text" + text: str + + +class ResponseThinkingBlock(BaseModel): + """ + A thinking block in a response. + + The signature is always empty: it authenticates thinking replayed to the + Anthropic API, and there is nothing to authenticate against locally. The + field is emitted anyway because SDK response models require it. + """ + + type: Literal["thinking"] = "thinking" + thinking: str + signature: str = "" + + +class ResponseToolUseBlock(BaseModel): + """A tool call in a response.""" + + type: Literal["tool_use"] = "tool_use" + id: str + name: str + input: Dict[str, Any] = Field(default_factory=dict) + + +ResponseContentBlock = Union[ResponseThinkingBlock, ResponseTextBlock, ResponseToolUseBlock] + + +class Usage(BaseModel): + """Token usage for a response.""" + + input_tokens: int + output_tokens: int + + # Always zero. Emitted because SDK response models expect the fields and + # clients divide by them when reporting cache efficiency. + cache_creation_input_tokens: int = 0 + cache_read_input_tokens: int = 0 + + +class MessagesResponse(BaseModel): + """Represents an Anthropic Messages response.""" + + id: str = Field(default_factory=lambda: f"msg_{uuid4().hex}") + type: Literal["message"] = "message" + role: Literal["assistant"] = "assistant" + content: List[ResponseContentBlock] + model: str + stop_reason: Optional[str] = None + stop_sequence: Optional[str] = None + usage: Usage + + +class CountTokensResponse(BaseModel): + """Represents an Anthropic token counting response.""" + + input_tokens: int diff --git a/endpoints/Anthropic/utils/convert.py b/endpoints/Anthropic/utils/convert.py new file mode 100644 index 00000000..d10b7782 --- /dev/null +++ b/endpoints/Anthropic/utils/convert.py @@ -0,0 +1,499 @@ +"""Translation between Anthropic Messages requests and TabbyAPI's internal +chat completion request. + +Anthropic requests are converted rather than served by a separate inference +path, so the Anthropic API inherits chat templating, reasoning tag parsing, +samplers and context length handling from the chat completion pipeline. +""" + +import json +from typing import List, NamedTuple, Optional, Tuple, Union + +from common import model +from common.logger import xlogger +from endpoints.Anthropic.errors import request_error +from endpoints.Anthropic.types.messages import ( + AnthropicMessage, + CountTokensRequest, + ImageBlock, + MessagesRequest, + RedactedThinkingBlock, + RequestContentBlock, + TextBlock, + ThinkingBlock, + ToolChoice, + ToolDefinition, + ToolResultBlock, + ToolUseBlock, +) +from endpoints.OAI.types.chat_completion import ( + ChatCompletionImageUrl, + ChatCompletionMessage, + ChatCompletionMessagePart, + ChatCompletionRequest, +) +from endpoints.OAI.types.common import ChatCompletionStreamOptions +from endpoints.OAI.types.tools import ( + Function, + NamedToolChoice, + NamedToolFunction, + Tool, + ToolCall, + ToolSpec, +) + +# Anthropic clients routinely split one turn across several text blocks (a +# system prompt and an environment block, say), which are separate pieces of +# context rather than a continuing sentence. Joining on a blank line keeps +# them apart in the flat string a chat template renders. +BLOCK_SEPARATOR = "\n\n" + +# Anthropic tool_choice modes mapped onto their OAI equivalents +TOOL_CHOICE_MODES = {"auto": "auto", "any": "required", "none": "none"} + +# Block types this server handles. A block naming one of these that still fell +# through to the permissive fallback failed its own validation, which is a +# different problem from an unhandled type and worth saying so. +SUPPORTED_BLOCK_TYPES = frozenset( + {"text", "image", "thinking", "redacted_thinking", "tool_use", "tool_result"} +) + + +def _unsupported_block_error(block_type: str, accepted: str, location: str = ""): + """Build the error for a block this server can't turn into content.""" + + if block_type in SUPPORTED_BLOCK_TYPES: + return request_error( + 400, + f"The {block_type} block{location} is missing required fields or has the wrong shape.", + ) + + return request_error( + 400, + f"Content block type '{block_type}'{location} is not supported. This server " + f"accepts {accepted}.", + ) + + +def _system_prompt(system: Optional[Union[str, List[TextBlock]]]) -> Optional[str]: + """Flatten the system field into a single prompt string.""" + + if system is None: + return None + + if isinstance(system, str): + return system or None + + texts = [block.text for block in system if block.text] + + return BLOCK_SEPARATOR.join(texts) or None + + +def _image_url(block: ImageBlock) -> str: + """ + Convert an image source into the URL form the image loader accepts. + + Base64 data becomes a data URL, which is also how the OAI path carries + inline images, so both APIs reach the loader the same way. + """ + + source = block.source + + if source.type == "base64": + if not source.data or not source.media_type: + raise request_error(400, "A base64 image source needs media_type and data.") + + return f"data:{source.media_type};base64,{source.data}" + + if source.type == "url": + if not source.url: + raise request_error(400, "A url image source needs a url.") + + return source.url + + # file sources reference the Anthropic Files API, which has no local + # counterpart to resolve the id against + raise request_error( + 400, + f"Image source type '{source.type}' is not supported. This server accepts " + "base64 and url sources.", + ) + + +def _check_vision_support(): + """ + Reject images unless the loaded model can see them. + + The templating step only builds embeddings for a vision model, so without + this an image would be dropped and the model asked about a picture it was + never shown. + """ + + if not getattr(model.container, "use_vision", False): + raise request_error( + 400, + "The loaded model does not support images. Load a model with vision " + "enabled to send image blocks.", + ) + + +def _content_from_items(items: List[Tuple[str, str]]): + """ + Build message content from ordered (kind, value) items. + + Returns a plain string when there are no images, keeping the common case + the flat string a template renders, and a part list otherwise so the + templating step can turn images into embeddings in place. + """ + + if not items: + return None + + if all(kind == "text" for kind, _ in items): + return BLOCK_SEPARATOR.join(value for _, value in items) or None + + parts: List[ChatCompletionMessagePart] = [] + + for kind, value in items: + if kind == "image": + parts.append( + ChatCompletionMessagePart( + type="image_url", image_url=ChatCompletionImageUrl(url=value) + ) + ) + elif parts and parts[-1].type == "text": + # Keep consecutive text blocks apart, as they are when no image + # splits them + parts[-1].text += BLOCK_SEPARATOR + value + else: + parts.append(ChatCompletionMessagePart(type="text", text=value)) + + return parts + + +def _tool_result_content(block: ToolResultBlock): + """ + Flatten a tool result into content a chat template can render. + + Chat templates have no concept of a failed tool call, so is_error is + folded into the text: the model can only act on the failure if it can + read it. + """ + + items: List[Tuple[str, str]] = [] + + if isinstance(block.content, str): + if block.content: + items.append(("text", block.content)) + elif block.content is not None: + for inner in block.content: + if isinstance(inner, TextBlock): + if inner.text: + items.append(("text", inner.text)) + elif isinstance(inner, ImageBlock): + _check_vision_support() + items.append(("image", _image_url(inner))) + else: + raise _unsupported_block_error( + inner.type, "text and image blocks there", " inside a tool_result" + ) + + if block.is_error: + if items and items[0][0] == "text": + items[0] = ("text", f"Error: {items[0][1]}") + else: + items.insert(0, ("text", "Error")) + + return _content_from_items(items) or "" + + +def _system_turn(content, mid_conversation: bool) -> ChatCompletionMessage: + """ + Turn a system-role message into one a chat template will render. + + Anthropic's mid-conversation system messages carry operator instructions + without disturbing the cached prefix ahead of them. Chat templates almost + universally allow a system turn only in first position — Qwen's raises + "System message must be at the beginning" otherwise — so anything later is + carried in a user turn tagged as a system reminder, the same fallback the + Anthropic API documents for models lacking the feature. + + Folding it into the leading system prompt instead would defeat the point: + that prompt sits at the front of the prefix, so rewriting it every turn + invalidates the whole prompt cache. + """ + + if not mid_conversation: + return ChatCompletionMessage(role="system", content=content) + + if isinstance(content, str): + content = f"\n{content}\n" + else: + xlogger.debug("Non-text system message content; passing through untagged.") + + return ChatCompletionMessage(role="user", content=content) + + +class MessageParts(NamedTuple): + """One Anthropic message, split along the axes a chat message needs.""" + + content: Optional[Union[str, List[ChatCompletionMessagePart]]] + reasoning: Optional[str] + tool_calls: List[ToolCall] + tool_results: List[ToolResultBlock] + + +def _split_content(content: Union[str, List[RequestContentBlock]]) -> MessageParts: + """ + Split one message's content by block kind. + + Raises for block types this server does not handle yet, so an unsupported + request fails with a readable message instead of silently dropping the + part of the conversation the client cared about. + """ + + if isinstance(content, str): + return MessageParts(content or None, None, [], []) + + items: List[Tuple[str, str]] = [] + reasoning_parts: List[str] = [] + tool_calls: List[ToolCall] = [] + tool_results: List[ToolResultBlock] = [] + + for block in content: + if isinstance(block, TextBlock): + if block.text: + items.append(("text", block.text)) + elif isinstance(block, ImageBlock): + _check_vision_support() + items.append(("image", _image_url(block))) + elif isinstance(block, ThinkingBlock): + if block.thinking: + reasoning_parts.append(block.thinking) + elif isinstance(block, RedactedThinkingBlock): + # Encrypted by the Anthropic API and unreadable here. Dropping it + # loses nothing a local model could have used. + continue + elif isinstance(block, ToolUseBlock): + # Templates render OAI-shaped tool calls, whose arguments are a + # JSON string; format_messages_with_template parses it back for + # the templates that want a mapping + tool_calls.append( + ToolCall( + id=block.id, + function=Tool(name=block.name, arguments=json.dumps(block.input)), + ) + ) + elif isinstance(block, ToolResultBlock): + tool_results.append(block) + else: + raise _unsupported_block_error( + block.type, "text, image, thinking, tool_use and tool_result blocks" + ) + + return MessageParts( + _content_from_items(items), + BLOCK_SEPARATOR.join(reasoning_parts) or None, + tool_calls, + tool_results, + ) + + +def build_chat_messages( + system: Optional[Union[str, List[TextBlock]]], + messages: List[AnthropicMessage], +) -> List[ChatCompletionMessage]: + """Convert an Anthropic system prompt and message list for templating.""" + + chat_messages: List[ChatCompletionMessage] = [] + + system_prompt = _system_prompt(system) + if system_prompt: + chat_messages.append(ChatCompletionMessage(role="system", content=system_prompt)) + + for message in messages: + parts = _split_content(message.content) + + if message.role == "system": + chat_messages.append(_system_turn(parts.content, bool(chat_messages))) + continue + + # Anthropic packs every tool result for a turn into one user message, + # but chat templates expect one tool message per result, so a single + # message can fan out. They lead the turn, which is also the order + # Anthropic requires them in. + for result in parts.tool_results: + chat_messages.append( + ChatCompletionMessage( + role="tool", + content=_tool_result_content(result), + tool_call_id=result.tool_use_id, + ) + ) + + # An assistant turn that only called tools carries no text, and a user + # turn of nothing but tool results adds no message of its own + if parts.content or parts.reasoning or parts.tool_calls: + chat_messages.append( + ChatCompletionMessage( + role=message.role, + content=parts.content, + reasoning_content=parts.reasoning, + tool_calls=parts.tool_calls or None, + ) + ) + + return chat_messages + + +def convert_tools(tools: Optional[List[ToolDefinition]]) -> Optional[List[ToolSpec]]: + """ + Convert tool definitions into the OAI shape chat templates render. + + Every tool call format in the pipeline is driven from that shape, so + translating here means the Anthropic API inherits all of them. + """ + + if not tools: + return None + + specs = [] + + for tool in tools: + # Anthropic's own server-side tools run on their infrastructure and + # have no local equivalent + if tool.type and tool.type != "custom": + raise request_error( + 400, + f"Server-side tool '{tool.type}' is not supported. This server only " + "serves client-defined tools.", + ) + + if not tool.name or tool.input_schema is None: + raise request_error(400, "Each tool needs a name and an input_schema.") + + specs.append( + ToolSpec( + type="function", + function=Function( + name=tool.name, + description=tool.description or "", + parameters=tool.input_schema, + ), + ) + ) + + return specs + + +def convert_tool_choice(choice: Optional[ToolChoice]): + """ + Convert tool_choice, returning the choice and the parallel call setting. + + Returns (None, None) when the request left it unset, so the pipeline's own + defaults apply. + """ + + if choice is None: + return None, None + + parallel = None + if choice.disable_parallel_tool_use is not None: + parallel = not choice.disable_parallel_tool_use + + if choice.type == "tool": + if not choice.name: + raise request_error(400, "A tool_choice of type 'tool' needs a tool name.") + + return NamedToolChoice(function=NamedToolFunction(name=choice.name)), parallel + + mode = TOOL_CHOICE_MODES.get(choice.type) + if mode is None: + raise request_error( + 400, + f"Unknown tool_choice type '{choice.type}'. Expected auto, any, tool or none.", + ) + + return mode, parallel + + +def _sampler_params(data: MessagesRequest) -> dict: + """ + Collect the sampler fields the request actually set. + + Unset fields are left out entirely so the model's sampler defaults and + any configured overrides still apply. + """ + + params = {"max_tokens": data.max_tokens} + + if data.stop_sequences: + params["stop"] = list(data.stop_sequences) + if data.temperature is not None: + params["temperature"] = data.temperature + if data.top_p is not None: + params["top_p"] = data.top_p + if data.top_k is not None: + params["top_k"] = data.top_k + if data.metadata and data.metadata.user_id: + params["user"] = data.metadata.user_id + + return params + + +def convert_messages_request(data: MessagesRequest) -> ChatCompletionRequest: + """Convert an Anthropic Messages request into a chat completion request.""" + + template_vars = {} + if data.thinking is not None: + template_vars["enable_thinking"] = data.thinking.type != "disabled" + + if data.thinking.budget_tokens is not None: + # Thinking length is a property of the model and its template + # here, not something a request can allocate + xlogger.debug("thinking.budget_tokens is not supported; ignoring.") + + # Explicit variables win over the one derived from thinking, matching how + # the chat completion path ranks template_vars above its own flat fields. + # Both still lose to the model's template_vars_force. + template_vars.update(data.template_vars or {}) + + tool_choice, parallel_tool_calls = convert_tool_choice(data.tool_choice) + + optional = {} + if parallel_tool_calls is not None: + optional["parallel_tool_calls"] = parallel_tool_calls + + return ChatCompletionRequest( + messages=build_chat_messages(data.system, data.messages), + model=data.model, + template_vars=template_vars, + tools=convert_tools(data.tools), + tool_choice=tool_choice, + **optional, + # Anthropic returns usage on every response, and the chat completion + # pipeline only assembles it when asked + stream_options=ChatCompletionStreamOptions(include_usage=True), + # The Messages API has no multi-choice concept + n=1, + **_sampler_params(data), + ) + + +def convert_count_tokens_request(data: CountTokensRequest) -> ChatCompletionRequest: + """ + Convert a token counting request into a chat completion request. + + max_tokens is a placeholder: nothing is generated, but the prompt is + rendered through the same path a real request would take so the count + reflects what generation would actually see. + """ + + return ChatCompletionRequest( + messages=build_chat_messages(data.system, data.messages), + model=data.model, + tools=convert_tools(data.tools), + template_vars=data.template_vars or {}, + max_tokens=1, + n=1, + ) diff --git a/endpoints/Anthropic/utils/messages.py b/endpoints/Anthropic/utils/messages.py new file mode 100644 index 00000000..798c1406 --- /dev/null +++ b/endpoints/Anthropic/utils/messages.py @@ -0,0 +1,164 @@ +"""Message utilities for the Anthropic server.""" + +import json +from typing import List, Optional, Tuple + +from common import model +from common.logger import xlogger +from common.utils import unwrap +from endpoints.Anthropic.types.messages import ( + CountTokensRequest, + CountTokensResponse, + MessagesRequest, + MessagesResponse, + ResponseContentBlock, + ResponseTextBlock, + ResponseThinkingBlock, + ResponseToolUseBlock, + Usage, +) +from endpoints.Anthropic.utils.convert import convert_count_tokens_request +from endpoints.OAI.types.chat_completion import ChatCompletionResponse +from endpoints.OAI.utils.chat_completion import apply_chat_template + + +def tool_call_input(name: str, arguments: str) -> dict: + """ + Parse tool call arguments into the object a tool_use block carries. + + The pipeline hands back arguments as a JSON string, per the OAI shape the + tool call parsers emit. A parser that produced something unparseable would + otherwise fail the whole response, so the call is surfaced with empty + input and a warning instead: the tool name is the useful part. + """ + + try: + parsed = json.loads(arguments) + except (json.JSONDecodeError, TypeError): + parsed = None + + if not isinstance(parsed, dict): + xlogger.warning( + "Tool call arguments could not be parsed into an object", + {"name": name, "arguments": arguments}, + ) + + return {} + + return parsed + + +def usage_from_stats(usage) -> Usage: + """ + Split prompt tokens into fresh and cache-read, as Anthropic counts them. + + TabbyAPI's prompt_tokens is the whole prompt with cached_tokens the part + the prefix cache served, while Anthropic's input_tokens counts only what + was not read from cache. Reporting the whole prompt as input made every + replayed turn look freshly processed, which is what inflates a client's + cost estimate over a long conversation. + + cache_creation stays zero: the backend does not distinguish writing to the + cache from ordinary prefill, and clients price cache writes above plain + input, so guessing there would overstate rather than understate. + """ + + if usage is None: + return Usage(input_tokens=0, output_tokens=0) + + prompt_tokens = usage.prompt_tokens or 0 + + # Defensive: a cached count above the prompt length would make input + # tokens negative + cached_tokens = min(usage.cached_tokens or 0, prompt_tokens) + + return Usage( + input_tokens=prompt_tokens - cached_tokens, + output_tokens=usage.completion_tokens or 0, + cache_read_input_tokens=cached_tokens, + ) + + +def stop_reason( + finish_reason: Optional[str], + eos_reason: Optional[str], + stop_str: Optional[str], + stop_sequences: Optional[List[str]], +) -> Tuple[str, Optional[str]]: + """ + Map a finished generation onto an Anthropic stop reason. + + Takes the raw fields rather than a response object so the streaming and + non-streaming paths share one implementation; a stop reason that differs + between them is a class of bug worth designing out. + + The stop_sequence field only reports sequences the client asked for. A + prompt template contributes its own stop strings, and surfacing one of + those as a stop_sequence would name a string the client never sent. + """ + + if finish_reason == "tool_calls": + return "tool_use", None + + if finish_reason == "length": + return "max_tokens", None + + if eos_reason == "stop_string" and stop_str in (stop_sequences or []): + return "stop_sequence", stop_str + + return "end_turn", None + + +def convert_response( + completion: ChatCompletionResponse, + data: MessagesRequest, + model_name: str, +) -> MessagesResponse: + """Convert a chat completion response into an Anthropic message.""" + + choice = completion.choices[0] + message = choice.message + + # Reasoning precedes the answer it produced, and tool calls follow the + # text introducing them, matching the block order the Anthropic API emits + content: List[ResponseContentBlock] = [] + if message.reasoning_content: + content.append(ResponseThinkingBlock(thinking=message.reasoning_content)) + if message.content: + content.append(ResponseTextBlock(text=message.content)) + for tool_call in message.tool_calls or []: + content.append( + ResponseToolUseBlock( + id=tool_call.id, + name=tool_call.function.name, + input=tool_call_input(tool_call.function.name, tool_call.function.arguments), + ) + ) + + reason, stop_sequence = stop_reason( + choice.finish_reason, choice.eos_reason, choice.stop_str, data.stop_sequences + ) + + return MessagesResponse( + content=content, + model=model_name, + stop_reason=reason, + stop_sequence=stop_sequence, + usage=usage_from_stats(completion.usage), + ) + + +async def count_tokens(data: CountTokensRequest) -> CountTokensResponse: + """ + Count the tokens an equivalent Messages request would consume. + + The prompt is rendered through the same template path as generation, so + the count includes the template's own structure and generation prompt. + """ + + converted = convert_count_tokens_request(data) + prompt, mm_embeddings = await apply_chat_template(converted) + + raw_tokens = model.container.encode_tokens(prompt, embeddings=mm_embeddings) + + return CountTokensResponse(input_tokens=len(unwrap(raw_tokens, []))) diff --git a/endpoints/Anthropic/utils/stream.py b/endpoints/Anthropic/utils/stream.py new file mode 100644 index 00000000..627a4dab --- /dev/null +++ b/endpoints/Anthropic/utils/stream.py @@ -0,0 +1,380 @@ +"""Streaming translation for the Anthropic Messages API. + +Anthropic's stream is a block-structured event sequence, not the flat chunk +stream the OAI API emits. Each event carries an SSE event name, and content +arrives inside explicitly opened and closed blocks: + + message_start + content_block_start -> content_block_delta* -> content_block_stop + ... + message_delta (stop reason and output token count) + message_stop + +The generation pipeline emits reasoning and content as separate deltas on the +same chunk, so translating means tracking which block is open and closing it +when the channel changes. Exactly one block is open at a time and indices are +assigned in the order blocks are opened. + +There is no [DONE] sentinel: message_stop terminates the stream. +""" + +import asyncio +import json +import pathlib +from asyncio import CancelledError +from typing import List, Optional +from uuid import uuid4 + +from fastapi import Request +from sse_starlette import ServerSentEvent + +from common.errors import ContextLengthExceededError +from common.logger import xlogger +from common.multimodal import MultimodalEmbeddingWrapper +from common.networking import DisconnectHandler +from endpoints.Anthropic.errors import error_content +from endpoints.Anthropic.types.messages import MessagesRequest, Usage +from endpoints.Anthropic.utils.messages import stop_reason, usage_from_stats +from endpoints.OAI.types.chat_completion import ChatCompletionRequest +from endpoints.OAI.utils.chat_completion import ( + _chat_stream_collector, + _resolve_start_in_reasoning, +) +from endpoints.OAI.utils.common_ import get_usage_stats + +# Block kinds, matching the content block types they open +TEXT = "text" +THINKING = "thinking" +TOOL_USE = "tool_use" + + +def _event(name: str, payload: dict) -> ServerSentEvent: + """Build a named SSE event carrying a JSON payload.""" + + return ServerSentEvent(data=json.dumps(payload, ensure_ascii=False), event=name) + + +def _empty_block(block_kind: str) -> dict: + """ + Build the empty content block a content_block_start announces. + + The thinking block carries an empty signature for the same reason the + non-streaming path does: there is nothing to authenticate locally, but SDK + response models require the field. + """ + + if block_kind == THINKING: + return {"type": THINKING, "thinking": "", "signature": ""} + + return {"type": TEXT, "text": ""} + + +def _delta(block_kind: str, text: str) -> dict: + """Build the delta payload for a block kind.""" + + if block_kind == THINKING: + return {"type": "thinking_delta", "thinking": text} + + return {"type": "text_delta", "text": text} + + +class ContentBlockTracker: + """ + Tracks the open content block and hands out block indices. + + Blocks are opened lazily: a kind that never produces text never appears in + the stream, so a response without reasoning has its text at index 0. + """ + + def __init__(self): + self.open_kind: Optional[str] = None + self.index = -1 + + def _open(self, block_kind: str) -> ServerSentEvent: + self.index += 1 + self.open_kind = block_kind + + return _event( + "content_block_start", + { + "type": "content_block_start", + "index": self.index, + "content_block": _empty_block(block_kind), + }, + ) + + def close(self) -> List[ServerSentEvent]: + """Close the open block, if any.""" + + if self.open_kind is None: + return [] + + self.open_kind = None + + return [_event("content_block_stop", {"type": "content_block_stop", "index": self.index})] + + def write(self, block_kind: str, text: str) -> List[ServerSentEvent]: + """Emit text into a block of the given kind, switching blocks if needed.""" + + if not text: + return [] + + events: List[ServerSentEvent] = [] + + if self.open_kind != block_kind: + events += self.close() + events.append(self._open(block_kind)) + + events.append( + _event( + "content_block_delta", + { + "type": "content_block_delta", + "index": self.index, + "delta": _delta(block_kind, text), + }, + ) + ) + + return events + + def write_tool_call(self, tool_call: dict) -> List[ServerSentEvent]: + """ + Emit a complete tool call as its own block. + + The pipeline parses tool calls only once the generation has finished, + so the arguments arrive whole rather than incrementally. They are sent + as a single input_json_delta, which is what an accumulating client + expects: it concatenates the fragments and parses the result. + """ + + function = tool_call.get("function") or {} + events = self.close() + + self.index += 1 + self.open_kind = TOOL_USE + + events.append( + _event( + "content_block_start", + { + "type": "content_block_start", + "index": self.index, + "content_block": { + "type": TOOL_USE, + "id": tool_call.get("id"), + "name": function.get("name"), + "input": {}, + }, + }, + ) + ) + events.append( + _event( + "content_block_delta", + { + "type": "content_block_delta", + "index": self.index, + "delta": { + "type": "input_json_delta", + "partial_json": function.get("arguments") or "{}", + }, + }, + ) + ) + + return events + self.close() + + +def _message_start(message_id: str, model_name: str, input_tokens: int) -> ServerSentEvent: + """Build the opening event, which carries the prompt token count.""" + + return _event( + "message_start", + { + "type": "message_start", + "message": { + "id": message_id, + "type": "message", + "role": "assistant", + "content": [], + "model": model_name, + "stop_reason": None, + "stop_sequence": None, + "usage": { + "input_tokens": input_tokens, + "output_tokens": 0, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 0, + }, + }, + }, + ) + + +def _message_delta(reason: str, stop_sequence: Optional[str], usage: Usage) -> ServerSentEvent: + """ + Build the closing metadata event. + + The whole usage object is repeated here, not just the output count: the + prefix cache split is only known once prefill has run, so message_start + could not carry it. + """ + + return _event( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": reason, "stop_sequence": stop_sequence}, + "usage": usage.model_dump(), + }, + ) + + +def _error_event(message: str, error_type: str = "api_error") -> ServerSentEvent: + """Build an error event for a stream that has already committed HTTP 200.""" + + return _event("error", error_content(message, error_type)) + + +async def _next_generation(gen_queue: asyncio.Queue, gen_task: asyncio.Task): + """ + Await the next chunk, or return None once the collector is finished. + + The collector normally ends the stream by emitting a finish chunk, but a + plain return would otherwise leave this generator awaiting a chunk that + will never arrive, holding the connection and the model slot open. Racing + the queue against the task bounds that wait. + """ + + getter = asyncio.create_task(gen_queue.get()) + await asyncio.wait({getter, gen_task}, return_when=asyncio.FIRST_COMPLETED) + + if getter.done(): + return getter.result() + + # The collector finished; take anything it queued on the way out. Nothing + # was consumed from the queue, so cancelling the getter cannot drop a chunk + getter.cancel() + + return gen_queue.get_nowait() if not gen_queue.empty() else None + + +async def stream_generate_message( + prompt: str, + embeddings: MultimodalEmbeddingWrapper, + data: MessagesRequest, + converted: ChatCompletionRequest, + request: Request, + model_path: pathlib.Path, + disconnect_handler: DisconnectHandler, + input_tokens: int, +): + """Generator translating the generation stream into Anthropic events.""" + + gen_queue = asyncio.Queue() + gen_task: Optional[asyncio.Task] = None + message_id = f"msg_{uuid4().hex}" + model_name = model_path.name + + try: + xlogger.info( + f"Received Anthropic streaming request {request.state.id}", + { + "prompt": prompt, + "data": data.model_dump(mode="json"), + "model_path": str(model_path), + }, + ) + + start_in_reasoning_mode = _resolve_start_in_reasoning(prompt, converted) + + # The Messages API has no multi-choice concept, so there is exactly + # one collector and no need to track which choice a chunk belongs to + gen_task = asyncio.create_task( + _chat_stream_collector( + 0, + gen_queue, + request.state.id, + prompt, + converted, + start_in_reasoning_mode, + mm_embeddings=embeddings, + streaming_mode=True, + disconnect_handler=disconnect_handler, + ) + ) + + yield _message_start(message_id, model_name, input_tokens) + + blocks = ContentBlockTracker() + reason = "end_turn" + stop_sequence = None + + # Stands in until the finish chunk reports what prefill actually did + final_usage = Usage(input_tokens=input_tokens, output_tokens=0) + + while True: + generation = await _next_generation(gen_queue, gen_task) + + # The collector finished without a finish chunk; close out the + # stream with what arrived rather than waiting on nothing + if generation is None: + break + + # The collector pushes an exception to the queue if it fails + if isinstance(generation, Exception): + raise generation + + for event in blocks.write(THINKING, generation.get("delta_reasoning_content") or ""): + yield event + for event in blocks.write(TEXT, generation.get("delta_content") or ""): + yield event + + # Tool calls are parsed once generation finishes, so they arrive + # whole on the finish chunk rather than as incremental deltas + for tool_call in generation.get("delta_tool_calls") or []: + for event in blocks.write_tool_call(tool_call): + yield event + + finish_reason = generation.get("finish_reason") + if finish_reason: + reason, stop_sequence = stop_reason( + finish_reason, + generation.get("eos_reason"), + generation.get("stop_str"), + data.stop_sequences, + ) + + # The finish chunk is authoritative: it knows the output + # count and how much of the prompt the prefix cache served + usage = get_usage_stats(generation) + if usage: + final_usage = usage_from_stats(usage) + + break + + for event in blocks.close(): + yield event + + yield _message_delta(reason, stop_sequence, final_usage) + yield _event("message_stop", {"type": "message_stop"}) + + xlogger.info(f"Finished Anthropic streaming request {request.state.id}") + + except CancelledError: + raise + + except ContextLengthExceededError as exc: + yield _error_event(str(exc), "invalid_request_error") + + except Exception as exc: + xlogger.error("Error during Anthropic message stream", str(exc), details=f"\n{str(exc)}") + yield _error_event("Message generation aborted. Please check the server console.") + + finally: + # A client that hangs up mid-stream leaves the collector running + if gen_task is not None and not gen_task.done(): + gen_task.cancel() + + await disconnect_handler.cleanup() diff --git a/endpoints/OAI/types/common.py b/endpoints/OAI/types/common.py index 6737bea1..815bd53f 100644 --- a/endpoints/OAI/types/common.py +++ b/endpoints/OAI/types/common.py @@ -10,6 +10,11 @@ class UsageStats(BaseModel): """Represents usage stats.""" prompt_tokens: int + + # The part of prompt_tokens served from the prefix cache rather than + # processed. prompt_tokens is the whole prompt, so the two are not additive. + cached_tokens: Optional[int] = None + prompt_time: Optional[float] = None prompt_tokens_per_sec: Optional[Union[float, str]] = None completion_tokens: int diff --git a/endpoints/OAI/utils/common_.py b/endpoints/OAI/utils/common_.py index 3a68ea8c..a9855cee 100644 --- a/endpoints/OAI/utils/common_.py +++ b/endpoints/OAI/utils/common_.py @@ -21,6 +21,7 @@ def get_usage_stats( completion_tokens = generation.get("gen_tokens", 0) usage_stats = UsageStats( prompt_tokens=prompt_tokens, + cached_tokens=int(generation.get("cached_tokens") or 0), prompt_time=generation.get("prompt_time"), prompt_tokens_per_sec=generation.get("prompt_tokens_per_sec"), completion_tokens=completion_tokens, @@ -38,6 +39,8 @@ def aggregate_usage_stats(usage_stats_list: list[UsageStats]) -> UsageStats: usl = usage_stats_list prompt_tokens = usl[0].prompt_tokens + # Every choice shares one prompt, so the cached portion is shared too + cached_tokens = usl[0].cached_tokens prompt_time = usl[0].prompt_time prompt_tokens_per_sec = usl[0].prompt_tokens_per_sec completion_tokens = sum(us.completion_tokens for us in usl) @@ -48,6 +51,7 @@ def aggregate_usage_stats(usage_stats_list: list[UsageStats]) -> UsageStats: usage_stats = UsageStats( prompt_tokens=prompt_tokens, + cached_tokens=cached_tokens, prompt_time=prompt_time, prompt_tokens_per_sec=prompt_tokens_per_sec, completion_tokens=completion_tokens, diff --git a/endpoints/core/types/model.py b/endpoints/core/types/model.py index 84229294..836760eb 100644 --- a/endpoints/core/types/model.py +++ b/endpoints/core/types/model.py @@ -1,6 +1,7 @@ """Contains model card types.""" -from pydantic import BaseModel, Field, ConfigDict +from datetime import datetime, timezone +from pydantic import BaseModel, Field, ConfigDict, computed_field from time import time from typing import List, Literal, Optional, Union @@ -29,7 +30,14 @@ class ModelCardParameters(BaseModel): class ModelCard(BaseModel): - """Represents a single model card.""" + """ + Represents a single model card. + + Carries the OpenAI fields (object, created) alongside the Anthropic ones + (type, display_name, created_at), which name the same things differently. + Anthropic SDK model types require theirs, and serving both keeps one + listing usable by either client. + """ id: str = "test" object: str = "model" @@ -38,6 +46,43 @@ class ModelCard(BaseModel): logging: Optional[LoggingConfig] = None parameters: Optional[ModelCardParameters] = None + # Anthropic aliases, filled from the fields above when not set + type: str = "model" + display_name: Optional[str] = None + created_at: Optional[str] = None + + def model_post_init(self, __context): + if self.display_name is None: + self.display_name = self.id + + if self.created_at is None: + self.created_at = ( + datetime.fromtimestamp(self.created, tz=timezone.utc) + .isoformat() + .replace("+00:00", "Z") + ) + + @computed_field + @property + def max_input_tokens(self) -> Optional[int]: + """The context window under the name the Anthropic API gives it.""" + + return self.parameters.max_seq_len if self.parameters else None + + @computed_field + @property + def max_tokens(self) -> Optional[int]: + """ + The output ceiling, which is the context window here. + + TabbyAPI caps a generation by the context minus the prompt rather than + by a separate output limit, so the context length is the honest upper + bound. The field is served because Anthropic's model schema carries it + beside max_input_tokens and a client reading one expects the other. + """ + + return self.parameters.max_seq_len if self.parameters else None + class ModelList(BaseModel): """Represents a list of model cards.""" @@ -45,6 +90,21 @@ class ModelList(BaseModel): object: str = "list" data: List[ModelCard] = Field(default_factory=list) + # Anthropic pagination fields. TabbyAPI serves the whole list at once, so + # there is never another page. The ids are computed rather than stored + # because callers build the list empty and append to data afterwards. + has_more: bool = False + + @computed_field + @property + def first_id(self) -> Optional[str]: + return self.data[0].id if self.data else None + + @computed_field + @property + def last_id(self) -> Optional[str]: + return self.data[-1].id if self.data else None + class DraftModelLoadRequest(BaseModel): """Represents a draft model load request.""" diff --git a/endpoints/core/utils/model.py b/endpoints/core/utils/model.py index 6e93e1e6..2b499063 100644 --- a/endpoints/core/utils/model.py +++ b/endpoints/core/utils/model.py @@ -8,12 +8,33 @@ from common.tabby_config import config from endpoints.core.types.model import ( ModelCard, + ModelCardParameters, ModelList, ModelLoadRequest, ModelLoadResponse, ) +def get_listed_model_params() -> Optional[ModelCardParameters]: + """ + Parameters of the loaded model as a listing carries them. + + A listing is how a client discovers the context window, since neither the + OpenAI nor the Anthropic model schema has a field for it. The prompt + template content is dropped: it runs to kilobytes and nothing reading a + listing needs it, so it stays exclusive to /v1/model. + """ + + if model.container is None: + return None + + params = model.container.model_info().parameters + if params is None: + return None + + return params.model_copy(update={"prompt_template_content": None}) + + def get_model_list(model_path: pathlib.Path, draft_model_path: Optional[str] = None): """Get the list of models from the provided path.""" @@ -22,11 +43,18 @@ def get_model_list(model_path: pathlib.Path, draft_model_path: Optional[str] = N if draft_model_path: draft_model_path = pathlib.Path(draft_model_path).resolve() + loaded_model_path = model.container.model_dir.resolve() if model.container else None + model_card_list = ModelList() for path in model_path.iterdir(): # Don't include the draft models path if path.is_dir() and path != draft_model_path: model_card = ModelCard(id=path.name) + + # Only the loaded model has parameters to report + if loaded_model_path and path.resolve() == loaded_model_path: + model_card.parameters = get_listed_model_params() + model_card_list.data.append(model_card) # pylint: disable=no-member return model_card_list @@ -55,7 +83,13 @@ async def get_current_model_list(model_type: str = "model"): model_path = model.embeddings_container.model_dir if model_path: - current_models.append(ModelCard(id=model_path.name)) + model_card = ModelCard(id=model_path.name) + + # Draft and embedding cards would report the main model's parameters + if model_type == "model": + model_card.parameters = get_listed_model_params() + + current_models.append(model_card) return ModelList(data=current_models) diff --git a/endpoints/server.py b/endpoints/server.py index b3ed1e4a..d943203d 100644 --- a/endpoints/server.py +++ b/endpoints/server.py @@ -10,6 +10,7 @@ from common.errors import ContextLengthHTTPException, context_length_exception_handler from common.networking import get_global_depends from common.tabby_config import config +from endpoints.Anthropic import router as AnthropicRouter from endpoints.Kobold import router as KoboldRouter from endpoints.OAI import router as OAIRouter from endpoints.core.router import router as CoreRouter @@ -48,7 +49,11 @@ def setup_app(host: Optional[str] = None, port: Optional[int] = None): ) # Map for API id to server router - router_mapping = {"oai": OAIRouter, "kobold": KoboldRouter} + router_mapping = { + "oai": OAIRouter, + "kobold": KoboldRouter, + "anthropic": AnthropicRouter, + } # Include the OAI api by default for server in api_servers: diff --git a/tests/req_anthropic.py b/tests/req_anthropic.py new file mode 100644 index 00000000..46263920 --- /dev/null +++ b/tests/req_anthropic.py @@ -0,0 +1,305 @@ +""" +Manual checks for the Anthropic Messages API against a running server. + +Enable the API first, in config.yml: + + network: + api_servers: ["OAI", "Anthropic"] + +The requests are sent with the Anthropic header and auth conventions +(x-api-key, anthropic-version) so this also exercises the header path real +SDK clients use. +""" + +import json +from pprint import pprint + +import httpx + +from _common import load_api_keys + +BASE_URL = "http://localhost:5000/v1" +MODEL = "/mnt/str/models/qwen3.5-35b-a3b/exl3/4.09bpw/" + +simple_request = { + "model": MODEL, + "max_tokens": 512, + "system": "You are a concise assistant.", + "messages": [{"role": "user", "content": "Name three primary colors."}], +} + +block_request = { + "model": MODEL, + "max_tokens": 512, + # Clients routinely split the system prompt across blocks + "system": [ + { + "type": "text", + "text": "You are a concise assistant.", + "cache_control": {"type": "ephemeral"}, + }, + {"type": "text", "text": "The user is testing an API shim."}, + ], + "messages": [ + {"role": "user", "content": [{"type": "text", "text": "Say hello."}]}, + { + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "A greeting was requested.", "signature": "x"}, + {"type": "text", "text": "Hello!"}, + ], + }, + {"role": "user", "content": "Now say goodbye."}, + ], +} + +stop_sequence_request = { + "model": MODEL, + "max_tokens": 512, + "stop_sequences": ["END"], + "messages": [ + { + "role": "user", + "content": "Count from 1 to 10, one number per line, then write END.", + } + ], +} + + +tool_request = { + "model": MODEL, + "max_tokens": 512, + "tools": [ + { + "name": "get_weather", + "description": "Get the current weather for a location.", + "input_schema": { + "type": "object", + "properties": {"location": {"type": "string", "description": "City name"}}, + "required": ["location"], + }, + } + ], + "tool_choice": {"type": "auto"}, + "messages": [{"role": "user", "content": "What's the weather in Paris and London?"}], +} + +# A second turn feeding results back, which exercises the tool_result fan-out +tool_followup_request = { + "model": MODEL, + "max_tokens": 512, + "tools": tool_request["tools"], + "messages": [ + {"role": "user", "content": "What's the weather in Paris and London?"}, + { + "role": "assistant", + "content": [ + {"type": "text", "text": "Let me check both."}, + { + "type": "tool_use", + "id": "toolu_1", + "name": "get_weather", + "input": {"location": "Paris"}, + }, + { + "type": "tool_use", + "id": "toolu_2", + "name": "get_weather", + "input": {"location": "London"}, + }, + ], + }, + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "toolu_1", "content": "21C, sunny"}, + { + "type": "tool_result", + "tool_use_id": "toolu_2", + "content": "station offline", + "is_error": True, + }, + {"type": "text", "text": "Summarise what you found."}, + ], + }, + ], +} + + +# A 1x1 red PNG, enough to exercise the base64 path without a fixture file +RED_PIXEL_PNG = ( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmM" + "IQAAAABJRU5ErkJggg==" +) + +image_request = { + "model": MODEL, + "max_tokens": 256, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": RED_PIXEL_PNG, + }, + }, + {"type": "text", "text": "What colour is this image?"}, + ], + } + ], +} + + +def post(api_key, path, request): + return httpx.post( + f"{BASE_URL}{path}", + headers={"x-api-key": api_key, "anthropic-version": "2023-06-01"}, + json=request, + timeout=300, + ) + + +def test_message(api_key, request, label): + print("\n\n") + print("-" * 80) + print(f"MESSAGES REQUEST: {label}") + print("-" * 80) + + data = post(api_key, "/messages", request).json() + pprint(data, width=160) + + if data.get("type") == "error": + return data + + for block in data.get("content", []): + if block["type"] == "thinking": + print(f"\n[thinking]\n{block['thinking']}") + elif block["type"] == "text": + print(f"\n[text]\n{block['text']}") + elif block["type"] == "tool_use": + print(f"\n[tool_use] [{block['id']}] {block['name']}({json.dumps(block['input'])})") + + print(f"\nStop reason: {data.get('stop_reason')} (sequence: {data.get('stop_sequence')})") + print(f"Usage: {data.get('usage')}") + + return data + + +def test_message_streaming(api_key, request, label): + print("\n\n") + print("-" * 80) + print(f"STREAMING MESSAGES REQUEST: {label}") + print("-" * 80) + + request = {**request, "stream": True} + + # Accumulate the way an SDK does, so a malformed block lifecycle shows up + blocks = {} + order = [] + final = {} + event_names = [] + + with httpx.stream( + "POST", + f"{BASE_URL}/messages", + headers={"x-api-key": api_key, "anthropic-version": "2023-06-01"}, + json=request, + timeout=300, + ) as response: + name = None + for line in response.iter_lines(): + line = line.rstrip("\r") + if line.startswith("event:"): + name = line[len("event:") :].strip() + event_names.append(name) + continue + if not line.startswith("data:"): + continue + + payload = json.loads(line[len("data:") :].strip()) + + if name == "content_block_start": + block = dict(payload["content_block"]) + if block["type"] == "tool_use": + # input arrives as JSON fragments to concatenate + block["_json"] = "" + blocks[payload["index"]] = block + order.append(payload["index"]) + label = block["type"] + if label == "tool_use": + label += f" {block['name']}" + print(f"\n\n[{label}][{payload['index']}]") + elif name == "content_block_delta": + delta = payload["delta"] + if delta["type"] == "input_json_delta": + blocks[payload["index"]]["_json"] += delta["partial_json"] + print(delta["partial_json"], end="", flush=True) + else: + key = "thinking" if delta["type"] == "thinking_delta" else "text" + blocks[payload["index"]][key] += delta[key] + print(delta[key], end="", flush=True) + elif name == "content_block_stop": + block = blocks[payload["index"]] + if block["type"] == "tool_use": + block["input"] = json.loads(block.pop("_json") or "{}") + elif name == "message_delta": + final = payload + elif name == "error": + print(f"\n\n[error] {payload}") + + print(f"\n\nEvent order: {event_names}") + print(f"Accumulated blocks: {[blocks[i] for i in order]}") + print(f"Stop: {final.get('delta')}") + print(f"Usage: {final.get('usage')}") + + return blocks + + +def test_count_tokens(api_key, request, label): + print("\n\n") + print("-" * 80) + print(f"COUNT TOKENS REQUEST: {label}") + print("-" * 80) + + counted = {key: request[key] for key in ("model", "system", "messages") if key in request} + data = post(api_key, "/messages/count_tokens", counted).json() + pprint(data, width=160) + + return data + + +def main(): + api_key, _ = load_api_keys() + + test_message(api_key, simple_request.copy(), "plain text") + test_message(api_key, block_request.copy(), "content blocks and replayed thinking") + test_message(api_key, stop_sequence_request.copy(), "client stop sequence") + + test_message(api_key, tool_request.copy(), "tool call") + test_message(api_key, tool_followup_request.copy(), "tool results fed back") + + test_message_streaming(api_key, simple_request.copy(), "plain text") + test_message_streaming(api_key, block_request.copy(), "content blocks") + test_message_streaming(api_key, stop_sequence_request.copy(), "client stop sequence") + test_message_streaming(api_key, tool_request.copy(), "tool call") + + test_count_tokens(api_key, simple_request, "plain text") + test_count_tokens(api_key, block_request, "content blocks") + + # Needs a vision model loaded; errors cleanly otherwise + test_message(api_key, image_request.copy(), "base64 image") + + # Unsupported blocks must fail loudly rather than drop conversation + unsupported = simple_request.copy() + unsupported["messages"] = [ + {"role": "user", "content": [{"type": "document", "source": {"type": "base64"}}]} + ] + test_message(api_key, unsupported, "unsupported block (expects an error envelope)") + + +if __name__ == "__main__": + main() diff --git a/tests/test_anthropic_messages.py b/tests/test_anthropic_messages.py new file mode 100644 index 00000000..eb3d078d --- /dev/null +++ b/tests/test_anthropic_messages.py @@ -0,0 +1,1572 @@ +import asyncio +import json +import pathlib +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from fastapi import HTTPException +from fastapi.exceptions import RequestValidationError +from sse_starlette import EventSourceResponse + +import common.model # noqa: F401 - resolve import cycle ordering +from common.tabby_config import config +from endpoints.Anthropic.errors import AnthropicHTTPException, exception_to_response +from endpoints.Anthropic.router import count_tokens_request, messages_request as messages_endpoint +from endpoints.Anthropic.types.messages import ( + CountTokensRequest, + MessagesRequest, + ResponseTextBlock, + ResponseThinkingBlock, + ResponseToolUseBlock, +) +from endpoints.Anthropic.utils.convert import ( + _sampler_params, + convert_count_tokens_request, + convert_messages_request, +) +from endpoints.Anthropic.utils.messages import ( + convert_response, + stop_reason, + tool_call_input, + usage_from_stats, +) +from endpoints.Anthropic.utils.stream import ContentBlockTracker, stream_generate_message +from endpoints.OAI.types.chat_completion import ( + ChatCompletionMessage, + ChatCompletionRespChoice, + ChatCompletionResponse, +) +from endpoints.OAI.types.common import UsageStats +from endpoints.OAI.utils.chat_completion import resolve_template_vars +from endpoints.OAI.utils.common_ import aggregate_usage_stats, get_usage_stats +from endpoints.OAI.types.tools import Tool, ToolCall + +MODEL_DIR = pathlib.Path("/models/test-model") + + +def messages_request(**kwargs): + kwargs.setdefault("messages", [{"role": "user", "content": "hi"}]) + kwargs.setdefault("max_tokens", 64) + return MessagesRequest(**kwargs) + + +def choice( + content="Hello", + reasoning_content=None, + finish_reason="stop", + eos_reason="stop_token", + stop_str=None, +): + return ChatCompletionRespChoice( + finish_reason=finish_reason, + eos_reason=eos_reason, + stop_str=stop_str, + message=ChatCompletionMessage( + role="assistant", + content=content, + reasoning_content=reasoning_content, + ), + ) + + +def completion(usage=None, **kwargs): + return ChatCompletionResponse( + choices=[choice(**kwargs)], + model="test-model", + usage=usage, + ) + + +class SystemPromptTests(unittest.TestCase): + def test_string_system(self): + converted = convert_messages_request(messages_request(system="You are helpful.")) + self.assertEqual(converted.messages[0].role, "system") + self.assertEqual(converted.messages[0].content, "You are helpful.") + + def test_block_list_system_joined_on_blank_line(self): + converted = convert_messages_request( + messages_request( + system=[ + {"type": "text", "text": "You are helpful."}, + {"type": "text", "text": "cwd=/tmp"}, + ] + ) + ) + self.assertEqual(converted.messages[0].content, "You are helpful.\n\ncwd=/tmp") + + def test_no_system_message_added_when_absent(self): + converted = convert_messages_request(messages_request()) + self.assertEqual([m.role for m in converted.messages], ["user"]) + + def test_empty_system_string_adds_no_message(self): + converted = convert_messages_request(messages_request(system="")) + self.assertEqual([m.role for m in converted.messages], ["user"]) + + def test_cache_control_is_ignored(self): + converted = convert_messages_request( + messages_request( + system=[ + { + "type": "text", + "text": "cached", + "cache_control": {"type": "ephemeral"}, + } + ] + ) + ) + self.assertEqual(converted.messages[0].content, "cached") + + +class MidConversationSystemTests(unittest.TestCase): + """Claude Code sends operator instructions as system-role messages.""" + + def convert(self, messages, **kwargs): + return convert_messages_request(messages_request(messages=messages, **kwargs)) + + def test_mid_conversation_system_becomes_a_tagged_user_turn(self): + # Chat templates almost universally reject a system turn that isn't + # first; Qwen's raises outright + converted = self.convert( + [ + {"role": "user", "content": "hi"}, + {"role": "system", "content": "Terse mode enabled."}, + ], + system="Top level prompt.", + ) + + self.assertEqual([m.role for m in converted.messages], ["system", "user", "user"]) + self.assertEqual( + converted.messages[2].content, + "\nTerse mode enabled.\n", + ) + + def test_leading_system_message_becomes_the_system_prompt(self): + converted = self.convert( + [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "hi"}, + ] + ) + + self.assertEqual([m.role for m in converted.messages], ["system", "user"]) + self.assertEqual(converted.messages[0].content, "You are helpful.") + + def test_leading_system_message_yields_to_the_top_level_prompt(self): + # Only one turn can be first, and the top-level prompt already took it + converted = self.convert( + [{"role": "system", "content": "Second one."}, {"role": "user", "content": "hi"}], + system="Top level prompt.", + ) + + self.assertEqual([m.role for m in converted.messages], ["system", "user", "user"]) + self.assertEqual(converted.messages[0].content, "Top level prompt.") + self.assertIn("Second one.", converted.messages[1].content) + + def test_system_message_content_blocks(self): + converted = self.convert( + [ + {"role": "user", "content": "hi"}, + { + "role": "system", + "content": [ + {"type": "text", "text": "one"}, + {"type": "text", "text": "two"}, + ], + }, + ] + ) + + self.assertEqual( + converted.messages[-1].content, + "\none\n\ntwo\n", + ) + + def test_several_system_messages(self): + converted = self.convert( + [ + {"role": "user", "content": "hi"}, + {"role": "system", "content": "first"}, + {"role": "assistant", "content": "ok"}, + {"role": "system", "content": "second"}, + ] + ) + + self.assertEqual( + [m.role for m in converted.messages], ["user", "user", "assistant", "user"] + ) + self.assertIn("first", converted.messages[1].content) + self.assertIn("second", converted.messages[3].content) + + +class MessageContentTests(unittest.TestCase): + def test_string_content(self): + converted = convert_messages_request( + messages_request(messages=[{"role": "user", "content": "hi"}]) + ) + self.assertEqual(converted.messages[0].content, "hi") + + def test_multiple_text_blocks_joined(self): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "part one"}, + {"type": "text", "text": "part two"}, + ], + } + ] + ) + ) + self.assertEqual(converted.messages[0].content, "part one\n\npart two") + + def test_thinking_block_becomes_reasoning_content(self): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "hmm", "signature": "sig"}, + {"type": "text", "text": "Hello!"}, + ], + } + ] + ) + ) + message = converted.messages[0] + self.assertEqual(message.reasoning_content, "hmm") + self.assertEqual(message.content, "Hello!") + + def test_redacted_thinking_dropped(self): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "assistant", + "content": [ + {"type": "redacted_thinking", "data": "encrypted"}, + {"type": "text", "text": "Hello!"}, + ], + } + ] + ) + ) + message = converted.messages[0] + self.assertIsNone(message.reasoning_content) + self.assertEqual(message.content, "Hello!") + + def test_unsupported_block_raises_invalid_request(self): + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [{"type": "document", "source": {"type": "base64"}}], + } + ] + ) + ) + self.assertEqual(ctx.exception.status_code, 400) + self.assertEqual(ctx.exception.error_type, "invalid_request_error") + self.assertIn("document", ctx.exception.detail) + + def test_malformed_known_block_is_reported_as_malformed(self): + # A supported type that failed validation lands in the same fallback + # as an unknown type, but saying it "is not supported" would be wrong + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request( + messages_request( + messages=[{"role": "user", "content": [{"type": "image", "source": {}}]}] + ) + ) + + self.assertIn("image", ctx.exception.detail) + self.assertNotIn("not supported", ctx.exception.detail) + + def test_unsupported_block_inside_tool_result(self): + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "x", + "content": [{"type": "document", "source": {}}], + } + ], + } + ] + ) + ) + self.assertIn("tool_result", ctx.exception.detail) + + +PNG = "iVBORw0KGgoAAAANSUhEUg==" + + +def vision_container(use_vision=True): + return patch.object(common.model, "container", SimpleNamespace(use_vision=use_vision)) + + +class ImageTests(unittest.TestCase): + def base64_message(self, **source): + block = { + "type": "image", + "source": {"type": "base64", "media_type": "image/png", "data": PNG, **source}, + } + return messages_request( + messages=[{"role": "user", "content": [block, {"type": "text", "text": "what is it?"}]}] + ) + + def test_base64_image_becomes_a_data_url_part(self): + with vision_container(): + converted = convert_messages_request(self.base64_message()) + + parts = converted.messages[0].content + self.assertEqual([p.type for p in parts], ["image_url", "text"]) + self.assertEqual(parts[0].image_url.url, f"data:image/png;base64,{PNG}") + self.assertEqual(parts[1].text, "what is it?") + + def test_url_image_is_passed_through(self): + with vision_container(): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + { + "type": "image", + "source": {"type": "url", "url": "https://x.test/a.png"}, + } + ], + } + ] + ) + ) + + self.assertEqual(converted.messages[0].content[0].image_url.url, "https://x.test/a.png") + + def test_text_only_message_stays_a_plain_string(self): + # The common case must not become a part list just because images exist + with vision_container(): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "a"}, + {"type": "text", "text": "b"}, + ], + } + ] + ) + ) + + self.assertEqual(converted.messages[0].content, "a\n\nb") + + def test_consecutive_text_around_an_image_keeps_its_separator(self): + with vision_container(): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "a"}, + {"type": "text", "text": "b"}, + { + "type": "image", + "source": {"type": "url", "url": "https://x.test/a.png"}, + }, + {"type": "text", "text": "c"}, + ], + } + ] + ) + ) + + parts = converted.messages[0].content + self.assertEqual([p.type for p in parts], ["text", "image_url", "text"]) + self.assertEqual(parts[0].text, "a\n\nb") + self.assertEqual(parts[2].text, "c") + + def test_image_rejected_without_a_vision_model(self): + # Templating only builds embeddings for a vision model, so the image + # would otherwise be dropped and the model asked about nothing + with vision_container(use_vision=False): + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request(self.base64_message()) + + self.assertEqual(ctx.exception.status_code, 400) + self.assertIn("does not support images", ctx.exception.detail) + + def test_base64_source_without_data_is_rejected(self): + with vision_container(): + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + { + "type": "image", + "source": {"type": "base64", "media_type": "image/png"}, + } + ], + } + ] + ) + ) + + self.assertIn("media_type and data", ctx.exception.detail) + + def test_file_source_is_rejected(self): + with vision_container(): + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + { + "type": "image", + "source": {"type": "file", "file_id": "file_1"}, + } + ], + } + ] + ) + ) + + self.assertIn("file", ctx.exception.detail) + + def test_image_inside_tool_result(self): + with vision_container(): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_1", + "content": [ + {"type": "text", "text": "screenshot:"}, + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": PNG, + }, + }, + ], + } + ], + } + ] + ) + ) + + message = converted.messages[0] + self.assertEqual(message.role, "tool") + self.assertEqual([p.type for p in message.content], ["text", "image_url"]) + + def test_error_tool_result_with_an_image_keeps_the_error_marker(self): + with vision_container(): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_1", + "is_error": True, + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": PNG, + }, + } + ], + } + ], + } + ] + ) + ) + + parts = converted.messages[0].content + self.assertEqual(parts[0].text, "Error") + self.assertEqual(parts[1].type, "image_url") + + +class ToolConversionTests(unittest.TestCase): + def test_tool_use_block_becomes_tool_call(self): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": "toolu_1", + "name": "get_weather", + "input": {"city": "Paris"}, + } + ], + } + ] + ) + ) + + message = converted.messages[0] + self.assertEqual(message.role, "assistant") + + # An assistant turn that only called tools carries no text + self.assertIsNone(message.content) + self.assertEqual(len(message.tool_calls), 1) + + call = message.tool_calls[0] + self.assertEqual(call.id, "toolu_1") + self.assertEqual(call.function.name, "get_weather") + + # Templates render the OAI shape, whose arguments are a JSON string + self.assertEqual(json.loads(call.function.arguments), {"city": "Paris"}) + + def test_tool_results_fan_out_to_one_message_each(self): + # Anthropic packs every result into one user message; templates expect + # one tool message per result + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "toolu_1", "content": "21C"}, + {"type": "tool_result", "tool_use_id": "toolu_2", "content": "rainy"}, + ], + } + ] + ) + ) + + self.assertEqual([m.role for m in converted.messages], ["tool", "tool"]) + self.assertEqual([m.tool_call_id for m in converted.messages], ["toolu_1", "toolu_2"]) + self.assertEqual([m.content for m in converted.messages], ["21C", "rainy"]) + + def test_tool_result_with_text_blocks(self): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_1", + "content": [ + {"type": "text", "text": "line one"}, + {"type": "text", "text": "line two"}, + ], + } + ], + } + ] + ) + ) + + self.assertEqual(converted.messages[0].content, "line one\n\nline two") + + def test_tool_results_precede_trailing_user_text(self): + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "toolu_1", "content": "21C"}, + {"type": "text", "text": "What should I wear?"}, + ], + } + ] + ) + ) + + self.assertEqual([m.role for m in converted.messages], ["tool", "user"]) + self.assertEqual(converted.messages[1].content, "What should I wear?") + + def test_error_tool_result_is_marked_in_the_text(self): + # Templates have no concept of a failed call, so the model can only + # act on the failure if it can read it + converted = convert_messages_request( + messages_request( + messages=[ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_1", + "content": "no such city", + "is_error": True, + } + ], + } + ] + ) + ) + + self.assertEqual(converted.messages[0].content, "Error: no such city") + + def test_full_tool_round_trip_message_order(self): + converted = convert_messages_request( + messages_request( + system="be helpful", + messages=[ + {"role": "user", "content": "weather in Paris?"}, + { + "role": "assistant", + "content": [ + {"type": "text", "text": "Checking."}, + { + "type": "tool_use", + "id": "toolu_1", + "name": "get_weather", + "input": {"city": "Paris"}, + }, + ], + }, + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "toolu_1", "content": "21C"} + ], + }, + ], + ) + ) + + self.assertEqual( + [m.role for m in converted.messages], ["system", "user", "assistant", "tool"] + ) + self.assertEqual(converted.messages[2].content, "Checking.") + self.assertEqual(len(converted.messages[2].tool_calls), 1) + + def test_tool_definitions_become_oai_specs(self): + converted = convert_messages_request( + messages_request( + tools=[ + { + "name": "get_weather", + "description": "Get the weather", + "input_schema": { + "type": "object", + "properties": {"city": {"type": "string"}}, + }, + } + ] + ) + ) + + spec = converted.tools[0] + self.assertEqual(spec.type, "function") + self.assertEqual(spec.function.name, "get_weather") + self.assertEqual(spec.function.description, "Get the weather") + self.assertEqual(spec.function.parameters["properties"], {"city": {"type": "string"}}) + + def test_tool_cache_control_is_ignored(self): + converted = convert_messages_request( + messages_request( + tools=[ + { + "name": "get_weather", + "input_schema": {"type": "object"}, + "cache_control": {"type": "ephemeral"}, + } + ] + ) + ) + self.assertEqual(converted.tools[0].function.name, "get_weather") + + def test_server_tool_is_rejected(self): + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request( + messages_request(tools=[{"type": "web_search_20260209", "name": "web_search"}]) + ) + + self.assertEqual(ctx.exception.status_code, 400) + self.assertIn("web_search_20260209", ctx.exception.detail) + + def test_tool_without_schema_is_rejected(self): + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request(messages_request(tools=[{"name": "broken"}])) + + self.assertEqual(ctx.exception.status_code, 400) + + def test_no_tools_leaves_field_unset(self): + self.assertIsNone(convert_messages_request(messages_request()).tools) + + def test_tool_choice_modes(self): + for anthropic_mode, expected in [("auto", "auto"), ("any", "required"), ("none", "none")]: + converted = convert_messages_request( + messages_request(tool_choice={"type": anthropic_mode}) + ) + self.assertEqual(converted.tool_choice, expected) + + def test_named_tool_choice(self): + converted = convert_messages_request( + messages_request(tool_choice={"type": "tool", "name": "get_weather"}) + ) + self.assertEqual(converted.tool_choice.function.name, "get_weather") + + def test_named_tool_choice_without_name_is_rejected(self): + with self.assertRaises(AnthropicHTTPException): + convert_messages_request(messages_request(tool_choice={"type": "tool"})) + + def test_unknown_tool_choice_is_rejected(self): + with self.assertRaises(AnthropicHTTPException) as ctx: + convert_messages_request(messages_request(tool_choice={"type": "sometimes"})) + + self.assertIn("sometimes", ctx.exception.detail) + + def test_disable_parallel_tool_use(self): + converted = convert_messages_request( + messages_request(tool_choice={"type": "auto", "disable_parallel_tool_use": True}) + ) + self.assertIs(converted.parallel_tool_calls, False) + + def test_parallel_tool_use_left_alone_when_unspecified(self): + converted = convert_messages_request(messages_request(tool_choice={"type": "auto"})) + self.assertIs(converted.parallel_tool_calls, True) + + +class SamplerMappingTests(unittest.TestCase): + def test_unset_samplers_are_omitted(self): + params = _sampler_params(messages_request()) + self.assertEqual(set(params), {"max_tokens"}) + + def test_set_samplers_are_mapped(self): + params = _sampler_params( + messages_request( + temperature=0.7, + top_p=0.9, + top_k=40, + stop_sequences=["STOP"], + metadata={"user_id": "u1"}, + ) + ) + self.assertEqual(params["temperature"], 0.7) + self.assertEqual(params["top_p"], 0.9) + self.assertEqual(params["top_k"], 40) + self.assertEqual(params["stop"], ["STOP"]) + self.assertEqual(params["user"], "u1") + + def test_request_shape(self): + converted = convert_messages_request(messages_request(max_tokens=64)) + self.assertEqual(converted.max_tokens, 64) + self.assertEqual(converted.n, 1) + self.assertTrue(converted.stream_options.include_usage) + + def test_thinking_enabled(self): + converted = convert_messages_request(messages_request(thinking={"type": "enabled"})) + self.assertIs(converted.template_vars["enable_thinking"], True) + + def test_thinking_disabled(self): + converted = convert_messages_request(messages_request(thinking={"type": "disabled"})) + self.assertIs(converted.template_vars["enable_thinking"], False) + + def test_thinking_absent_sets_no_template_var(self): + converted = convert_messages_request(messages_request()) + self.assertNotIn("enable_thinking", converted.template_vars) + + def test_template_vars_passthrough(self): + converted = convert_messages_request( + messages_request(template_vars={"reasoning_effort": "high"}) + ) + self.assertEqual(converted.template_vars["reasoning_effort"], "high") + + def test_template_vars_accepts_the_oai_alias(self): + converted = convert_messages_request( + messages_request(chat_template_kwargs={"verbosity": "low"}) + ) + self.assertEqual(converted.template_vars["verbosity"], "low") + + def test_explicit_template_vars_beat_the_thinking_field(self): + # Mirrors the chat completion path, where template_vars outrank the + # flat reasoning fields + converted = convert_messages_request( + messages_request(thinking={"type": "disabled"}, template_vars={"enable_thinking": True}) + ) + self.assertIs(converted.template_vars["enable_thinking"], True) + + def test_template_vars_still_lose_to_force(self): + converted = convert_messages_request( + messages_request(template_vars={"reasoning_effort": "high", "preserve_thinking": False}) + ) + container = SimpleNamespace( + template_vars_default={"verbosity": "medium"}, + template_vars_force={"preserve_thinking": True}, + ) + resolved = resolve_template_vars(converted, container) + + self.assertEqual(resolved["verbosity"], "medium") + self.assertEqual(resolved["reasoning_effort"], "high") + self.assertIs(resolved["preserve_thinking"], True) + + def test_count_tokens_takes_template_vars(self): + # Counting has to render the prompt generation would + converted = convert_count_tokens_request( + CountTokensRequest( + messages=[{"role": "user", "content": "hi"}], + template_vars={"enable_thinking": False}, + ) + ) + self.assertIs(converted.template_vars["enable_thinking"], False) + + def test_count_tokens_request_conversion(self): + converted = convert_count_tokens_request( + CountTokensRequest(system="sys", messages=[{"role": "user", "content": "hi"}]) + ) + self.assertEqual([m.role for m in converted.messages], ["system", "user"]) + + +class StopReasonTests(unittest.TestCase): + def test_length_maps_to_max_tokens(self): + reason, sequence = stop_reason("length", "max_new_tokens", None, None) + self.assertEqual(reason, "max_tokens") + self.assertIsNone(sequence) + + def test_tool_calls_maps_to_tool_use(self): + reason, _ = stop_reason("tool_calls", None, None, None) + self.assertEqual(reason, "tool_use") + + def test_eos_token_maps_to_end_turn(self): + reason, sequence = stop_reason("stop", "stop_token", "<|im_end|>", ["STOP"]) + self.assertEqual(reason, "end_turn") + self.assertIsNone(sequence) + + def test_client_stop_sequence_is_reported(self): + reason, sequence = stop_reason("stop", "stop_string", "STOP", ["STOP"]) + self.assertEqual(reason, "stop_sequence") + self.assertEqual(sequence, "STOP") + + def test_template_stop_string_is_not_reported_as_stop_sequence(self): + # The prompt template contributes stop strings the client never sent; + # naming one in stop_sequence would be a lie + reason, sequence = stop_reason("stop", "stop_string", "<|end|>", ["STOP"]) + self.assertEqual(reason, "end_turn") + self.assertIsNone(sequence) + + def test_streaming_and_non_streaming_agree(self): + # Both paths must derive the stop reason from the same inputs + c = choice(finish_reason="length") + response = convert_response( + ChatCompletionResponse(choices=[c], model="test-model"), + messages_request(), + "test-model", + ) + streamed, _ = stop_reason(c.finish_reason, c.eos_reason, c.stop_str, None) + self.assertEqual(response.stop_reason, streamed) + + +class ConvertResponseTests(unittest.TestCase): + def test_thinking_precedes_text(self): + response = convert_response( + completion(content="Hello", reasoning_content="hmm"), + messages_request(), + "test-model", + ) + self.assertIsInstance(response.content[0], ResponseThinkingBlock) + self.assertEqual(response.content[0].thinking, "hmm") + self.assertEqual(response.content[0].signature, "") + self.assertIsInstance(response.content[1], ResponseTextBlock) + self.assertEqual(response.content[1].text, "Hello") + + def test_text_only(self): + response = convert_response(completion(), messages_request(), "test-model") + self.assertEqual(len(response.content), 1) + self.assertEqual(response.content[0].type, "text") + + def test_empty_content(self): + response = convert_response(completion(content=None), messages_request(), "test-model") + self.assertEqual(response.content, []) + + def test_envelope_fields(self): + response = convert_response(completion(), messages_request(), "test-model") + self.assertTrue(response.id.startswith("msg_")) + self.assertEqual(response.type, "message") + self.assertEqual(response.role, "assistant") + self.assertEqual(response.model, "test-model") + + def test_usage_mapping(self): + response = convert_response( + completion(usage=UsageStats(prompt_tokens=12, completion_tokens=5, total_tokens=17)), + messages_request(), + "test-model", + ) + self.assertEqual(response.usage.input_tokens, 12) + self.assertEqual(response.usage.output_tokens, 5) + self.assertEqual(response.usage.cache_read_input_tokens, 0) + + def test_missing_usage_defaults_to_zero(self): + response = convert_response(completion(usage=None), messages_request(), "test-model") + self.assertEqual(response.usage.input_tokens, 0) + + +class FakeRequest: + """Minimal stand-in for the parts of Request the endpoints touch.""" + + def __init__(self, body=None): + self.body = body or {} + self.state = SimpleNamespace(id="test-request") + + async def json(self): + return self.body + + async def is_disconnected(self): + return False + + +class EndpointTests(unittest.TestCase): + """Checks over the endpoint functions with inference stubbed out.""" + + def setUp(self): + container = SimpleNamespace( + prompt_template=SimpleNamespace(name="test"), + model_dir=MODEL_DIR, + encode_tokens=lambda text, **kwargs: list(range(7)), + validate_context_length=lambda *args, **kwargs: 12, + ) + + async def fake_apply_chat_template(data): + return "PROMPT", None + + async def fake_generate(*args, **kwargs): + return completion( + usage=UsageStats(prompt_tokens=12, completion_tokens=5, total_tokens=17) + ) + + async def fake_check_model_container(): + return None + + async def fake_load_inline_model(model_name, request): + return None + + patches = [ + patch.object(common.model, "container", container), + patch("endpoints.Anthropic.router.check_model_container", fake_check_model_container), + patch("endpoints.Anthropic.router.load_inline_model", fake_load_inline_model), + patch("endpoints.Anthropic.router.apply_chat_template", fake_apply_chat_template), + patch("endpoints.Anthropic.router.generate_chat_completion", fake_generate), + patch( + "endpoints.Anthropic.utils.messages.apply_chat_template", + fake_apply_chat_template, + ), + ] + for entry in patches: + entry.start() + self.addCleanup(entry.stop) + + def test_messages_success(self): + data = messages_request(model="test-model") + response = asyncio.run(messages_endpoint(FakeRequest(), data)) + + body = response.model_dump() + self.assertEqual(body["type"], "message") + self.assertEqual(body["role"], "assistant") + self.assertEqual(body["content"], [{"type": "text", "text": "Hello"}]) + self.assertEqual(body["stop_reason"], "end_turn") + self.assertEqual(body["model"], MODEL_DIR.name) + self.assertEqual(body["usage"]["input_tokens"], 12) + self.assertEqual(body["usage"]["output_tokens"], 5) + + def test_count_tokens(self): + data = CountTokensRequest(messages=[{"role": "user", "content": "hi"}]) + response = asyncio.run(count_tokens_request(FakeRequest(), data)) + + self.assertEqual(response.input_tokens, 7) + + def test_streaming_returns_an_event_stream(self): + response = asyncio.run(messages_endpoint(FakeRequest(), messages_request(stream=True))) + + self.assertIsInstance(response, EventSourceResponse) + + def test_streaming_rejected_when_disabled_in_config(self): + # Returning a non-streaming body to a client expecting SSE would fail + # in the client's parser rather than say what went wrong + with patch.object(config.developer, "disable_request_streaming", True): + with self.assertRaises(AnthropicHTTPException) as ctx: + asyncio.run(messages_endpoint(FakeRequest(), messages_request(stream=True))) + + self.assertEqual(ctx.exception.status_code, 400) + self.assertEqual(ctx.exception.error_type, "invalid_request_error") + + def test_missing_prompt_template_rejected(self): + with patch.object( + common.model, + "container", + SimpleNamespace(prompt_template=None, model_dir=MODEL_DIR), + ): + with self.assertRaises(AnthropicHTTPException) as ctx: + asyncio.run(messages_endpoint(FakeRequest(), messages_request())) + + self.assertEqual(ctx.exception.status_code, 422) + + def test_load_lock_released_on_error(self): + # A failure past the lock must not wedge every later request + from endpoints.Anthropic.router import load_lock + + with patch.object( + common.model, + "container", + SimpleNamespace(prompt_template=None, model_dir=MODEL_DIR), + ): + with self.assertRaises(AnthropicHTTPException): + asyncio.run(messages_endpoint(FakeRequest(), messages_request())) + + self.assertFalse(load_lock.locked()) + + +class ToolResponseTests(unittest.TestCase): + def tool_call(self, name="get_weather", arguments='{"city": "Paris"}', call_id="toolu_1"): + return ToolCall(id=call_id, function=Tool(name=name, arguments=arguments)) + + def test_tool_use_block_shape(self): + c = choice(content=None, finish_reason="tool_calls") + c.message.tool_calls = [self.tool_call()] + + response = convert_response( + ChatCompletionResponse(choices=[c], model="test-model"), + messages_request(), + "test-model", + ) + + block = response.content[0] + self.assertIsInstance(block, ResponseToolUseBlock) + self.assertEqual(block.type, "tool_use") + self.assertEqual(block.id, "toolu_1") + self.assertEqual(block.name, "get_weather") + + # The wire form is an object, not the JSON string the pipeline uses + self.assertEqual(block.input, {"city": "Paris"}) + + def test_text_precedes_tool_use(self): + c = choice(content="Checking.", reasoning_content="hmm", finish_reason="tool_calls") + c.message.tool_calls = [self.tool_call()] + + response = convert_response( + ChatCompletionResponse(choices=[c], model="test-model"), + messages_request(), + "test-model", + ) + + self.assertEqual([b.type for b in response.content], ["thinking", "text", "tool_use"]) + + def test_parallel_tool_calls_become_separate_blocks(self): + c = choice(content=None, finish_reason="tool_calls") + c.message.tool_calls = [ + self.tool_call(call_id="toolu_1"), + self.tool_call(name="get_time", arguments="{}", call_id="toolu_2"), + ] + + response = convert_response( + ChatCompletionResponse(choices=[c], model="test-model"), + messages_request(), + "test-model", + ) + + self.assertEqual([b.id for b in response.content], ["toolu_1", "toolu_2"]) + self.assertEqual(response.content[1].input, {}) + + def test_unparseable_arguments_yield_empty_input(self): + # Surfacing the call with an empty input beats failing the response; + # the tool name is the useful part + self.assertEqual(tool_call_input("get_weather", "not json"), {}) + self.assertEqual(tool_call_input("get_weather", "[1, 2]"), {}) + self.assertEqual(tool_call_input("get_weather", '{"a": 1}'), {"a": 1}) + + +class UsageAccountingTests(unittest.TestCase): + """Prompt tokens served from the prefix cache are counted separately.""" + + def stats(self, prompt_tokens=100, cached_tokens=None, completion_tokens=5): + return UsageStats( + prompt_tokens=prompt_tokens, + cached_tokens=cached_tokens, + completion_tokens=completion_tokens, + total_tokens=prompt_tokens + completion_tokens, + ) + + def test_cached_prefix_is_not_counted_as_fresh_input(self): + # TabbyAPI's prompt_tokens is the whole prompt; Anthropic's + # input_tokens is only the part that was not read from cache + usage = usage_from_stats(self.stats(prompt_tokens=100, cached_tokens=90)) + + self.assertEqual(usage.input_tokens, 10) + self.assertEqual(usage.cache_read_input_tokens, 90) + self.assertEqual(usage.output_tokens, 5) + + def test_uncached_prompt_is_all_input(self): + usage = usage_from_stats(self.stats(prompt_tokens=100, cached_tokens=0)) + + self.assertEqual(usage.input_tokens, 100) + self.assertEqual(usage.cache_read_input_tokens, 0) + + def test_missing_cached_count_is_treated_as_uncached(self): + usage = usage_from_stats(self.stats(prompt_tokens=100, cached_tokens=None)) + + self.assertEqual(usage.input_tokens, 100) + self.assertEqual(usage.cache_read_input_tokens, 0) + + def test_cached_count_cannot_drive_input_negative(self): + usage = usage_from_stats(self.stats(prompt_tokens=100, cached_tokens=150)) + + self.assertEqual(usage.input_tokens, 0) + self.assertEqual(usage.cache_read_input_tokens, 100) + + def test_cache_creation_is_never_guessed(self): + # The backend cannot distinguish a cache write from ordinary prefill, + # and clients price writes above plain input + usage = usage_from_stats(self.stats(prompt_tokens=100, cached_tokens=40)) + + self.assertEqual(usage.cache_creation_input_tokens, 0) + + def test_absent_usage_is_all_zero(self): + usage = usage_from_stats(None) + + self.assertEqual(usage.input_tokens, 0) + self.assertEqual(usage.output_tokens, 0) + + def test_non_streaming_response_reports_the_split(self): + response = convert_response( + completion(usage=self.stats(prompt_tokens=100, cached_tokens=90)), + messages_request(), + "test-model", + ) + + self.assertEqual(response.usage.input_tokens, 10) + self.assertEqual(response.usage.cache_read_input_tokens, 90) + + def test_pipeline_carries_cached_tokens_from_the_finish_chunk(self): + stats = get_usage_stats( + {"finish_reason": "stop", "prompt_tokens": 100, "cached_tokens": 90.0, "gen_tokens": 5} + ) + self.assertEqual(stats.cached_tokens, 90) + + def test_aggregate_keeps_the_shared_cached_count(self): + # One prompt is shared across choices, so its cached portion is too + stats = [ + UsageStats(prompt_tokens=100, cached_tokens=90, completion_tokens=5, total_tokens=105), + UsageStats(prompt_tokens=100, cached_tokens=90, completion_tokens=7, total_tokens=107), + ] + for entry in stats: + entry.prompt_time = 0.1 + entry.completion_time = 0.2 + + aggregated = aggregate_usage_stats(stats) + + self.assertEqual(aggregated.cached_tokens, 90) + self.assertEqual(aggregated.prompt_tokens, 100) + self.assertEqual(aggregated.completion_tokens, 12) + + +class ContentBlockTrackerTests(unittest.TestCase): + def names(self, events): + return [event.event for event in events] + + def test_blocks_open_lazily(self): + # A response without reasoning must put its text at index 0 + tracker = ContentBlockTracker() + events = tracker.write("text", "Hello") + + self.assertEqual(self.names(events), ["content_block_start", "content_block_delta"]) + self.assertEqual(json.loads(events[0].data)["index"], 0) + self.assertEqual(json.loads(events[0].data)["content_block"]["type"], "text") + + def test_empty_text_emits_nothing(self): + tracker = ContentBlockTracker() + self.assertEqual(tracker.write("text", ""), []) + self.assertEqual(tracker.write("thinking", ""), []) + + def test_switching_kind_closes_previous_block(self): + tracker = ContentBlockTracker() + events = tracker.write("thinking", "hm") + tracker.write("text", "Hi") + + self.assertEqual( + self.names(events), + [ + "content_block_start", + "content_block_delta", + "content_block_stop", + "content_block_start", + "content_block_delta", + ], + ) + self.assertEqual(json.loads(events[2].data)["index"], 0) + self.assertEqual(json.loads(events[3].data)["index"], 1) + + def test_same_kind_reuses_open_block(self): + tracker = ContentBlockTracker() + events = tracker.write("text", "a") + tracker.write("text", "b") + + self.assertEqual( + self.names(events), + ["content_block_start", "content_block_delta", "content_block_delta"], + ) + self.assertTrue(all(json.loads(e.data)["index"] == 0 for e in events)) + + def test_close_without_open_block_emits_nothing(self): + self.assertEqual(ContentBlockTracker().close(), []) + + def test_close_is_not_repeated(self): + tracker = ContentBlockTracker() + tracker.write("text", "a") + self.assertEqual(len(tracker.close()), 1) + self.assertEqual(tracker.close(), []) + + def test_thinking_block_carries_empty_signature(self): + events = ContentBlockTracker().write("thinking", "hm") + self.assertEqual(json.loads(events[0].data)["content_block"]["signature"], "") + + +class FakeDisconnectHandler: + def __init__(self): + self.cleaned = False + + async def cleanup(self): + self.cleaned = True + + +def run_stream(data, chunks, input_tokens=12): + """Drive the stream generator with a stubbed collector.""" + + async def fake_collector(task_idx, gen_queue, *args, **kwargs): + for chunk in chunks: + await gen_queue.put(chunk) + + handler = FakeDisconnectHandler() + + async def drive(): + events = [] + with ( + patch("endpoints.Anthropic.utils.stream._chat_stream_collector", fake_collector), + patch( + "endpoints.Anthropic.utils.stream._resolve_start_in_reasoning", + lambda prompt, params: False, + ), + ): + generator = stream_generate_message( + "PROMPT", + None, + data, + convert_messages_request(data), + FakeRequest(), + MODEL_DIR, + handler, + input_tokens, + ) + async for event in generator: + events.append((event.event, json.loads(event.data))) + + return events + + return asyncio.run(drive()), handler + + +def finish_chunk(**kwargs): + chunk = { + "index": 0, + "finish_reason": "stop", + "eos_reason": "stop_token", + "stop_str": None, + "prompt_tokens": 12, + "cached_tokens": 8, + "gen_tokens": 5, + "delta_content": "", + "delta_reasoning_content": "", + } + chunk.update(kwargs) + return chunk + + +class StreamEventTests(unittest.TestCase): + def test_full_event_sequence(self): + events, handler = run_stream( + messages_request(), + [ + {"index": 0, "delta_content": "Hel", "delta_reasoning_content": ""}, + {"index": 0, "delta_content": "lo", "delta_reasoning_content": ""}, + finish_chunk(), + ], + ) + + self.assertEqual( + [name for name, _ in events], + [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ], + ) + self.assertTrue(handler.cleaned) + + def test_message_start_carries_input_tokens(self): + events, _ = run_stream(messages_request(), [finish_chunk()], input_tokens=99) + + name, payload = events[0] + self.assertEqual(name, "message_start") + self.assertEqual(payload["message"]["usage"]["input_tokens"], 99) + self.assertEqual(payload["message"]["usage"]["output_tokens"], 0) + self.assertEqual(payload["message"]["role"], "assistant") + self.assertEqual(payload["message"]["content"], []) + self.assertTrue(payload["message"]["id"].startswith("msg_")) + self.assertEqual(payload["message"]["model"], MODEL_DIR.name) + + def test_no_done_sentinel(self): + # Anthropic terminates on message_stop; a [DONE] would be an OAI-ism + events, _ = run_stream(messages_request(), [finish_chunk()]) + self.assertEqual(events[-1][0], "message_stop") + + def test_reasoning_then_content_uses_two_blocks(self): + events, _ = run_stream( + messages_request(), + [ + {"index": 0, "delta_reasoning_content": "hm", "delta_content": ""}, + {"index": 0, "delta_reasoning_content": "", "delta_content": "Hi"}, + finish_chunk(), + ], + ) + + starts = [p for n, p in events if n == "content_block_start"] + self.assertEqual([s["content_block"]["type"] for s in starts], ["thinking", "text"]) + self.assertEqual([s["index"] for s in starts], [0, 1]) + + deltas = [p["delta"] for n, p in events if n == "content_block_delta"] + self.assertEqual(deltas[0], {"type": "thinking_delta", "thinking": "hm"}) + self.assertEqual(deltas[1], {"type": "text_delta", "text": "Hi"}) + + def test_message_delta_reports_stop_reason_and_usage(self): + events, _ = run_stream( + messages_request(), + [{"index": 0, "delta_content": "Hi", "delta_reasoning_content": ""}, finish_chunk()], + ) + + payload = dict(events)["message_delta"] + self.assertEqual(payload["delta"]["stop_reason"], "end_turn") + self.assertIsNone(payload["delta"]["stop_sequence"]) + self.assertEqual(payload["usage"]["output_tokens"], 5) + + # 12 prompt tokens of which 8 came from the prefix cache + self.assertEqual(payload["usage"]["input_tokens"], 4) + self.assertEqual(payload["usage"]["cache_read_input_tokens"], 8) + + def test_max_tokens_stop_reason(self): + events, _ = run_stream( + messages_request(), + [finish_chunk(finish_reason="length", eos_reason="max_new_tokens")], + ) + self.assertEqual(dict(events)["message_delta"]["delta"]["stop_reason"], "max_tokens") + + def test_client_stop_sequence_reported_in_message_delta(self): + events, _ = run_stream( + messages_request(stop_sequences=["END"]), + [finish_chunk(eos_reason="stop_string", stop_str="END")], + ) + + delta = dict(events)["message_delta"]["delta"] + self.assertEqual(delta["stop_reason"], "stop_sequence") + self.assertEqual(delta["stop_sequence"], "END") + + def test_collector_exception_becomes_error_event(self): + events, handler = run_stream(messages_request(), [RuntimeError("backend exploded")]) + + names = [name for name, _ in events] + self.assertEqual(names[0], "message_start") + self.assertEqual(names[-1], "error") + self.assertEqual(dict(events)["error"]["type"], "error") + self.assertEqual(dict(events)["error"]["error"]["type"], "api_error") + self.assertTrue(handler.cleaned) + + def test_collector_finishing_without_finish_chunk_closes_stream(self): + # Must not wait forever on a chunk that will never arrive + events, _ = run_stream( + messages_request(), + [{"index": 0, "delta_content": "Hi", "delta_reasoning_content": ""}], + ) + + names = [name for name, _ in events] + self.assertEqual(names[-1], "message_stop") + self.assertIn("content_block_stop", names) + + def test_tool_call_streams_as_its_own_block(self): + events, _ = run_stream( + messages_request(), + [ + finish_chunk( + finish_reason="tool_calls", + delta_tool_calls=[ + { + "id": "toolu_1", + "type": "function", + "index": 0, + "function": {"name": "get_weather", "arguments": '{"city": "Paris"}'}, + } + ], + ) + ], + ) + + self.assertEqual( + [name for name, _ in events], + [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ], + ) + + start = [p for n, p in events if n == "content_block_start"][0] + self.assertEqual(start["content_block"]["type"], "tool_use") + self.assertEqual(start["content_block"]["id"], "toolu_1") + self.assertEqual(start["content_block"]["name"], "get_weather") + self.assertEqual(start["content_block"]["input"], {}) + + delta = [p for n, p in events if n == "content_block_delta"][0]["delta"] + self.assertEqual(delta["type"], "input_json_delta") + self.assertEqual(json.loads(delta["partial_json"]), {"city": "Paris"}) + + self.assertEqual(dict(events)["message_delta"]["delta"]["stop_reason"], "tool_use") + + def test_text_block_is_closed_before_a_tool_block_opens(self): + events, _ = run_stream( + messages_request(), + [ + {"index": 0, "delta_content": "Checking.", "delta_reasoning_content": ""}, + finish_chunk( + finish_reason="tool_calls", + delta_tool_calls=[ + { + "id": "toolu_1", + "function": {"name": "get_weather", "arguments": "{}"}, + } + ], + ), + ], + ) + + names = [name for name, _ in events] + self.assertEqual( + names, + [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ], + ) + + starts = [p for n, p in events if n == "content_block_start"] + self.assertEqual([s["content_block"]["type"] for s in starts], ["text", "tool_use"]) + self.assertEqual([s["index"] for s in starts], [0, 1]) + + def test_parallel_tool_calls_stream_as_separate_blocks(self): + events, _ = run_stream( + messages_request(), + [ + finish_chunk( + finish_reason="tool_calls", + delta_tool_calls=[ + {"id": "toolu_1", "function": {"name": "a", "arguments": "{}"}}, + {"id": "toolu_2", "function": {"name": "b", "arguments": "{}"}}, + ], + ) + ], + ) + + starts = [p for n, p in events if n == "content_block_start"] + self.assertEqual([s["content_block"]["id"] for s in starts], ["toolu_1", "toolu_2"]) + self.assertEqual([s["index"] for s in starts], [0, 1]) + + # Each block must be closed before the next opens + names = [name for name, _ in events] + self.assertEqual(names.count("content_block_stop"), 2) + + def test_open_block_is_closed_before_message_delta(self): + events, _ = run_stream( + messages_request(), + [{"index": 0, "delta_content": "Hi", "delta_reasoning_content": ""}, finish_chunk()], + ) + + names = [name for name, _ in events] + self.assertLess(names.index("content_block_stop"), names.index("message_delta")) + + +class ErrorEnvelopeTests(unittest.TestCase): + def test_anthropic_exception_keeps_its_type(self): + status, content = exception_to_response( + AnthropicHTTPException(429, "slow down", "rate_limit_error") + ) + self.assertEqual(status, 429) + self.assertEqual(content["type"], "error") + self.assertEqual(content["error"]["type"], "rate_limit_error") + self.assertEqual(content["error"]["message"], "slow down") + + def test_shared_http_exception_is_reshaped(self): + # Raised by check_api_key, which knows nothing about this API + status, content = exception_to_response(HTTPException(401, "Invalid API key")) + self.assertEqual(status, 401) + self.assertEqual(content["error"]["type"], "authentication_error") + self.assertEqual(content["error"]["message"], "Invalid API key") + + def test_template_failure_maps_to_invalid_request(self): + status, content = exception_to_response(HTTPException(422, "TemplateError: boom")) + self.assertEqual(status, 422) + self.assertEqual(content["error"]["type"], "invalid_request_error") + + def test_unloaded_model_maps_to_api_error(self): + _, content = exception_to_response(HTTPException(503, "no model")) + self.assertEqual(content["error"]["type"], "api_error") + + def test_validation_error_is_reshaped(self): + status, content = exception_to_response(RequestValidationError([])) + self.assertEqual(status, 422) + self.assertEqual(content["error"]["type"], "invalid_request_error") + + def test_unknown_exception_is_reraised(self): + with self.assertRaises(ValueError): + exception_to_response(ValueError("not an HTTP error")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_context_length_errors.py b/tests/test_context_length_errors.py index 76ab6d8f..38e07c8a 100644 --- a/tests/test_context_length_errors.py +++ b/tests/test_context_length_errors.py @@ -125,19 +125,26 @@ def test_streaming_preflight_returns_400_for_context_length_error(self): def test_streaming_preflight_checks_each_batched_prompt(self): checked_prompts = [] - container = SimpleNamespace( - validate_context_length=lambda prompt, *args: checked_prompts.append(prompt) - ) + + def validate(prompt, *args): + checked_prompts.append(prompt) + return len(prompt) + + container = SimpleNamespace(validate_context_length=validate) original_container = model.container model.container = container try: - model.check_context_length(["first", "second"], DummyRequestData()) + longest = model.check_context_length(["first", "second"], DummyRequestData()) finally: model.container = original_container self.assertEqual(checked_prompts, ["first", "second"]) + # The longest prompt's length is returned so a caller needing the + # prompt token count up front doesn't tokenize a second time + self.assertEqual(longest, len("second")) + async def test_completion_returns_400_for_context_length_error(self): error = ContextLengthExceededError("Prompt length 9 is greater than max_seq_len 8") diff --git a/tests/test_model_card.py b/tests/test_model_card.py new file mode 100644 index 00000000..45f59e3e --- /dev/null +++ b/tests/test_model_card.py @@ -0,0 +1,151 @@ +import pathlib +import tempfile +import unittest +from unittest.mock import patch + +import common.model # noqa: F401 - resolve import cycle ordering +from common import model +from endpoints.core.types.model import ModelCard, ModelCardParameters, ModelList +from endpoints.core.utils.model import get_current_model_list, get_model_list + + +class ModelCardTests(unittest.TestCase): + def test_openai_fields(self): + card = ModelCard(id="my-model") + self.assertEqual(card.object, "model") + self.assertEqual(card.owned_by, "tabbyAPI") + self.assertIsInstance(card.created, int) + + def test_anthropic_fields_are_derived(self): + card = ModelCard(id="my-model") + self.assertEqual(card.type, "model") + self.assertEqual(card.display_name, "my-model") + + def test_created_at_is_iso_utc(self): + card = ModelCard(id="my-model", created=0) + self.assertEqual(card.created_at, "1970-01-01T00:00:00Z") + + def test_explicit_values_are_kept(self): + card = ModelCard( + id="my-model", display_name="Pretty Name", created_at="2020-01-01T00:00:00Z" + ) + self.assertEqual(card.display_name, "Pretty Name") + self.assertEqual(card.created_at, "2020-01-01T00:00:00Z") + + +class ModelCardTokenFieldTests(unittest.TestCase): + def test_token_fields_follow_max_seq_len(self): + card = ModelCard(id="my-model", parameters=ModelCardParameters(max_seq_len=262144)) + + self.assertEqual(card.max_input_tokens, 262144) + self.assertEqual(card.max_tokens, 262144) + + def test_token_fields_are_null_without_parameters(self): + card = ModelCard(id="my-model") + + self.assertIsNone(card.max_input_tokens) + self.assertIsNone(card.max_tokens) + + def test_token_fields_are_serialized(self): + card = ModelCard(id="my-model", parameters=ModelCardParameters(max_seq_len=4096)) + dumped = card.model_dump() + + self.assertEqual(dumped["max_input_tokens"], 4096) + self.assertEqual(dumped["max_tokens"], 4096) + + +class ModelListTests(unittest.TestCase): + def test_empty_list(self): + listing = ModelList() + self.assertFalse(listing.has_more) + self.assertIsNone(listing.first_id) + self.assertIsNone(listing.last_id) + + def test_ids_track_appends(self): + # Callers build the list empty and append afterwards, so the ids + # cannot be captured at construction time + listing = ModelList() + listing.data.append(ModelCard(id="a")) + listing.data.append(ModelCard(id="b")) + + self.assertEqual(listing.first_id, "a") + self.assertEqual(listing.last_id, "b") + + def test_ids_are_serialized(self): + listing = ModelList(data=[ModelCard(id="a")]) + dumped = listing.model_dump() + + self.assertEqual(dumped["first_id"], "a") + self.assertEqual(dumped["last_id"], "a") + self.assertFalse(dumped["has_more"]) + + +class DummyContainer: + """Stands in for a loaded model container.""" + + def __init__(self, model_dir: pathlib.Path): + self.model_dir = model_dir + self.draft_model_dir = model_dir / "draft" + + def model_info(self): + return ModelCard( + id=self.model_dir.name, + parameters=ModelCardParameters( + max_seq_len=4096, + prompt_template="chat_template", + prompt_template_content="{{ messages }}", + ), + ) + + +class ModelListParametersTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + + self.model_dir = pathlib.Path(self.temp_dir.name) + (self.model_dir / "loaded-model").mkdir() + (self.model_dir / "other-model").mkdir() + + self.container = DummyContainer(self.model_dir / "loaded-model") + + def test_directory_listing_fills_the_loaded_model_only(self): + with patch.object(model, "container", self.container): + listing = get_model_list(self.model_dir) + + cards = {card.id: card for card in listing.data} + + self.assertEqual(cards["loaded-model"].parameters.max_seq_len, 4096) + self.assertIsNone(cards["other-model"].parameters) + + def test_listing_drops_the_template_content(self): + # Every card would otherwise carry kilobytes of Jinja + with patch.object(model, "container", self.container): + listing = get_model_list(self.model_dir) + + card = next(card for card in listing.data if card.id == "loaded-model") + + self.assertEqual(card.parameters.prompt_template, "chat_template") + self.assertIsNone(card.parameters.prompt_template_content) + + def test_directory_listing_without_a_loaded_model(self): + with patch.object(model, "container", None): + listing = get_model_list(self.model_dir) + + self.assertTrue(all(card.parameters is None for card in listing.data)) + + async def test_current_model_list_fills_parameters(self): + with patch.object(model, "container", self.container): + listing = await get_current_model_list() + + self.assertEqual(listing.data[0].parameters.max_seq_len, 4096) + + async def test_draft_list_does_not_report_the_main_model(self): + with patch.object(model, "container", self.container): + listing = await get_current_model_list(model_type="draft") + + self.assertIsNone(listing.data[0].parameters) + + +if __name__ == "__main__": + unittest.main()