Files
free-claude-code/api/models/anthropic.py
T
Alishahryar1 f3a7528d49 Major refactor: API, providers, messaging, and Anthropic protocol
Consolidates the incremental refactor work into a single change set: modular web tools (api/web_tools), native Anthropic request building and SSE block policy, OpenAI conversion and error handling, provider transports and rate limiting, messaging handler and tree queue, safe logging, smoke tests, and broad test coverage.
2026-04-26 03:01:14 -07:00

164 lines
4.8 KiB
Python

"""Pydantic models for Anthropic-compatible requests."""
from enum import StrEnum
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
# =============================================================================
# Content Block Types
# =============================================================================
class Role(StrEnum):
user = "user"
assistant = "assistant"
system = "system"
class _AnthropicBlockBase(BaseModel):
"""Pass through provider fields (e.g. ``cache_control``) for native transports."""
model_config = ConfigDict(extra="allow")
class ContentBlockText(_AnthropicBlockBase):
type: Literal["text"]
text: str
class ContentBlockImage(_AnthropicBlockBase):
type: Literal["image"]
source: dict[str, Any]
class ContentBlockToolUse(_AnthropicBlockBase):
type: Literal["tool_use"]
id: str
name: str
input: dict[str, Any]
class ContentBlockToolResult(_AnthropicBlockBase):
type: Literal["tool_result"]
tool_use_id: str
content: str | list[Any] | dict[str, Any]
class ContentBlockThinking(_AnthropicBlockBase):
type: Literal["thinking"]
thinking: str
signature: str | None = None
class ContentBlockRedactedThinking(_AnthropicBlockBase):
type: Literal["redacted_thinking"]
data: str
class ContentBlockServerToolUse(_AnthropicBlockBase):
"""Anthropic server-side tool invocation (e.g. ``web_search``, ``web_fetch``)."""
type: Literal["server_tool_use"]
id: str
name: str
input: dict[str, Any]
class ContentBlockWebSearchToolResult(_AnthropicBlockBase):
type: Literal["web_search_tool_result"]
tool_use_id: str
content: Any
class ContentBlockWebFetchToolResult(_AnthropicBlockBase):
type: Literal["web_fetch_tool_result"]
tool_use_id: str
content: Any
class SystemContent(_AnthropicBlockBase):
type: Literal["text"]
text: str
# =============================================================================
# Message Types
# =============================================================================
class Message(BaseModel):
role: Literal["user", "assistant"]
content: (
str
| list[
ContentBlockText
| ContentBlockImage
| ContentBlockToolUse
| ContentBlockToolResult
| ContentBlockThinking
| ContentBlockRedactedThinking
| ContentBlockServerToolUse
| ContentBlockWebSearchToolResult
| ContentBlockWebFetchToolResult
]
)
reasoning_content: str | None = None
class Tool(_AnthropicBlockBase):
name: str
# Anthropic server tools (e.g. web_search beta tools) include a ``type`` and
# may omit ``input_schema`` because the provider owns the schema.
type: str | None = None
description: str | None = None
input_schema: dict[str, Any] | None = None
class ThinkingConfig(BaseModel):
enabled: bool | None = True
type: str | None = None
budget_tokens: int | None = None
# =============================================================================
# Request Models
# =============================================================================
class MessagesRequest(BaseModel):
model_config = ConfigDict(extra="allow")
model: str
# Internal routing / debug: accepted on parse but not serialized to providers.
original_model: str | None = Field(default=None, exclude=True)
resolved_provider_model: str | None = Field(default=None, exclude=True)
max_tokens: int | None = None
messages: list[Message]
system: str | list[SystemContent] | None = None
stop_sequences: list[str] | None = None
stream: bool | None = True
temperature: float | None = None
top_p: float | None = None
top_k: int | None = None
metadata: dict[str, Any] | None = None
tools: list[Tool] | None = None
tool_choice: dict[str, Any] | None = None
thinking: ThinkingConfig | None = None
# Native Anthropic / SDK client hints: ignored (not forwarded) for OpenAI Chat conversion.
context_management: dict[str, Any] | None = None
output_config: dict[str, Any] | None = None
mcp_servers: list[dict[str, Any]] | None = None
extra_body: dict[str, Any] | None = None
class TokenCountRequest(BaseModel):
model_config = ConfigDict(extra="allow")
model: str
original_model: str | None = Field(default=None, exclude=True)
resolved_provider_model: str | None = Field(default=None, exclude=True)
messages: list[Message]
system: str | list[SystemContent] | None = None
tools: list[Tool] | None = None
thinking: ThinkingConfig | None = None
tool_choice: dict[str, Any] | None = None
context_management: dict[str, Any] | None = None
output_config: dict[str, Any] | None = None
mcp_servers: list[dict[str, Any]] | None = None