« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Request-direction conversion for the Claude mapper."""
3from __future__ import annotations
5from dataclasses import replace
6from typing import TYPE_CHECKING, Any
8from lexigram.ai.relay.context import ConversionContext
9from lexigram.ai.relay.errors import (
10 missing_required_option,
11 translate,
12 unsupported_format,
13)
14from lexigram.ai.relay.mappers.base import record_loss
15from lexigram.ai.relay.mappers.claude.utils import (
16 _TARGET,
17 _tool_call_from_block,
18)
19from lexigram.contracts.ai.agents import ToolDefinition
20from lexigram.contracts.ai.exceptions import RelayError
21from lexigram.contracts.ai.llm import ChatMessage, ToolCall
22from lexigram.contracts.ai.multimodal import (
23 ContentPart,
24 ImageBase64Part,
25 ImageUrlPart,
26 TextPart,
27)
28from lexigram.contracts.ai.relay.dto import ClaudeContent, ClaudeMessage, ClaudeRequest
29from lexigram.contracts.ai.relay.ir import RelayRequest
30from lexigram.contracts.ai.thinking import ThinkingConfig
31from lexigram.contracts.core.result import Err, Ok, Result
33if TYPE_CHECKING:
34 from lexigram.ai.relay.mappers.claude import ClaudeMapper
37class RequestMixin:
38 """Request conversion: wire ``ClaudeRequest`` to IR and back."""
40 def request_to_ir(
41 self: ClaudeMapper, payload: Any, *, context: ConversionContext
42 ) -> Result[RelayRequest, RelayError]:
43 """Convert a ``ClaudeRequest`` into canonical ``RelayRequest``.
45 Args:
46 payload: A wire request DTO.
47 context: Per-conversion context with loss sink.
49 Returns:
50 Ok(request) on success, Err(relay_error) on malformed payload.
51 """
52 if not isinstance(payload, ClaudeRequest):
53 return Err(
54 unsupported_format(
55 f"expected ClaudeRequest, got {type(payload).__name__}"
56 )
57 )
58 try:
59 tool_names: dict[str, str] = {}
60 messages: list[ChatMessage] = []
61 for index, message in enumerate(payload.messages):
62 messages.extend(
63 self._message_to_ir(message, context, index, tool_names)
64 )
65 thinking: ThinkingConfig | None = None
66 if isinstance(payload.thinking, dict):
67 if payload.thinking.get("type") == "enabled":
68 thinking = ThinkingConfig(
69 budget_tokens=int(payload.thinking.get("budget_tokens", 0) or 0)
70 )
71 metadata: dict[str, Any] = {}
72 if payload.metadata is not None:
73 metadata["metadata"] = payload.metadata
74 return Ok(
75 RelayRequest(
76 model=context.normalize_model(payload.model),
77 messages=messages,
78 system=self._system_to_ir(payload.system, context),
79 tools=self._tools_to_ir(payload.tools, context),
80 tool_choice=payload.tool_choice,
81 temperature=payload.temperature,
82 top_p=payload.top_p,
83 max_tokens=payload.max_tokens,
84 stop_sequences=list(payload.stop_sequences or []),
85 stream=payload.stream,
86 thinking=thinking,
87 metadata=metadata,
88 passthrough=dict(payload.passthrough),
89 )
90 )
91 except (RelayError, ValueError, TypeError, KeyError) as exc:
92 return Err(translate(exc, detail="request_to_ir"))
94 def ir_to_request(
95 self: ClaudeMapper, request: RelayRequest, *, context: ConversionContext
96 ) -> Result[Any, RelayError]:
97 """Convert canonical ``RelayRequest`` into a ``ClaudeRequest``.
99 Args:
100 request: Canonical request IR.
101 context: Per-conversion context with loss sink.
103 Returns:
104 Ok(request) on success, Err(relay_error) on failure.
105 """
106 max_tokens = request.max_tokens
107 if max_tokens is None:
108 max_tokens = context.max_tokens_for(request.model)
109 if max_tokens is None:
110 return Err(missing_required_option("claude requires max_tokens"))
111 model = request.model
112 temperature = request.temperature
113 top_p = request.top_p
114 thinking = self._thinking_from_ir(request, context)
115 claude_options = context.options.claude
116 if claude_options.thinking_adapter_enabled and model.endswith("-thinking"):
117 if (
118 claude_options.minimum_max_tokens > 0
119 and max_tokens < claude_options.minimum_max_tokens
120 ):
121 max_tokens = claude_options.minimum_max_tokens
122 record_loss(
123 context,
124 field="max_tokens",
125 target=_TARGET,
126 reason="max_tokens_floored",
127 )
128 if thinking is None and claude_options.thinking_budget_percentage > 0:
129 thinking = {
130 "type": "enabled",
131 "budget_tokens": int(
132 max_tokens * claude_options.thinking_budget_percentage / 100
133 ),
134 }
135 temperature = 1.0
136 top_p = None
137 if (
138 not context.preserve_thinking_suffix(model)
139 and not context.options.model_suffix_preserved
140 ):
141 model = model[: -len("-thinking")]
142 try:
143 messages: list[ClaudeMessage] = []
144 system_parts: list[str] = []
145 if request.system:
146 system_parts.append(request.system)
147 for message in request.messages:
148 if message.role == "system":
149 system_parts.append(self._text_from_content(message.content))
150 continue
151 prepared = message
152 if message.role == "assistant" and message.tool_calls:
153 if any(not tool_call.id for tool_call in message.tool_calls):
154 prepared = replace(
155 message,
156 tool_calls=[
157 tool_call
158 if tool_call.id
159 else replace(tool_call, id=f"call_{index + 1}")
160 for index, tool_call in enumerate(message.tool_calls)
161 ],
162 )
163 elif message.role == "tool" and not message.tool_call_id:
164 prepared = replace(message, tool_call_id="call_0")
165 claude_message = self._message_from_ir(prepared, context)
166 if claude_message.is_err():
167 return claude_message
168 messages.append(claude_message.unwrap())
169 tool_choice = request.tool_choice
170 if isinstance(tool_choice, str):
171 tool_choice = {"type": tool_choice}
172 return Ok(
173 ClaudeRequest(
174 model=context.resolve_model(model),
175 max_tokens=max_tokens,
176 messages=messages,
177 system=(
178 [{"type": "text", "text": "\n".join(system_parts)}]
179 if system_parts
180 else None
181 ),
182 temperature=temperature,
183 top_p=top_p,
184 stream=request.stream,
185 tools=(
186 [self._tool_from_ir(tool) for tool in request.tools]
187 if request.tools
188 else None
189 ),
190 tool_choice=tool_choice,
191 stop_sequences=list(request.stop_sequences) or None,
192 thinking=thinking,
193 metadata=(
194 request.metadata.get("metadata")
195 if isinstance(request.metadata.get("metadata"), dict)
196 else None
197 ),
198 passthrough={
199 **request.passthrough,
200 **{
201 key: value
202 for key, value in request.metadata.items()
203 if key
204 not in {
205 "metadata",
206 "max_tokens_kind",
207 "generation_config",
208 "safety_settings",
209 "tool_config",
210 "reasoning",
211 "stream_options",
212 "service_tier",
213 }
214 },
215 },
216 )
217 )
218 except (RelayError, ValueError, TypeError, KeyError) as exc:
219 return Err(translate(exc, detail="ir_to_request"))
221 def _message_to_ir(
222 self: ClaudeMapper,
223 message: ClaudeMessage,
224 context: ConversionContext,
225 index: int,
226 tool_names: dict[str, str],
227 ) -> list[ChatMessage]:
228 """Convert one Claude message into one or more canonical messages."""
229 if message.role == "assistant":
230 assistant_message = self._assistant_to_ir(message)
231 for tool_call in assistant_message.tool_calls or []:
232 if tool_call.id and tool_call.function and tool_call.function.name:
233 tool_names[tool_call.id] = tool_call.function.name
234 return [assistant_message]
235 if message.role == "user":
236 return self._user_to_ir(message, context, index, tool_names)
237 record_loss(
238 context,
239 field=f"messages[{index}].role",
240 target=_TARGET,
241 reason="unknown_role_dropped",
242 )
243 return []
245 def _assistant_to_ir(self: ClaudeMapper, message: ClaudeMessage) -> ChatMessage:
246 """Convert an assistant message, separating thinking/tool blocks."""
247 content = message.content
248 if isinstance(content, str):
249 content = [ClaudeContent(type="text", text=content)]
250 text_parts: list[str] = []
251 thinking_blocks: list[dict[str, Any]] = []
252 tool_calls: list[ToolCall] = []
253 for block in content:
254 if block.type == "text":
255 if block.text is not None:
256 text_parts.append(block.text)
257 elif block.type == "thinking":
258 thinking_blocks.append(
259 {
260 "type": "thinking",
261 "thinking": block.thinking or "",
262 "signature": block.signature or "",
263 }
264 )
265 elif block.type == "tool_use":
266 tool_calls.append(_tool_call_from_block(block))
267 return ChatMessage(
268 role="assistant",
269 content="".join(text_parts),
270 tool_calls=tool_calls or None,
271 thinking_blocks=thinking_blocks or None,
272 )
274 def _user_to_ir(
275 self: ClaudeMapper,
276 message: ClaudeMessage,
277 context: ConversionContext,
278 index: int,
279 tool_names: dict[str, str],
280 ) -> list[ChatMessage]:
281 """Convert a user message, unwrapping tool_result blocks."""
282 content = message.content
283 if isinstance(content, str):
284 content = [ClaudeContent(type="text", text=content)]
285 parts: list[ContentPart] = []
286 tool_results: list[ChatMessage] = []
287 has_tool_results = False
288 has_other = False
289 for block in content:
290 if block.type == "tool_result":
291 has_tool_results = True
292 result_text = "".join(
293 part.text or ""
294 for part in (block.tool_result_content or [])
295 if part.type == "text"
296 )
297 metadata: dict[str, Any] | None = dict(block.passthrough) or None
298 tool_results.append(
299 ChatMessage(
300 role="tool",
301 content=result_text,
302 tool_call_id=block.tool_use_id,
303 name=tool_names.get(block.tool_use_id or ""),
304 metadata=metadata,
305 )
306 )
307 elif block.type == "text":
308 has_other = True
309 if block.text is not None:
310 parts.append(TextPart(text=block.text))
311 elif block.type == "image":
312 has_other = True
313 image = self._image_to_part(block, context, index)
314 if image is not None:
315 parts.append(image)
316 else:
317 record_loss(
318 context,
319 field=f"messages[{index}].content.{block.type}",
320 target=_TARGET,
321 reason="unknown_block_dropped",
322 )
323 if has_tool_results and has_other:
324 record_loss(
325 context,
326 field=f"messages[{index}]",
327 target=_TARGET,
328 reason="mixed_user_content_reordered",
329 )
330 turns: list[ChatMessage] = []
331 if parts:
332 turns.append(
333 ChatMessage(
334 role="user",
335 content=(
336 parts[0].text
337 if len(parts) == 1 and isinstance(parts[0], TextPart)
338 else list(parts)
339 ),
340 )
341 )
342 turns.extend(tool_results)
343 if not turns:
344 record_loss(
345 context,
346 field=f"messages[{index}]",
347 target=_TARGET,
348 reason="empty_message_dropped",
349 )
350 return turns
352 @staticmethod
353 def _image_to_part(
354 block: ClaudeContent, context: ConversionContext, index: int
355 ) -> ContentPart | None:
356 """Convert a Claude image block into a canonical image part."""
357 source = block.image_source
358 if not isinstance(source, dict):
359 record_loss(
360 context,
361 field=f"messages[{index}].image",
362 target=_TARGET,
363 reason="missing_source",
364 )
365 return None
366 source_type = source.get("type")
367 if source_type == "base64":
368 return ImageBase64Part(
369 data=str(source.get("data", "")),
370 media_type=str(source.get("media_type", "")),
371 )
372 if source_type == "url":
373 return ImageUrlPart(url=str(source.get("url", "")))
374 record_loss(
375 context,
376 field=f"messages[{index}].image",
377 target=_TARGET,
378 reason="unknown_source_type",
379 )
380 return None
382 @staticmethod
383 def _system_to_ir(
384 system: str | list[dict[str, Any]] | None, context: ConversionContext
385 ) -> str | None:
386 """Normalize the Claude ``system`` field into canonical system text."""
387 if system is None:
388 return None
389 if isinstance(system, str):
390 return system
391 texts: list[str] = []
392 for block in system:
393 if isinstance(block, dict) and block.get("type") == "text":
394 texts.append(str(block.get("text", "")))
395 else:
396 record_loss(
397 context,
398 field="system",
399 target=_TARGET,
400 reason="non_text_system_block_dropped",
401 )
402 return "\n".join(texts)
404 @staticmethod
405 def _tools_to_ir(
406 tools: list[dict[str, Any]] | None, context: ConversionContext
407 ) -> list[ToolDefinition]:
408 """Convert Claude wire tools into canonical ``ToolDefinition`` objects."""
409 definitions: list[ToolDefinition] = []
410 for index, tool in enumerate(tools or []):
411 if not isinstance(tool, dict):
412 record_loss(
413 context,
414 field=f"tools[{index}]",
415 target=_TARGET,
416 reason="non_dict_tool_dropped",
417 )
418 continue
419 schema = tool.get("input_schema", {})
420 definitions.append(
421 ToolDefinition(
422 name=str(tool.get("name", "")),
423 description=str(tool.get("description", "")),
424 parameters=schema if isinstance(schema, dict) else {},
425 )
426 )
427 return definitions