Coverage for src / lexigram / ai / relay / mappers / claude.py: 87%

1"""Anthropic Claude Messages request and response mapper. 

2 

3Converts the Claude Messages wire DTOs 

4(:class:`ClaudeRequest` / :class:`ClaudeResponse`) into the canonical 

5relay IR and back. Stream conversion is handled by the shared stream 

6lifecycle task and reports ``unsupported_feature`` until then. 

7""" 

8 

9from __future__ import annotations 

10 

11from dataclasses import replace 

12from typing import Any 

13 

14from lexigram.ai.relay.context import ConversionContext 

15from lexigram.ai.relay.errors import ( 

16 missing_required_option, 

17 translate, 

18 unsupported_feature, 

19 unsupported_format, 

20) 

21from lexigram.ai.relay.finish_reasons import ( 

22 FINISH_REASON_TO_WIRE, 

23 finish_reason_to_wire, 

24) 

25from lexigram.ai.relay.mappers.base import new_uuid, record_loss 

26from lexigram.ai.relay.media import resolve_media 

27from lexigram.contracts.ai.agents import ToolDefinition 

28from lexigram.contracts.ai.exceptions import RelayError 

29from lexigram.contracts.ai.llm import ChatMessage, FunctionCall, ToolCall 

30from lexigram.contracts.ai.multimodal import ( 

31 ContentPart, 

32 ImageBase64Part, 

33 ImageUrlPart, 

34 TextPart, 

35) 

36from lexigram.contracts.ai.relay.dto import ( 

37 ClaudeContent, 

38 ClaudeMessage, 

39 ClaudeRequest, 

40 ClaudeResponse, 

41 ClaudeUsage, 

42) 

43from lexigram.contracts.ai.relay.ir import ( 

44 RelayRequest, 

45 RelayResponse, 

46 StreamDelta, 

47 StreamState, 

48 normalize_finish_reason, 

49) 

50from lexigram.contracts.ai.relay.types import RelayFormat, RelayUsage 

51from lexigram.contracts.ai.thinking import ThinkingConfig, ThinkingResult 

52from lexigram.contracts.core.result import Err, Ok, Result 

53from lexigram.serialization import loads_str 

54 

55__all__ = ["ClaudeMapper"] 

56 

57_TARGET = RelayFormat.CLAUDE 

58 

59 

60def _tool_call_from_block(block: ClaudeContent) -> ToolCall: 

61 """Convert a Claude ``tool_use`` block into a canonical ``ToolCall``.""" 

62 return ToolCall( 

63 id=block.tool_use_id or "", 

64 type="custom", 

65 function=FunctionCall(name=block.name or "", arguments=block.input or {}), 

66 ) 

67 

68 

69def _tool_call_to_block(tool_call: ToolCall) -> ClaudeContent: 

70 """Serialize a canonical ``ToolCall`` as a Claude ``tool_use`` block.""" 

71 arguments: Any = tool_call.function.arguments if tool_call.function else {} 

72 if isinstance(arguments, str): 

73 try: 

74 arguments = loads_str(arguments) 

75 except ValueError: 

76 arguments = {} 

77 elif not isinstance(arguments, dict): 

78 arguments = {} 

79 return ClaudeContent( 

80 type="tool_use", 

81 tool_use_id=tool_call.id or f"call_{new_uuid()}", 

82 name=tool_call.function.name if tool_call.function else "", 

83 input=arguments, 

84 ) 

85 

86 

87class ClaudeMapper: 

88 """Bidirectional Anthropic Claude Messages converter. 

89 

90 Attributes: 

91 format: The wire format this mapper handles. 

92 """ 

93 

94 format = _TARGET 

95 

96 def request_to_ir( 

97 self, payload: Any, *, context: ConversionContext 

98 ) -> Result[RelayRequest, RelayError]: 

99 """Convert a ``ClaudeRequest`` into canonical ``RelayRequest``. 

100 

101 Args: 

102 payload: A wire request DTO. 

103 context: Per-conversion context with loss sink. 

104 

105 Returns: 

106 Ok(request) on success, Err(relay_error) on malformed payload. 

107 """ 

