Jack1808/Claude_Code

1"""Message and tool format converters."""2 3import json4from typing import Any5 6 7def get_block_attr(block: Any, attr: str, default: Any = None) -> Any:8    """Get attribute from object or dict."""9    if hasattr(block, attr):10        return getattr(block, attr)11    if isinstance(block, dict):12        return block.get(attr, default)13    return default14 15 16def get_block_type(block: Any) -> str | None:17    """Get block type from object or dict."""18    return get_block_attr(block, "type")19 20 21class AnthropicToOpenAIConverter:22    """Converts Anthropic message format to OpenAI format."""23 24    @staticmethod25    def convert_messages(26        messages: list[Any],27        *,28        include_reasoning_for_openrouter: bool = False,29    ) -> list[dict[str, Any]]:30        """Convert a list of Anthropic messages to OpenAI format.31 32        When include_reasoning_for_openrouter is True, assistant messages with33        thinking blocks get reasoning_content added for OpenRouter multi-turn34        reasoning continuation.35        """36        result = []37 38        for msg in messages:39            role = msg.role40            content = msg.content41 42            if role == "system":43                text = content if isinstance(content, str) else " ".join(44                    getattr(b, "text", "") for b in content if getattr(b, "type", None) == "text"45                )46                result.append({"role": "system", "content": text})47            elif isinstance(content, str):48                result.append({"role": role, "content": content})49            elif isinstance(content, list):50                if role == "assistant":51                    result.extend(52                        AnthropicToOpenAIConverter._convert_assistant_message(53                            content,54                            include_reasoning_for_openrouter=include_reasoning_for_openrouter,55                        )56                    )57                elif role == "user":58                    result.extend(59                        AnthropicToOpenAIConverter._convert_user_message(content)60                    )61            else:62                result.append({"role": role, "content": str(content)})63 64        return result65 66    @staticmethod67    def _convert_assistant_message(68        content: list[Any],69        *,70        include_reasoning_for_openrouter: bool = False,71    ) -> list[dict[str, Any]]:72        """Convert assistant message blocks, preserving interleaved thinking+text order."""73        content_parts: list[str] = []74        thinking_parts: list[str] = []75        tool_calls: list[dict[str, Any]] = []76 77        for block in content:78            block_type = get_block_type(block)79 80            if block_type == "text":81                content_parts.append(get_block_attr(block, "text", ""))82            elif block_type == "thinking":83                thinking = get_block_attr(block, "thinking", "")84                content_parts.append(f"<think>\n{thinking}\n</think>")85                if include_reasoning_for_openrouter:86                    thinking_parts.append(thinking)87            elif block_type == "tool_use":88                tool_input = get_block_attr(block, "input", {})89                tool_calls.append(90                    {91                        "id": get_block_attr(block, "id"),92                        "type": "function",93                        "function": {94                            "name": get_block_attr(block, "name"),95                            "arguments": json.dumps(tool_input)96                            if isinstance(tool_input, dict)97                            else str(tool_input),98                        },99                    }100                )101 102        content_str = "\n\n".join(content_parts)103 104        # Ensure content is never an empty string for assistant messages105        # NIM (especially Mistral models) requires non-empty content if there are no tool calls106        if not content_str and not tool_calls:107            content_str = " "108 109        msg: dict[str, Any] = {110            "role": "assistant",111            "content": content_str,112        }113        if tool_calls:114            msg["tool_calls"] = tool_calls115        if include_reasoning_for_openrouter and thinking_parts:116            msg["reasoning_content"] = "\n".join(thinking_parts)117 118        return [msg]119 120    @staticmethod121    def _convert_user_message(content: list[Any]) -> list[dict[str, Any]]:122        """Convert user message blocks (including tool results), preserving order."""123        result: list[dict[str, Any]] = []124        text_parts: list[str] = []125 126        def flush_text() -> None:127            if text_parts:128                result.append({"role": "user", "content": "\n".join(text_parts)})129                text_parts.clear()130 131        for block in content:132            block_type = get_block_type(block)133 134            if block_type == "text":135                text_parts.append(get_block_attr(block, "text", ""))136            elif block_type == "tool_result":137                flush_text()138                tool_content = get_block_attr(block, "content", "")139                if isinstance(tool_content, list):140                    tool_content = "\n".join(141                        item.get("text", str(item))142                        if isinstance(item, dict)143                        else str(item)144                        for item in tool_content145                    )146                result.append(147                    {148                        "role": "tool",149                        "tool_call_id": get_block_attr(block, "tool_use_id"),150                        "content": str(tool_content) if tool_content else "",151                    }152                )153 154        flush_text()155        return result156 157    @staticmethod158    def convert_tools(tools: list[Any]) -> list[dict[str, Any]]:159        """Convert Anthropic tools to OpenAI format."""160        return [161            {162                "type": "function",163                "function": {164                    "name": tool.name,165                    "description": tool.description or "",166                    "parameters": tool.input_schema,167                },168            }169            for tool in tools170        ]171 172    @staticmethod173    def convert_system_prompt(system: Any) -> dict[str, str] | None:174        """Convert Anthropic system prompt to OpenAI format."""175        if isinstance(system, str):176            return {"role": "system", "content": system}177        elif isinstance(system, list):178            text_parts = [179                get_block_attr(block, "text", "")180                for block in system181                if get_block_type(block) == "text"182            ]183            if text_parts:184                return {"role": "system", "content": "\n\n".join(text_parts).strip()}185        return None186 187 188def build_base_request_body(189    request_data: Any,190    *,191    default_max_tokens: int | None = None,192    include_reasoning_for_openrouter: bool = False,193) -> dict[str, Any]:194    """Build the common parts of an OpenAI-format request body.195 196    Handles message conversion, system prompt, max_tokens, temperature,197    top_p, stop sequences, tools, and tool_choice. Provider-specific198    parameters (extra_body, penalties, NIM settings) are added by callers.199    """200    from providers.common.utils import set_if_not_none201 202    messages = AnthropicToOpenAIConverter.convert_messages(203        request_data.messages,204        include_reasoning_for_openrouter=include_reasoning_for_openrouter,205    )206 207    system = getattr(request_data, "system", None)208    if system:209        system_msg = AnthropicToOpenAIConverter.convert_system_prompt(system)210        if system_msg:211            messages.insert(0, system_msg)212 213    body: dict[str, Any] = {"model": request_data.model, "messages": messages}214 215    max_tokens = getattr(request_data, "max_tokens", None)216    set_if_not_none(body, "max_tokens", max_tokens or default_max_tokens)217    set_if_not_none(body, "temperature", getattr(request_data, "temperature", None))218    set_if_not_none(body, "top_p", getattr(request_data, "top_p", None))219 220    stop_sequences = getattr(request_data, "stop_sequences", None)221    if stop_sequences:222        body["stop"] = stop_sequences223 224    tools = getattr(request_data, "tools", None)225    if tools:226        body["tools"] = AnthropicToOpenAIConverter.convert_tools(tools)227    tool_choice = getattr(request_data, "tool_choice", None)228    if tool_choice:229        body["tool_choice"] = tool_choice230 231    return body232