Coverage for src/lilbee/server/anthropic_api/models.py: 100%

1"""Pydantic models for the Anthropic Messages API wire shapes.""" 

2 

3from __future__ import annotations 

4 

5from enum import StrEnum 

6from typing import Annotated, Any, Literal, get_args 

7 

8from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator 

9 

10ThinkingType = Literal["enabled", "disabled"] 

11_THINKING_DISABLED = "disabled" 

12_THINKING_TYPES: frozenset[str] = frozenset(get_args(ThinkingType)) 

13 

14MIN_THINKING_BUDGET_TOKENS = 1024 

15"""Anthropic's documented minimum for ``thinking.budget_tokens``.""" 

16 

17 

18class AnthropicEventType(StrEnum): 

19 """SSE event vocabulary of the Anthropic Messages stream.""" 

20 

21 MESSAGE_START = "message_start" 

22 CONTENT_BLOCK_START = "content_block_start" 

23 CONTENT_BLOCK_DELTA = "content_block_delta" 

24 CONTENT_BLOCK_STOP = "content_block_stop" 

25 MESSAGE_DELTA = "message_delta" 

26 MESSAGE_STOP = "message_stop" 

27 PING = "ping" 

28 ERROR = "error" 

29 

30 

31class _AnthropicModel(BaseModel): 

32 """Base for request models: unknown fields parse and are ignored. 

33 

34 Anthropic clients send fields this surface does not act on (``metadata``, 

35 ``cache_control``, ``output_config``, ``betas``). Rejecting them with a 400 

36 hard-fails Claude Code, so they are tolerated instead. 

37 """ 

38 

39 model_config = ConfigDict(extra="allow") 

40 

41 

42class SystemTextBlock(_AnthropicModel): 

43 """One text block of a block-form ``system`` prompt.""" 

44 

45 type: Literal["text"] 

46 text: str 

47 

48 

49class TextBlockParam(_AnthropicModel): 

50 """Text content block inside a request message.""" 

51 

52 type: Literal["text"] 

53 text: str 

54 

55 

56class ToolUseBlockParam(_AnthropicModel): 

57 """Assistant-side tool invocation replayed in the conversation.""" 

58 

59 type: Literal["tool_use"] 

60 id: str 

61 name: str 

62 input: dict[str, Any] = Field(default_factory=dict) 

63 

64 

65class ImageBlockParam(_AnthropicModel): 

66 """Image content block; parsed so the translator can reject it clearly.""" 

67 

68 type: Literal["image"] 

69 source: dict[str, Any] = Field(default_factory=dict) 

70 

71 

72class UnknownBlockParam(_AnthropicModel): 

73 """Catch-all for block types this surface ignores (``thinking``, ...). 

74 

75 Claude Code replays ``thinking``/``redacted_thinking`` blocks from earlier 

76 assistant turns; failing validation on them would break every follow-up 

77 turn, so they parse here and the translator drops them. 

78 """ 

79 

80 type: str 

81 

82 

83class ToolResultBlockParam(_AnthropicModel): 

84 """Caller-supplied result for a prior tool_use, paired by id.""" 

85 

86 type: Literal["tool_result"] 

87 tool_use_id: str 

88 content: ( 

89 str 

90 | list[ 

91 Annotated[ 

92 TextBlockParam | ImageBlockParam | UnknownBlockParam, 

93 Field(union_mode="left_to_right"), 

94 ] 

95 ] 

96 | None 

97 ) = None 

98 is_error: bool = False 

99 

100 

101ContentBlockParam = Annotated[ 

102 TextBlockParam | ToolUseBlockParam | ToolResultBlockParam | ImageBlockParam | UnknownBlockParam, 

103 # Left-to-right keeps dispatch deterministic: each known type matches its 

104 # literal or fails fast, and anything new lands on the catch-all. 

105 Field(union_mode="left_to_right"), 

106] 

107 

108 

109class AnthropicMessage(_AnthropicModel): 