108 if not isinstance(payload, ClaudeRequest): 

109 return Err( 

110 unsupported_format( 

111 f"expected ClaudeRequest, got {type(payload).__name__}" 

112 ) 

113 ) 

114 try: 

115 tool_names: dict[str, str] = {} 

116 messages: list[ChatMessage] = [] 

117 for index, message in enumerate(payload.messages): 

118 messages.extend( 

119 self._message_to_ir(message, context, index, tool_names) 

120 ) 

121 thinking: ThinkingConfig | None = None 

122 if isinstance(payload.thinking, dict): 

123 if payload.thinking.get("type") == "enabled": 

124 thinking = ThinkingConfig( 

125 budget_tokens=int(payload.thinking.get("budget_tokens", 0) or 0) 

126 ) 

127 metadata: dict[str, Any] = {} 

128 if payload.metadata is not None: 

129 metadata["metadata"] = payload.metadata 

130 return Ok( 

131 RelayRequest( 

132 model=context.normalize_model(payload.model), 

133 messages=messages, 

134 system=self._system_to_ir(payload.system, context), 

135 tools=self._tools_to_ir(payload.tools, context), 

136 tool_choice=payload.tool_choice, 

137 temperature=payload.temperature, 

138 top_p=payload.top_p, 

139 max_tokens=payload.max_tokens, 

140 stop_sequences=list(payload.stop_sequences or []), 

141 stream=payload.stream, 

142 thinking=thinking, 

143 metadata=metadata, 

144 passthrough=dict(payload.passthrough), 

145 ) 

146 ) 

147 except (RelayError, ValueError, TypeError, KeyError) as exc: 

148 return Err(translate(exc, detail="request_to_ir")) 

149 

150 def ir_to_request( 

151 self, request: RelayRequest, *, context: ConversionContext 

152 ) -> Result[Any, RelayError]: 

153 """Convert canonical ``RelayRequest`` into a ``ClaudeRequest``. 

154 

155 Args: 

156 request: Canonical request IR. 

157 context: Per-conversion context with loss sink. 

158 

159 Returns: 

160 Ok(request) on success, Err(relay_error) on failure. 

161 """ 

162 max_tokens = request.max_tokens 

163 if max_tokens is None: 

164 max_tokens = context.max_tokens_for(request.model) 

165 if max_tokens is None: 

166 return Err(missing_required_option("claude requires max_tokens")) 

167 model = request.model 

168 temperature = request.temperature 

169 top_p = request.top_p 

170 thinking = self._thinking_from_ir(request, context) 

171 claude_options = context.options.claude 

172 if claude_options.thinking_adapter_enabled and model.endswith("-thinking"): 

173 if ( 

174 claude_options.minimum_max_tokens > 0 

175 and max_tokens < claude_options.minimum_max_tokens 

176 ): 

177 max_tokens = claude_options.minimum_max_tokens 

178 record_loss( 

179 context, 

180 field="max_tokens", 

181 target=_TARGET, 

182 reason="max_tokens_floored", 

183 ) 

184 if thinking is None and claude_options.thinking_budget_percentage > 0: 

185 thinking = { 

186 "type": "enabled", 

187 "budget_tokens": int( 

188 max_tokens * claude_options.thinking_budget_percentage / 100 

189 ), 

190 } 

191 temperature = 1.0 

192 top_p = None 

193 if ( 

194 not context.preserve_thinking_suffix(model) 

195 and not context.options.model_suffix_preserved 

196 ): 

197 model = model[: -len("-thinking")] 

198 try: 

199 messages: list[ClaudeMessage] = [] 

200 system_parts: list[str] = [] 

201 if request.system: 

202 system_parts.append(request.system) 

203 for message in request.messages: 

204 if message.role == "system": 

205 system_parts.append(self._text_from_content(message.content)) 

206 continue 

207 prepared = message 

208 if message.role == "assistant" and message.tool_calls: 

209 if any(not tool_call.id for tool_call in message.tool_calls): 

