def convert__anthropic_message_to_openai_chat__request(
request: anthropic_models.CreateMessageRequest,
) -> openai_models.ChatCompletionRequest:
"""Convert Anthropic CreateMessageRequest to OpenAI ChatCompletionRequest using typed models."""
openai_messages: list[dict[str, Any]] = []
# System prompt
if request.system:
if isinstance(request.system, str):
sys_content = request.system
else:
sys_content = "".join(block.text for block in request.system)
if sys_content:
openai_messages.append({"role": "system", "content": sys_content})
# User/assistant messages with text + data-url images
for msg in request.messages:
role = msg.role
content = msg.content
# Handle tool usage and results
if role == "assistant" and isinstance(content, list):
tool_calls = []
text_parts = []
for block in content:
block_type = getattr(block, "type", None)
if block_type == "tool_use":
# Type guard for ToolUseBlock
if hasattr(block, "id") and hasattr(block, "name"):
# Safely get input with fallback to empty dict
tool_input = getattr(block, "input", {}) or {}
# Ensure input is properly serialized as JSON
try:
args_str = json.dumps(tool_input)
except Exception:
args_str = json.dumps({"arguments": str(tool_input)})
tool_calls.append(
{
"id": block.id,
"type": "function",
"function": {
"name": block.name,
"arguments": args_str,
},
}
)
elif block_type == "text":
# Type guard for TextBlock
if hasattr(block, "text"):
text_parts.append(block.text)
if tool_calls:
assistant_msg: dict[str, Any] = {
"role": "assistant",
"tool_calls": tool_calls,
}
assistant_msg["content"] = " ".join(text_parts) if text_parts else None
openai_messages.append(assistant_msg)
continue
elif role == "user" and isinstance(content, list):
is_tool_result = any(
getattr(b, "type", None) == "tool_result" for b in content
)
if is_tool_result:
for block in content:
if getattr(block, "type", None) == "tool_result":
# Type guard for ToolResultBlock
if hasattr(block, "tool_use_id"):
# Get content with an empty string fallback
result_content = getattr(block, "content", "")
# Convert complex content to string representation
if not isinstance(result_content, str):
try:
if isinstance(result_content, list):
# Handle list of text blocks
text_parts = []
for part in result_content:
if (
hasattr(part, "text")
and hasattr(part, "type")
and part.type == "text"
):
text_parts.append(part.text)
if text_parts:
result_content = " ".join(text_parts)
else:
result_content = json.dumps(result_content)
else:
# Convert other non-string content to JSON
result_content = json.dumps(result_content)
except Exception:
# Fallback to string representation
result_content = str(result_content)
openai_messages.append(
{
"role": "tool",
"tool_call_id": block.tool_use_id,
"content": result_content,
}
)
continue
if isinstance(content, list):
parts: list[dict[str, Any]] = []
text_accum: list[str] = []
for block in content:
# Support both raw dicts and Anthropic model instances
if isinstance(block, dict):
btype = block.get("type")
if btype == "text" and isinstance(block.get("text"), str):
text_accum.append(block.get("text") or "")
elif btype == "image":
source = block.get("source") or {}
if (
isinstance(source, dict)
and source.get("type") == "base64"
and isinstance(source.get("media_type"), str)
and isinstance(source.get("data"), str)
):
url = f"data:{source['media_type']};base64,{source['data']}"
parts.append(
{
"type": "image_url",
"image_url": {"url": url},
}
)
else:
# Pydantic models
btype = getattr(block, "type", None)
if (
btype == "text"
and hasattr(block, "text")
and isinstance(getattr(block, "text", None), str)
):
text_accum.append(block.text or "")
elif btype == "image":
source = getattr(block, "source", None)
if (
source is not None
and getattr(source, "type", None) == "base64"
and isinstance(getattr(source, "media_type", None), str)
and isinstance(getattr(source, "data", None), str)
):
url = f"data:{source.media_type};base64,{source.data}"
parts.append(
{
"type": "image_url",
"image_url": {"url": url},
}
)
if parts or len(text_accum) > 1:
if text_accum:
parts.insert(0, {"type": "text", "text": " ".join(text_accum)})
openai_messages.append({"role": role, "content": parts})
else:
openai_messages.append(
{"role": role, "content": (text_accum[0] if text_accum else "")}
)
else:
openai_messages.append({"role": role, "content": content})
# Tools mapping (custom tools -> function tools)
tools: list[dict[str, Any]] = []
if request.tools:
for tool in request.tools:
if isinstance(tool, anthropic_models.Tool):
tools.append(
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.input_schema,
},
}
)
params: dict[str, Any] = {
"model": request.model,
"messages": openai_messages,
"max_completion_tokens": request.max_tokens,
"stream": request.stream or None,
}
if tools:
params["tools"] = tools
# tool_choice mapping
tc = request.tool_choice
if tc is not None:
tc_type = getattr(tc, "type", None)
if tc_type == "none":
params["tool_choice"] = "none"
elif tc_type == "auto":
params["tool_choice"] = "auto"
elif tc_type == "any":
params["tool_choice"] = "required"
elif tc_type == "tool":
name = getattr(tc, "name", None)
if name:
params["tool_choice"] = {
"type": "function",
"function": {"name": name},
}
# parallel_tool_calls from disable_parallel_tool_use
disable_parallel = getattr(tc, "disable_parallel_tool_use", None)
if isinstance(disable_parallel, bool):
params["parallel_tool_calls"] = not disable_parallel
# Validate against OpenAI model
return openai_models.ChatCompletionRequest.model_validate(params)