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 

7 

8from pydantic import BaseModel, ConfigDict, Field 

9 

10 

11class AnthropicEventType(StrEnum): 

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

13 

14 MESSAGE_START = "message_start" 

15 CONTENT_BLOCK_START = "content_block_start" 

16 CONTENT_BLOCK_DELTA = "content_block_delta" 

17 CONTENT_BLOCK_STOP = "content_block_stop" 

18 MESSAGE_DELTA = "message_delta" 

19 MESSAGE_STOP = "message_stop" 

20 PING = "ping" 

21 ERROR = "error" 

22 

23 

24class _AnthropicModel(BaseModel): 

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

26 

27 Anthropic clients send fields this surface does not act on (``thinking``, 

28 ``metadata``, ``cache_control``, ``output_config``, ``betas``). Rejecting 

29 them with a 400 hard-fails Claude Code, so they are tolerated instead. 

30 """ 

31 

32 model_config = ConfigDict(extra="allow") 

33 

34 

35class SystemTextBlock(_AnthropicModel): 

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

37 

38 type: Literal["text"] 

39 text: str 

40 

41 

42class TextBlockParam(_AnthropicModel): 

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

44 

45 type: Literal["text"] 

46 text: str 

47 

48 

49class ToolUseBlockParam(_AnthropicModel): 

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

51 

52 type: Literal["tool_use"] 

53 id: str 

54 name: str 

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

56 

57 

58class ImageBlockParam(_AnthropicModel): 

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

60 

61 type: Literal["image"] 

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

63 

64 

65class UnknownBlockParam(_AnthropicModel): 

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

67 

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

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

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

71 """ 

72 

73 type: str 

74 

75 

76class ToolResultBlockParam(_AnthropicModel): 

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

78 

79 type: Literal["tool_result"] 

80 tool_use_id: str 

81 content: ( 

82 str 

83 | list[ 

84 Annotated[ 

85 TextBlockParam | ImageBlockParam | UnknownBlockParam, 

86 Field(union_mode="left_to_right"), 

87 ] 

88 ] 

89 | None 

90 ) = None 

91 is_error: bool = False 

92 

93 

94ContentBlockParam = Annotated[ 

95 TextBlockParam | ToolUseBlockParam | ToolResultBlockParam | ImageBlockParam | UnknownBlockParam, 

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

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

98 Field(union_mode="left_to_right"), 

99] 

100 

101 

102class AnthropicMessage(_AnthropicModel): 

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

104 

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

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

107 breaks every session after the first such turn. 

108 """ 

109 

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

111 content: str | list[ContentBlockParam] 

112 

113 

114class AnthropicTool(_AnthropicModel): 

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

116 

117 name: str 

118 description: str | None = None 

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

120 

121 

122class AnthropicToolChoice(_AnthropicModel): 

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

124 

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

126 name: str | None = None 

127 

128 

129class MessagesRequest(_AnthropicModel): 

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

131 

132 model: str 

133 max_tokens: int 

134 messages: list[AnthropicMessage] 

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

136 tools: list[AnthropicTool] | None = None 

137 tool_choice: AnthropicToolChoice | None = None 

138 temperature: float | None = None 

139 top_p: float | None = None 

140 top_k: int | None = None 

141 stop_sequences: list[str] | None = None 

142 stream: bool = False 

143 

144 

145class AnthropicUsage(BaseModel): 

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

147 

148 input_tokens: int 

149 output_tokens: int 

150 

151 

152class MessagesResponse(BaseModel): 

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

154 

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

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

157 because Anthropic SDK clients expect the field present. 

158 """ 

159 

160 id: str 

161 model: str 

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

163 stop_reason: str 

164 usage: AnthropicUsage 

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

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

167 stop_sequence: str | None = None