210 prepared = replace( 

211 message, 

212 tool_calls=[ 

213 tool_call 

214 if tool_call.id 

215 else replace(tool_call, id=f"call_{index + 1}") 

216 for index, tool_call in enumerate(message.tool_calls) 

217 ], 

218 ) 

219 elif message.role == "tool" and not message.tool_call_id: 

220 prepared = replace(message, tool_call_id="call_0") 

221 claude_message = self._message_from_ir(prepared, context) 

222 if claude_message.is_err(): 

223 return claude_message 

224 messages.append(claude_message.unwrap()) 

225 tool_choice = request.tool_choice 

226 if isinstance(tool_choice, str): 

227 tool_choice = {"type": tool_choice} 

228 return Ok( 

229 ClaudeRequest( 

230 model=context.resolve_model(model), 

231 max_tokens=max_tokens, 

232 messages=messages, 

233 system=( 

234 [{"type": "text", "text": "\n".join(system_parts)}] 

235 if system_parts 

236 else None 

237 ), 

238 temperature=temperature, 

239 top_p=top_p, 

240 stream=request.stream, 

241 tools=( 

242 [self._tool_from_ir(tool) for tool in request.tools] 

243 if request.tools 

244 else None 

245 ), 

246 tool_choice=tool_choice, 

247 stop_sequences=list(request.stop_sequences) or None, 

248 thinking=thinking, 

249 metadata=( 

250 request.metadata.get("metadata") 

251 if isinstance(request.metadata.get("metadata"), dict) 

252 else None 

253 ), 

254 passthrough={ 

255 **request.passthrough, 

256 **{ 

257 key: value 

258 for key, value in request.metadata.items() 

259 if key 

260 not in { 

261 "metadata", 

262 "max_tokens_kind", 

263 "generation_config", 

264 "safety_settings", 

265 "tool_config", 

266 "reasoning", 

267 "stream_options", 

268 "service_tier", 

269 } 

270 }, 

271 }, 

272 ) 

273 ) 

274 except (RelayError, ValueError, TypeError, KeyError) as exc: 

275 return Err(translate(exc, detail="ir_to_request")) 

276 

277 def response_to_ir( 

278 self, payload: Any, *, context: ConversionContext 

279 ) -> Result[RelayResponse, RelayError]: 

280 """Convert a ``ClaudeResponse`` into canonical ``RelayResponse``. 

281 

282 Args: 

283 payload: A wire response DTO. 

284 context: Per-conversion context with loss sink. 

285 

286 Returns: 

287 Ok(response) on success, Err(relay_error) on malformed payload. 

288 """ 

289 if not isinstance(payload, ClaudeResponse): 

290 return Err( 

291 unsupported_format( 

292 f"expected ClaudeResponse, got {type(payload).__name__}" 

293 ) 

294 ) 

295 try: 

296 text_parts: list[str] = [] 

297 thinking: ThinkingResult | None = None 

298 tool_calls: list[ToolCall] = [] 

299 tool_results: list[ChatMessage] = [] 

300 for block in payload.content: 

301 if block.type == "text": 

302 if block.text is not None: 

303 text_parts.append(block.text) 

304 elif block.type == "thinking": 

305 if thinking is None and block.thinking is not None: 

306 thinking = ThinkingResult( 

307 content=block.thinking, signature=block.signature 

308 ) 

309 elif block.type == "tool_use": 

310 tool_calls.append(_tool_call_from_block(block)) 

311 elif block.type == "tool_result": 

312 result_text = "".join( 

313 part.text or "" 

314 for part in (block.tool_result_content or []) 

315 if part.type == "text" 

316 ) 

317 tool_results.append( 

318 ChatMessage( 

319 role="tool", 

320 content=result_text, 

321 tool_call_id=block.tool_use_id, 

322 ) 

323 ) 

324 else: 

325 record_loss( 

326 context, 

327 field=f"content.{block.type}", 

328 target=_TARGET, 

329 reason="unknown_block_dropped", 

330 ) 

331 passthrough = dict(payload.passthrough) 

332 if payload.stop_sequence is not None: 

