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 _PromptBody(_AnthropicModel): 

164 """The prompt fields ``/v1/messages`` and its ``count_tokens`` sibling share.""" 

165 

166 model: str 

167 messages: list[AnthropicMessage] 

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

169 tools: list[AnthropicTool] | None = None 

170 tool_choice: AnthropicToolChoice | None = None 

171 thinking: AnthropicThinking | None = None 

172 

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

174 @classmethod 

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

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

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

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

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

180 return value 

181 return None 

182 

183 

184class MessagesRequest(_PromptBody): 

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

186 

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

188 ``messages_reasoning`` setting. 

189 """ 

190 

191 max_tokens: int 

192 temperature: float | None = None 

193 top_p: float | None = None 

194 top_k: int | None = None 

195 stop_sequences: list[str] | None = None 

196 stream: bool = False 

197 

198 

199class CountTokensRequest(_PromptBody): 

200 """The ``POST /v1/messages/count_tokens`` request body.""" 

201 

202 

203class CountTokensResponse(BaseModel): 

204 """The ``/v1/messages/count_tokens`` response body.""" 

205 

206 input_tokens: int 

207 

208 

209class AnthropicUsage(BaseModel): 

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

211 

212 input_tokens: int 

213 output_tokens: int 

214 # The three prompt-side counts are disjoint and sum to the whole prompt. 

215 # lilbee's engine has no paid cache write, so the creation count is always 0. 

216 cache_creation_input_tokens: int = 0 

217 cache_read_input_tokens: int = 0 

218 

219 

220class MessagesResponse(BaseModel): 

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

222 

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

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

225 because Anthropic SDK clients expect the field present. 

226 """ 

227 

228 id: str 

229 model: str 

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

231 stop_reason: str 

232 usage: AnthropicUsage 

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

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

235 stop_sequence: str | None = None