110 """One entry in the request ``messages`` list. 

111 

112 ``system`` is Anthropic's mid-conversation operator channel; Claude Code 

113 sends it routinely (mode switches, injected context), so rejecting it 

114 breaks every session after the first such turn. 

115 """ 

116 

117 role: Literal["user", "assistant", "system"] 

118 content: str | list[ContentBlockParam] 

119 

120 

121class AnthropicTool(_AnthropicModel): 

122 """Tool definition; server-tool entries parse with an empty schema.""" 

123 

124 name: str 

125 description: str | None = None 

126 input_schema: dict[str, Any] = Field(default_factory=dict) 

127 

128 

129class AnthropicToolChoice(_AnthropicModel): 

130 """Tool-choice selector; ``name`` accompanies ``type == "tool"``.""" 

131 

132 type: Literal["auto", "any", "tool", "none"] 

133 name: str | None = None 

134 

135 

136class AnthropicThinking(_AnthropicModel): 

137 """The ``thinking`` parameter: whether the model may reason on this call. 

138 

139 ``budget_tokens`` tightens the reasoning cap for this call; it never 

140 loosens it. ``1024`` is Anthropic's documented minimum. 

141 """ 

142 

143 type: ThinkingType 

144 budget_tokens: int | None = None 

145 

146 @model_validator(mode="after") 

147 def _budget_meets_the_floor(self) -> AnthropicThinking: 

148 """Hold ``enabled`` to Anthropic's minimum, and ignore a disabled budget. 

149 

150 Validating the floor per field would reject 

151 ``{"type": "disabled", "budget_tokens": 0}`` -- a request asking for no 

152 thinking at all, which is the last body that should 400. 

153 """ 

154 if self.type == _THINKING_DISABLED: 

155 object.__setattr__(self, "budget_tokens", None) 

156 elif self.budget_tokens is not None and self.budget_tokens < MIN_THINKING_BUDGET_TOKENS: 

157 raise ValueError( 

158 f"thinking.budget_tokens must be at least {MIN_THINKING_BUDGET_TOKENS}" 

159 ) 

160 return self 

161 

162 

163class MessagesRequest(_AnthropicModel): 

164 """The ``POST /v1/messages`` request body. 

165 

166 ``thinking`` picks the reasoning mode for this call, overriding the 

167 ``messages_reasoning`` setting. 

168 """ 

169 

170 model: str 

171 max_tokens: int 

172 messages: list[AnthropicMessage] 

173 system: str | list[SystemTextBlock] | None = None 

174 tools: list[AnthropicTool] | None = None 

175 tool_choice: AnthropicToolChoice | None = None 

176 temperature: float | None = None 

177 top_p: float | None = None 

178 top_k: int | None = None 

179 stop_sequences: list[str] | None = None 

180 stream: bool = False 

181 thinking: AnthropicThinking | None = None 

182 

183 @field_validator("thinking", mode="before") 

184 @classmethod 

185 def _known_thinking_shapes_only(cls, value: object) -> object: 

186 # Anthropic defines enabled/disabled today. A shape this surface does 

187 # not know falls back to the setting instead of failing the request, 

188 # because a 400 here stops the agent mid-session. 

189 if value is None or (isinstance(value, dict) and value.get("type") in _THINKING_TYPES): 

190 return value 

191 return None 

192 

193 

194class AnthropicUsage(BaseModel): 

195 """Token counts in the Anthropic response shape.""" 

196 

197 input_tokens: int 

198 output_tokens: int 

199 

200 

201class MessagesResponse(BaseModel): 

202 """The non-streaming ``/v1/messages`` response body. 

203 

204 ``content`` blocks vary in shape (``thinking``/``text``/``tool_use``), so 

205 they stay plain dicts; ``stop_sequence`` is emitted as an explicit null 

206 because Anthropic SDK clients expect the field present. 

207 """ 

208 

209 id: str 

210 model: str 

211 content: list[dict[str, Any]] 

212 stop_reason: str 

213 usage: AnthropicUsage 

214 type: Literal["message"] = "message" 

215 role: Literal["assistant"] = "assistant" 

216 stop_sequence: str | None = None