333 passthrough["stop_sequence"] = payload.stop_sequence 

334 return Ok( 

335 RelayResponse( 

336 model=payload.model, 

337 id=payload.id, 

338 content="".join(text_parts), 

339 thinking=thinking, 

340 tool_calls=tool_calls, 

341 tool_results=tool_results, 

342 finish_reason=normalize_finish_reason(payload.stop_reason), 

343 usage=self._usage_from_wire(payload.usage), 

344 passthrough=passthrough, 

345 ) 

346 ) 

347 except (RelayError, ValueError, TypeError, KeyError) as exc: 

348 return Err(translate(exc, detail="response_to_ir")) 

349 

350 def ir_to_response( 

351 self, response: RelayResponse, *, context: ConversionContext 

352 ) -> Result[Any, RelayError]: 

353 """Convert canonical ``RelayResponse`` into a ``ClaudeResponse``. 

354 

355 Args: 

356 response: Canonical response IR. 

357 context: Per-conversion context with loss sink. 

358 

359 Returns: 

360 Ok(response) on success, Err(relay_error) on failure. 

361 """ 

362 try: 

363 passthrough = dict(response.passthrough) 

364 stop_sequence = passthrough.pop("stop_sequence", None) 

365 blocks: list[ClaudeContent] = [] 

366 if response.content: 

367 blocks.append(ClaudeContent(type="text", text=response.content)) 

368 for tool_call in response.tool_calls: 

369 blocks.append(_tool_call_to_block(tool_call)) 

370 stop_reason: str | None = None 

371 if response.tool_calls: 

372 stop_reason = "tool_use" 

373 elif stop_sequence is not None: 

374 stop_reason = "stop_sequence" 

375 elif response.finish_reason is not None: 

376 stop_reason = self._stop_reason_from_ir(response.finish_reason, context) 

377 return Ok( 

378 ClaudeResponse( 

379 id=response.id or f"chatcmpl-{new_uuid()}", 

380 model=context.resolve_model(response.model), 

381 content=blocks, 

382 stop_reason=stop_reason, 

383 stop_sequence=stop_sequence 

384 if stop_reason == "stop_sequence" 

385 else None, 

386 usage=self._usage_to_wire(response.usage), 

387 passthrough=passthrough, 

388 ) 

389 ) 

390 except (RelayError, ValueError, TypeError, KeyError) as exc: 

391 return Err(translate(exc, detail="ir_to_response")) 

392 

393 def stream_to_delta( 

394 self, event: Any, *, state: StreamState 

395 ) -> Result[tuple[StreamDelta, ...], RelayError]: 

396 """Stream conversion is deferred to the shared stream lifecycle task.""" 

397 return Err( 

398 unsupported_feature("claude stream conversion is not implemented yet") 

399 ) 

400 

401 def delta_to_stream( 

402 self, delta: StreamDelta, *, state: StreamState 

403 ) -> Result[tuple[Any, ...], RelayError]: 

404 """Stream conversion is deferred to the shared stream lifecycle task.""" 

405 return Err( 

406 unsupported_feature("claude stream conversion is not implemented yet") 

407 ) 

408 

409 # -- helpers ------------------------------------------------------------- 

410 

411 def _message_to_ir( 

412 self, 

413 message: ClaudeMessage, 

414 context: ConversionContext, 

415 index: int, 

416 tool_names: dict[str, str], 

417 ) -> list[ChatMessage]: 

418 """Convert one Claude message into one or more canonical messages.""" 

419 if message.role == "assistant": 

420 assistant_message = self._assistant_to_ir(message) 

421 for tool_call in assistant_message.tool_calls or []: 

422 if tool_call.id and tool_call.function and tool_call.function.name: 

423 tool_names[tool_call.id] = tool_call.function.name 

424 return [assistant_message] 

425 if message.role == "user": 

426 return self._user_to_ir(message, context, index, tool_names) 

427 record_loss( 

428 context, 

429 field=f"messages[{index}].role", 

430 target=_TARGET, 

431 reason="unknown_role_dropped", 

432 ) 

