Coverage for src / lexigram / contracts / ai / relay / dto / claude.py: 34%
181 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
1"""Anthropic Claude Messages wire DTO family."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from typing import Any
8from lexigram.contracts.ai.relay.dto.common import require_field
10__all__ = [
11 "ClaudeContent",
12 "ClaudeMessage",
13 "ClaudeRequest",
14 "ClaudeResponse",
15 "ClaudeStreamEvent",
16 "ClaudeUsage",
17]
20@dataclass(frozen=True)
21class ClaudeContent:
22 """A content block in a Claude message.
24 Attributes:
25 type: ``text``, ``image``, ``tool_use``, ``tool_result``, ``thinking``.
26 text: Text for ``text`` blocks.
27 thinking: Thinking text for ``thinking`` blocks.
28 signature: Thinking signature for ``thinking`` blocks.
29 tool_use_id: Tool-use id.
30 name: Tool name.
31 input: Tool call arguments.
32 image_source: ``{"type": "base64", "media_type": ..., "data": ...}`` or ``None``.
33 tool_result_content: Content of a tool result block.
34 passthrough: Unknown fields preserved verbatim.
35 """
37 type: str = "text"
38 text: str | None = None
39 thinking: str | None = None
40 signature: str | None = None
41 tool_use_id: str | None = None
42 name: str | None = None
43 input: dict[str, Any] | None = None
44 image_source: dict[str, Any] | None = None
45 tool_result_content: list[ClaudeContent] | None = None
46 passthrough: dict[str, Any] = field(default_factory=dict)
48 def to_dict(self) -> dict[str, Any]:
49 """Serialize to wire dict."""
50 data: dict[str, Any] = {**self.passthrough, "type": self.type}
51 if self.text is not None:
52 data["text"] = self.text
53 if self.thinking is not None:
54 data["thinking"] = self.thinking
55 if self.signature is not None:
56 data["signature"] = self.signature
57 if self.tool_use_id is not None:
58 if self.type == "tool_result":
59 data["tool_use_id"] = self.tool_use_id
60 else:
61 data["id"] = self.tool_use_id
62 if self.name is not None:
63 data["name"] = self.name
64 if self.input is not None:
65 data["input"] = self.input
66 if self.image_source is not None:
67 data["source"] = self.image_source
68 if self.tool_result_content is not None:
69 texts = [
70 block.text for block in self.tool_result_content if block.type == "text"
71 ]
72 if len(self.tool_result_content) == 1 and texts[0] is not None:
73 data["content"] = texts[0]
74 else:
75 data["content"] = [c.to_dict() for c in self.tool_result_content]
76 return data
78 @classmethod
79 def from_dict(cls, data: dict[str, Any]) -> ClaudeContent:
80 """Build a content block from a wire dict, capturing unknown keys."""
81 known = {
82 "type",
83 "text",
84 "thinking",
85 "signature",
86 "id",
87 "tool_use_id",
88 "name",
89 "input",
90 "source",
91 "content",
92 }
93 block_type = data.get("type", "text")
94 content = data.get("content")
95 return cls(
96 type=block_type,
97 text=data.get("text"),
98 thinking=data.get("thinking"),
99 signature=data.get("signature"),
100 tool_use_id=(
101 data.get("tool_use_id")
102 if block_type == "tool_result"
103 else data.get("id")
104 ),
105 name=data.get("name"),
106 input=data.get("input"),
107 image_source=data.get("source"),
108 tool_result_content=(
109 [ClaudeContent.from_dict(c) for c in content if isinstance(c, dict)]
110 if isinstance(content, list)
111 else (
112 [ClaudeContent(type="text", text=content)]
113 if isinstance(content, str)
114 else None
115 )
116 ),
117 passthrough={k: v for k, v in data.items() if k not in known},
118 )
121@dataclass(frozen=True)
122class ClaudeMessage:
123 """A message in Claude Messages format.
125 Attributes:
126 role: ``user`` or ``assistant``.
127 content: Plain text, or a list of content blocks.
128 """
130 role: str
131 content: str | list[ClaudeContent]
133 def to_dict(self) -> dict[str, Any]:
134 """Serialize to wire dict."""
135 if isinstance(self.content, str):
136 return {"role": self.role, "content": self.content}
137 return {"role": self.role, "content": [c.to_dict() for c in self.content]}
139 @classmethod
140 def from_dict(cls, data: dict[str, Any]) -> ClaudeMessage:
141 """Build a message from a wire dict."""
142 content = data.get("content", [])
143 if isinstance(content, str):
144 content = [{"type": "text", "text": content}]
145 return cls(
146 role=data["role"],
147 content=[ClaudeContent.from_dict(c) for c in content],
148 )
151@dataclass(frozen=True)
152class ClaudeRequest:
153 """Claude Messages request body.
155 Attributes:
156 model: Model name.
157 max_tokens: Max output tokens (required by the protocol).
158 messages: Message list.
159 system: System text, or ``None``.
160 temperature: Sampling temperature, or ``None``.
161 top_p: Nucleus sampling, or ``None``.
162 stream: Whether the caller wants a stream.
163 tools: Raw tool definitions, or ``None``.
164 tool_choice: Tool choice directive, or ``None``.
165 stop_sequences: Stop strings, or ``None``.
166 thinking: Thinking config dict, or ``None``.
167 metadata: Request metadata, or ``None``.
168 passthrough: Unknown fields preserved verbatim.
169 """
171 model: str
172 max_tokens: int
173 messages: list[ClaudeMessage]
174 system: str | list[dict[str, Any]] | None = None
175 temperature: float | None = None
176 top_p: float | None = None
177 stream: bool = False
178 tools: list[dict[str, Any]] | None = None
179 tool_choice: Any | None = None
180 stop_sequences: list[str] | None = None
181 thinking: dict[str, Any] | None = None
182 metadata: dict[str, Any] | None = None
183 passthrough: dict[str, Any] = field(default_factory=dict)
185 def to_dict(self) -> dict[str, Any]:
186 """Serialize to wire dict, omitting ``None`` optional fields."""
187 data: dict[str, Any] = {
188 **self.passthrough,
189 "model": self.model,
190 "max_tokens": self.max_tokens,
191 "messages": [m.to_dict() for m in self.messages],
192 }
193 if self.system is not None:
194 data["system"] = self.system
195 if self.temperature is not None:
196 data["temperature"] = self.temperature
197 if self.top_p is not None:
198 data["top_p"] = self.top_p
199 if self.stream:
200 data["stream"] = True
201 if self.tools is not None:
202 data["tools"] = self.tools
203 if self.tool_choice is not None:
204 data["tool_choice"] = self.tool_choice
205 if self.stop_sequences is not None:
206 data["stop_sequences"] = self.stop_sequences
207 if self.thinking is not None:
208 data["thinking"] = self.thinking
209 if self.metadata is not None:
210 data["metadata"] = self.metadata
211 return data
213 @classmethod
214 def from_dict(cls, data: dict[str, Any]) -> ClaudeRequest:
215 """Build a request from a wire dict, capturing unknown keys.
217 Raises:
218 RelayError: With code ``malformed_payload`` when ``model``,
219 ``max_tokens``, or ``messages`` is absent.
220 """
221 known = {
222 "model",
223 "max_tokens",
224 "messages",
225 "system",
226 "temperature",
227 "top_p",
228 "stream",
229 "tools",
230 "tool_choice",
231 "stop_sequences",
232 "thinking",
233 "metadata",
234 }
235 return cls(
236 model=require_field(data, "model"),
237 max_tokens=require_field(data, "max_tokens"),
238 messages=[ClaudeMessage.from_dict(m) for m in data.get("messages", [])],
239 system=data.get("system"),
240 temperature=data.get("temperature"),
241 top_p=data.get("top_p"),
242 stream=bool(data.get("stream", False)),
243 tools=data.get("tools"),
244 tool_choice=data.get("tool_choice"),
245 stop_sequences=data.get("stop_sequences"),
246 thinking=data.get("thinking"),
247 metadata=data.get("metadata"),
248 passthrough={k: v for k, v in data.items() if k not in known},
249 )
252@dataclass(frozen=True)
253class ClaudeUsage:
254 """Claude Messages usage accounting.
256 Attributes:
257 input_tokens: Input tokens.
258 output_tokens: Output tokens.
259 cache_creation_input_tokens: Cache-creation input tokens.
260 cache_read_input_tokens: Cached input tokens read.
261 passthrough: Unknown fields preserved verbatim.
262 """
264 input_tokens: int = 0
265 output_tokens: int = 0
266 cache_creation_input_tokens: int = 0
267 cache_read_input_tokens: int = 0
268 passthrough: dict[str, Any] = field(default_factory=dict)
270 @property
271 def total_tokens(self) -> int:
272 """Total tokens (input + output)."""
273 return self.input_tokens + self.output_tokens
275 def to_dict(self) -> dict[str, Any]:
276 """Serialize to wire dict."""
277 data: dict[str, Any] = {
278 **self.passthrough,
279 "input_tokens": self.input_tokens,
280 "output_tokens": self.output_tokens,
281 "cache_creation_input_tokens": self.cache_creation_input_tokens,
282 "cache_read_input_tokens": self.cache_read_input_tokens,
283 }
284 return data
286 @classmethod
287 def from_dict(cls, data: dict[str, Any]) -> ClaudeUsage:
288 """Build usage from a wire dict, capturing unknown keys."""
289 known = {
290 "input_tokens",
291 "output_tokens",
292 "cache_creation_input_tokens",
293 "cache_read_input_tokens",
294 }
295 return cls(
296 input_tokens=data.get("input_tokens", 0),
297 output_tokens=data.get("output_tokens", 0),
298 cache_creation_input_tokens=data.get("cache_creation_input_tokens", 0),
299 cache_read_input_tokens=data.get("cache_read_input_tokens", 0),
300 passthrough={k: v for k, v in data.items() if k not in known},
301 )
304@dataclass(frozen=True)
305class ClaudeResponse:
306 """Non-streamed Claude Messages response body.
308 Attributes:
309 id: Message id.
310 model: Model name.
311 content: Content blocks (text, thinking, tool_use, ...).
312 type: Object type (``message``).
313 role: Message role (``assistant``).
314 stop_reason: ``end_turn``, ``max_tokens``, ``tool_use``, etc.
315 stop_sequence: Stop sequence that ended generation, or ``None``.
316 usage: Token usage, or ``None``.
317 passthrough: Unknown fields preserved verbatim.
318 """
320 id: str
321 model: str
322 content: list[ClaudeContent] = field(default_factory=list)
323 type: str = "message"
324 role: str = "assistant"
325 stop_reason: str | None = None
326 stop_sequence: str | None = None
327 usage: ClaudeUsage | None = None
328 passthrough: dict[str, Any] = field(default_factory=dict)
330 def to_dict(self) -> dict[str, Any]:
331 """Serialize to wire dict, omitting ``None`` optional fields."""
332 data: dict[str, Any] = {
333 **self.passthrough,
334 "id": self.id,
335 "type": self.type,
336 "role": self.role,
337 "model": self.model,
338 "content": [c.to_dict() for c in self.content],
339 }
340 if self.stop_reason is not None:
341 data["stop_reason"] = self.stop_reason
342 if self.stop_sequence is not None:
343 data["stop_sequence"] = self.stop_sequence
344 if self.usage is not None:
345 data["usage"] = self.usage.to_dict()
346 return data
348 @classmethod
349 def from_dict(cls, data: dict[str, Any]) -> ClaudeResponse:
350 """Build a response from a wire dict, capturing unknown keys.
352 Raises:
353 RelayError: With code ``malformed_payload`` when ``id`` or
354 ``model`` is absent.
355 """
356 known = {
357 "id",
358 "type",
359 "role",
360 "model",
361 "content",
362 "stop_reason",
363 "stop_sequence",
364 "usage",
365 }
366 usage = data.get("usage")
367 return cls(
368 id=require_field(data, "id"),
369 type=data.get("type", "message"),
370 role=data.get("role", "assistant"),
371 model=require_field(data, "model"),
372 content=[ClaudeContent.from_dict(c) for c in data.get("content", [])],
373 stop_reason=data.get("stop_reason"),
374 stop_sequence=data.get("stop_sequence"),
375 usage=ClaudeUsage.from_dict(usage) if isinstance(usage, dict) else None,
376 passthrough={k: v for k, v in data.items() if k not in known},
377 )
380@dataclass(frozen=True)
381class ClaudeStreamEvent:
382 """One SSE event in Claude Messages streaming.
384 The ``type`` field is the discriminator: ``message_start``,
385 ``content_block_start``, ``content_block_delta``,
386 ``content_block_stop``, ``message_delta``, ``message_stop``,
387 ``ping``, or ``error``.
389 Attributes:
390 type: Event type discriminator.
391 message: ``message_start`` payload.
392 index: Content block index for block lifecycle events.
393 content_block: ``content_block_start`` block payload.
394 delta: Raw delta payload (``text_delta``, ``input_json_delta``,
395 ``thinking_delta``, message_delta).
396 usage: Usage on ``message_start`` / ``message_delta``.
397 stop_reason: Stop reason on ``message_delta``.
398 stop_sequence: Stop sequence on ``message_delta``.
399 error: Raw error payload for ``error`` events.
400 passthrough: Unknown fields preserved verbatim.
401 """
403 type: str
404 message: ClaudeResponse | None = None
405 index: int | None = None
406 content_block: ClaudeContent | None = None
407 delta: dict[str, Any] | None = None
408 usage: ClaudeUsage | None = None
409 stop_reason: str | None = None
410 stop_sequence: str | None = None
411 error: dict[str, Any] | None = None
412 passthrough: dict[str, Any] = field(default_factory=dict)
414 def to_dict(self) -> dict[str, Any]:
415 """Serialize to wire dict, omitting ``None`` optional fields."""
416 data: dict[str, Any] = {**self.passthrough, "type": self.type}
417 if self.message is not None:
418 data["message"] = self.message.to_dict()
419 if self.index is not None:
420 data["index"] = self.index
421 if self.content_block is not None:
422 data["content_block"] = self.content_block.to_dict()
423 if self.delta is not None:
424 data["delta"] = self.delta
425 if self.usage is not None:
426 data["usage"] = self.usage.to_dict()
427 if self.stop_reason is not None:
428 data["stop_reason"] = self.stop_reason
429 if self.stop_sequence is not None:
430 data["stop_sequence"] = self.stop_sequence
431 if self.error is not None:
432 data["error"] = self.error
433 return data
435 @classmethod
436 def from_dict(cls, data: dict[str, Any]) -> ClaudeStreamEvent:
437 """Build an event from a wire dict, capturing unknown keys."""
438 known = {
439 "type",
440 "message",
441 "index",
442 "content_block",
443 "delta",
444 "usage",
445 "stop_reason",
446 "stop_sequence",
447 "error",
448 }
449 message = data.get("message")
450 content_block = data.get("content_block")
451 usage = data.get("usage")
452 return cls(
453 type=data.get("type", "ping"),
454 message=ClaudeResponse.from_dict(message)
455 if isinstance(message, dict)
456 else None,
457 index=data.get("index"),
458 content_block=(
459 ClaudeContent.from_dict(content_block)
460 if isinstance(content_block, dict)
461 else None
462 ),
463 delta=data.get("delta"),
464 usage=ClaudeUsage.from_dict(usage) if isinstance(usage, dict) else None,
465 stop_reason=data.get("stop_reason"),
466 stop_sequence=data.get("stop_sequence"),
467 error=data.get("error"),
468 passthrough={k: v for k, v in data.items() if k not in known},
469 )