433 return [] 

434 

435 def _assistant_to_ir(self, message: ClaudeMessage) -> ChatMessage: 

436 """Convert an assistant message, separating thinking/tool blocks.""" 

437 content = message.content 

438 if isinstance(content, str): 

439 content = [ClaudeContent(type="text", text=content)] 

440 text_parts: list[str] = [] 

441 thinking_blocks: list[dict[str, Any]] = [] 

442 tool_calls: list[ToolCall] = [] 

443 for block in content: 

444 if block.type == "text": 

445 if block.text is not None: 

446 text_parts.append(block.text) 

447 elif block.type == "thinking": 

448 thinking_blocks.append( 

449 { 

450 "type": "thinking", 

451 "thinking": block.thinking or "", 

452 "signature": block.signature or "", 

453 } 

454 ) 

455 elif block.type == "tool_use": 

456 tool_calls.append(_tool_call_from_block(block)) 

457 return ChatMessage( 

458 role="assistant", 

459 content="".join(text_parts), 

460 tool_calls=tool_calls or None, 

461 thinking_blocks=thinking_blocks or None, 

462 ) 

463 

464 def _user_to_ir( 

465 self, 

466 message: ClaudeMessage, 

467 context: ConversionContext, 

468 index: int, 

469 tool_names: dict[str, str], 

470 ) -> list[ChatMessage]: 

471 """Convert a user message, unwrapping tool_result blocks.""" 

472 content = message.content 

473 if isinstance(content, str): 

474 content = [ClaudeContent(type="text", text=content)] 

475 parts: list[ContentPart] = [] 

476 tool_results: list[ChatMessage] = [] 

477 has_tool_results = False 

478 has_other = False 

479 for block in content: 

480 if block.type == "tool_result": 

481 has_tool_results = True 

482 result_text = "".join( 

483 part.text or "" 

484 for part in (block.tool_result_content or []) 

485 if part.type == "text" 

486 ) 

487 metadata: dict[str, Any] | None = dict(block.passthrough) or None 

488 tool_results.append( 

489 ChatMessage( 

490 role="tool", 

491 content=result_text, 

492 tool_call_id=block.tool_use_id, 

493 name=tool_names.get(block.tool_use_id or ""), 

494 metadata=metadata, 

495 ) 

496 ) 

497 elif block.type == "text": 

498 has_other = True 

499 if block.text is not None: 

500 parts.append(TextPart(text=block.text)) 

501 elif block.type == "image": 

502 has_other = True 

503 image = self._image_to_part(block, context, index) 

504 if image is not None: 

505 parts.append(image) 

506 else: 

507 record_loss( 

508 context, 

509 field=f"messages[{index}].content.{block.type}", 

510 target=_TARGET, 

511 reason="unknown_block_dropped", 

512 ) 

513 if has_tool_results and has_other: 

514 record_loss( 

515 context, 

516 field=f"messages[{index}]", 

517 target=_TARGET, 

518 reason="mixed_user_content_reordered", 

519 ) 

520 turns: list[ChatMessage] = [] 

521 if parts: 

522 turns.append( 

523 ChatMessage( 

524 role="user", 

525 content=( 

526 parts[0].text 

527 if len(parts) == 1 and isinstance(parts[0], TextPart) 

528 else list(parts) 

529 ), 

530 ) 

531 ) 

532 turns.extend(tool_results) 

533 if not turns: 

534 record_loss( 

535 context, 

536 field=f"messages[{index}]", 

537 target=_TARGET, 

538 reason="empty_message_dropped", 

539 ) 

540 return turns 

541 

542 @staticmethod 

543 def _image_to_part( 

544 block: ClaudeContent, context: ConversionContext, index: int 

545 ) -> ContentPart | None: 

546 """Convert a Claude image block into a canonical image part.""" 

547 source = block.image_source 

548 if not isinstance(source, dict): 

549 record_loss( 

550 context, 

551 field=f"messages[{index}].image", 

552 target=_TARGET, 

553 reason="missing_source", 

554 ) 

555 return None 

556 source_type = source.get("type") 

557 if source_type == "base64": 

558 return ImageBase64Part( 

559 data=str(source.get("data", "")), 

560 media_type=str(source.get("media_type", "")), 

561 ) 

562 if source_type == "url": 

563 return ImageUrlPart(url=str(source.get("url", ""))) 

564 record_loss( 

565 context, 

566 field=f"messages[{index}].image", 

567 target=_TARGET, 

568 reason="unknown_source_type", 

569 ) 

570 return None 

571 

572 @staticmethod 

573 def _system_to_ir( 

574 system: str | list[dict[str, Any]] | None, context: ConversionContext 

575 ) -> str | None: 

576 """Normalize the Claude ``system`` field into canonical system text.""" 

577 if system is None: 

578 return None 

579 if isinstance(system, str): 

580 return system 

581 texts: list[str] = [] 

582 for block in system: 

583 if isinstance(block, dict) and block.get("type") == "text": 

584 texts.append(str(block.get("text", ""))) 

585 else: 

586 record_loss( 

587 context, 

588 field="system", 

589 target=_TARGET, 

590 reason="non_text_system_block_dropped", 

591 ) 

592 return "\n".join(texts) 

593 

594 @staticmethod 

595 def _tools_to_ir( 

596 tools: list[dict[str, Any]] | None, context: ConversionContext 

597 ) -> list[ToolDefinition]: 

598 """Convert Claude wire tools into canonical ``ToolDefinition`` objects.""" 

599 definitions: list[ToolDefinition] = [] 

600 for index, tool in enumerate(tools or []): 

601 if not isinstance(tool, dict): 

602 record_loss( 

603 context, 

604 field=f"tools[{index}]", 

605 target=_TARGET, 

606 reason="non_dict_tool_dropped", 

607 ) 

608 continue 

609 schema = tool.get("input_schema", {}) 

610 definitions.append( 

611 ToolDefinition( 

612 name=str(tool.get("name", "")), 

613 description=str(tool.get("description", "")), 

614 parameters=schema if isinstance(schema, dict) else {}, 

615 ) 

616 ) 

617 return definitions 

618 

619 def _message_from_ir( 

620 self, message: ChatMessage, context: ConversionContext 

621 ) -> Result[ClaudeMessage, RelayError]: 

622 """Convert one canonical message into a Claude message.""" 

623 if message.role == "tool": 

624 return Ok( 

625 ClaudeMessage( 

626 role="user", 

627 content=[ 

628 ClaudeContent( 

629 type="tool_result", 

630 tool_use_id=message.tool_call_id, 

631 tool_result_content=[ 

632 ClaudeContent( 

633 type="text", 

634 text=self._text_from_content(message.content), 

635 ) 

636 ], 

637 passthrough=dict(message.metadata or {}), 

638 ) 

639 ], 

640 ) 

641 ) 

642 if message.role == "assistant": 

643 blocks: list[ClaudeContent] = [] 

644 for block in message.thinking_blocks or []: 

645 if isinstance(block, dict) and block.get("type") == "thinking": 

646 blocks.append( 

647 ClaudeContent( 

648 type="thinking", 

649 thinking=str(block.get("thinking", "")), 

650 signature=( 

651 str(block["signature"]) 

652 if block.get("signature") 

653 else None 

654 ), 

655 ) 

656 ) 

657 content_blocks = self._content_to_blocks(message.content, context) 

658 if content_blocks.is_err(): 

659 return Err(content_blocks.unwrap_err()) 

660 wire_blocks = content_blocks.unwrap() 

661 has_text = any(block.type == "text" and block.text for block in wire_blocks) 

662 if not has_text: 

663 pure_tool_turn = bool( 

664 (message.metadata or {}).get("function_call_item_ids") 

665 ) 

666 if ( 

667 message.tool_calls and not pure_tool_turn 

668 ) or not message.tool_calls: 

669 wire_blocks = [ClaudeContent(type="text", text="...")] 

670 else: 

671 wire_blocks = [] 

672 blocks.extend(wire_blocks) 

673 for tool_call in message.tool_calls or []: 

674 blocks.append(_tool_call_to_block(tool_call)) 

675 return Ok( 

676 ClaudeMessage(role="assistant", content=self._collapse_content(blocks)) 

677 ) 

678 if message.role == "user": 

679 if isinstance(message.content, list) and any( 

680 isinstance(part, ImageBase64Part) for part in message.content 

681 ): 

682 return Ok(ClaudeMessage(role="user", content=[])) 

683 user_blocks = self._content_to_blocks(message.content, context) 

684 if user_blocks.is_err(): 

685 return Err(user_blocks.unwrap_err()) 

686 return Ok( 

687 ClaudeMessage( 

688 role="user", content=self._collapse_content(user_blocks.unwrap()) 

689 ) 

690 ) 

691 record_loss( 

692 context, 

693 field="messages", 

694 target=_TARGET, 

695 reason=f"unknown_role_{message.role}_dropped", 

696 ) 

697 return Ok( 

698 ClaudeMessage(role="user", content=[ClaudeContent(type="text", text="")]) 

699 ) 

700 

701 def _content_to_blocks( 

702 self, content: str | list[ContentPart], context: ConversionContext 

703 ) -> Result[list[ClaudeContent], RelayError]: 

704 """Convert canonical content into a Claude block list.""" 

705 if isinstance(content, str): 

706 return Ok([ClaudeContent(type="text", text=content)]) 

707 blocks: list[ClaudeContent] = [] 

708 for part in content: 

709 if isinstance(part, TextPart): 

710 blocks.append(ClaudeContent(type="text", text=part.text)) 

711 elif isinstance(part, ImageBase64Part): 

712 blocks.append( 

713 ClaudeContent( 

714 type="image", 

715 image_source={ 

716 "type": "base64", 

717 "media_type": part.media_type, 

718 "data": part.data, 

719 }, 

720 ) 

721 ) 

722 elif isinstance(part, ImageUrlPart): 

723 resolved = self._resolve_image(part, context) 

724 if resolved.is_err(): 

725 return Err(resolved.unwrap_err()) 

726 blocks.append( 

727 ClaudeContent( 

728 type="image", 

729 image_source={ 

730 "type": "base64", 

731 "media_type": resolved.unwrap()[0], 

732 "data": resolved.unwrap()[1], 

733 }, 

734 ) 

735 ) 

736 else: 

737 record_loss( 

738 context, 

739 field="message.content", 

740 target=_TARGET, 

741 reason="unknown_content_part", 

742 ) 

743 if not blocks: 

744 blocks.append(ClaudeContent(type="text", text="")) 

745 return Ok(blocks) 

746 

747 @staticmethod 

748 def _collapse_content( 

749 blocks: list[ClaudeContent], 

750 ) -> str | list[ClaudeContent]: 

751 """Collapse a single text block into plain string content. 

752 

753 The Claude wire protocol accepts either a plain string or a block 

754 list for message content; relaykit emits plain strings for 

755 single-text messages. 

756 """ 

757 if len(blocks) == 1 and blocks[0].type == "text" and blocks[0].text is not None: 

758 return blocks[0].text 

759 return blocks 

760 

761 @staticmethod 

762 def _resolve_image( 

763 part: ImageUrlPart, context: ConversionContext 

764 ) -> Result[tuple[str, str], RelayError]: 

765 """Resolve a URL or data-URI image for Claude. 

766 

767 Data URIs decode locally; URLs go through the context resolver. 

768 """ 

769 resolved = resolve_media( 

770 part.url, 

771 context, 

772 field="message.content", 

773 target=_TARGET, 

774 lossy=False, 

775 ) 

776 if resolved.is_err(): 

777 return Err(resolved.unwrap_err()) 

778 image = resolved.unwrap() 

779 assert image is not None # lossy=False never drops media 

780 return Ok(image) 

781 

782 @staticmethod 

783 def _text_from_content(content: str | list[ContentPart]) -> str: 

784 """Extract plain text from canonical content.""" 

785 if isinstance(content, str):