feat: 完善跨格式转换支持，添加 Gemini 双向转换器

- 实现 Claude/OpenAI -> Gemini 的请求和流式响应转换
- 实现 Gemini -> Claude/OpenAI 的请求转换
- 新增 ClaudeStreamConversionState 和 OpenAIStreamConversionState 状态类
- 添加 model_in_body 和 stream_in_body 元数据字段区分格式特性
- 格式转换后自动设置目标格式所需的 model/stream 字段
- 修复流式转换中 delta/choices 等字段的空值防护

src/api/handlers/base/chat_handler_base.py

@@ -245,6 +245,77 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
         """
         return request_body
 
+    def _set_model_after_conversion(
+        self,
+        request_body: Dict[str, Any],
+        provider_api_format: str,
+        mapped_model: Optional[str],
+        fallback_model: str,
+    ) -> None:
+        """
+        跨格式转换后设置 model 字段
+
+        根据目标格式的 model_in_body 属性决定是否在请求体中设置 model 字段。
+        Gemini 等格式通过 URL 路径传递模型名，不需要在请求体中设置。
+
+        Args:
+            request_body: 请求体字典（会被原地修改）
+            provider_api_format: Provider 侧 API 格式
+            mapped_model: 映射后的模型名
+            fallback_model: 兜底模型名（无映射时使用）
+        """
+        from src.core.api_format import APIFormat
+        from src.core.api_format.metadata import API_FORMAT_DEFINITIONS
+
+        try:
+            target_format = APIFormat(provider_api_format.upper())
+            target_meta = API_FORMAT_DEFINITIONS.get(target_format)
+            if target_meta and target_meta.model_in_body:
+                request_body["model"] = mapped_model or fallback_model
+        except (ValueError, KeyError):
+            # 未知格式，默认设置 model 字段
+            request_body["model"] = mapped_model or fallback_model
+
+    def _set_stream_after_conversion(
+        self,
+        request_body: Dict[str, Any],
+        client_api_format: str,
+        provider_api_format: str,
+        is_stream: bool,
+    ) -> None:
+        """
+        跨格式转换后设置 stream 字段
+
+        当客户端格式不使用 stream 字段（如 Gemini 通过 URL 端点区分流式），
+        而 Provider 格式需要 stream 字段（如 OpenAI/Claude）时，需要显式设置。
+
+        Args:
+            request_body: 请求体字典（会被原地修改）
+            client_api_format: 客户端 API 格式
+            provider_api_format: Provider 侧 API 格式
+            is_stream: 是否为流式请求
+        """
+        from src.core.api_format import APIFormat
+        from src.core.api_format.metadata import API_FORMAT_DEFINITIONS
+
+        try:
+            client_format = APIFormat(client_api_format.upper())
+            provider_format = APIFormat(provider_api_format.upper())
+
+            client_meta = API_FORMAT_DEFINITIONS.get(client_format)
+            provider_meta = API_FORMAT_DEFINITIONS.get(provider_format)
+
+            # 如果客户端格式不使用 stream 字段，但 Provider 格式需要
+            client_uses_stream = client_meta.stream_in_body if client_meta else True
+            provider_uses_stream = provider_meta.stream_in_body if provider_meta else True
+
+            if not client_uses_stream and provider_uses_stream:
+                request_body["stream"] = is_stream
+        except (ValueError, KeyError):
+            # 未知格式，保守处理：如果请求体中没有 stream 字段则设置
+            if "stream" not in request_body:
+                request_body["stream"] = is_stream
+
     async def _get_mapped_model(
         self,
         source_model: str,
@@ -404,7 +475,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
             # 透传提供商的响应头给客户端
             # 同时添加必要的 SSE 头以确保流式传输正常工作
             client_headers = filter_proxy_response_headers(ctx.response_headers)
-            # 添加/覆盖 SSE 必需的头
+            # 添加/覆盖 SSE 必需的头（所有格式统一使用 SSE）
             client_headers.update(build_sse_headers())
             client_headers["content-type"] = "text/event-stream"
 
@@ -488,6 +559,20 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                 str(client_api_format),
                 str(provider_api_format),
             )
+            # 格式转换后，为需要 model 字段的格式设置模型名
+            self._set_model_after_conversion(
+                request_body,
+                str(provider_api_format),
+                mapped_model,
+                ctx.model,
+            )
+            # 格式转换后，为需要 stream 字段的格式设置流式标志
+            self._set_stream_after_conversion(
+                request_body,
+                str(client_api_format),
+                str(provider_api_format),
+                is_stream=True,
+            )
         else:
             # 同格式：按原逻辑做轻量清理（子类可覆盖以移除不需要的字段）
             request_body = self.prepare_provider_request_body(request_body)
@@ -745,6 +830,20 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                     client_api_format,
                     provider_api_format,
                 )
+                # 格式转换后，为需要 model 字段的格式设置模型名
+                self._set_model_after_conversion(
+                    request_body,
+                    provider_api_format,
+                    mapped_model,
+                    model,
+                )
+                # 格式转换后，为需要 stream 字段的格式设置流式标志
+                self._set_stream_after_conversion(
+                    request_body,
+                    client_api_format,
+                    provider_api_format,
+                    is_stream=False,
+                )
             else:
                 # 同格式：按原逻辑做轻量清理（子类可覆盖以移除不需要的字段）
                 request_body = self.prepare_provider_request_body(request_body)
@@ -776,9 +875,6 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                 f"模型={model} -> {mapped_model or '无映射'}"
             )
             logger.debug(f"  [{self.request_id}] 请求URL: {redact_url_for_log(url)}")
-            logger.debug(
-                f"  [{self.request_id}] 请求体stream字段: {provider_payload.get('stream', 'N/A')}"
-            )
 
             # 获取复用的 HTTP 客户端（支持代理配置，从 Provider 读取）
             # 注意：使用 get_proxy_client 复用连接池，不再每次创建新客户端

src/api/handlers/base/stream_context.py

@@ -13,7 +13,12 @@ from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 
 if TYPE_CHECKING:
-    from src.core.api_format import GeminiStreamConversionState, StreamConversionState
+    from src.core.api_format import (
+        ClaudeStreamConversionState,
+        GeminiStreamConversionState,
+        OpenAIStreamConversionState,
+        StreamConversionState,
+    )
 
 
 @dataclass
@@ -91,7 +96,12 @@ class StreamContext:
 
     # 流式格式转换状态（跨 chunk 追踪）
     stream_conversion_state: Optional[
-        Union["StreamConversionState", "GeminiStreamConversionState"]
+        Union[
+            "StreamConversionState",
+            "GeminiStreamConversionState",
+            "ClaudeStreamConversionState",
+            "OpenAIStreamConversionState",
+        ]
     ] = None
 
     def reset_for_retry(self) -> None:

src/api/handlers/base/stream_processor.py

@@ -453,19 +453,32 @@ class StreamProcessor:
             if needs_conversion:
                 # 延迟导入：仅在需要转换时加载转换器模块
                 from src.core.api_format import (
+                    ClaudeStreamConversionState,
                     GeminiStreamConversionState,
+                    OpenAIStreamConversionState,
                     StreamConversionState,
                     converter_registry,
                 )
 
                 # 初始化流式转换状态（首次使用时，根据 Provider 格式选择状态类）
+                # 状态类对应 Provider 的响应格式，用于正确解析和累积流式数据
                 if ctx.stream_conversion_state is None:
                     if provider_format == "GEMINI":
                         ctx.stream_conversion_state = GeminiStreamConversionState(
                             model=ctx.mapped_model or ctx.model or "",
                             message_id=ctx.response_id or ctx.request_id or "",
                         )
+                    elif provider_format == "OPENAI":
+                        ctx.stream_conversion_state = OpenAIStreamConversionState(
+                            model=ctx.mapped_model or ctx.model or "",
+                        )
+                    elif provider_format == "CLAUDE":
+                        ctx.stream_conversion_state = ClaudeStreamConversionState(
+                            model=ctx.mapped_model or ctx.model or "",
+                            message_id=ctx.response_id or ctx.request_id or "",
+                        )
                     else:
+                        # 兜底：使用通用状态
                         ctx.stream_conversion_state = StreamConversionState(
                             model=ctx.mapped_model or ctx.model or "",
                             message_id=ctx.response_id or ctx.request_id or "",
@@ -545,6 +558,8 @@ class StreamProcessor:
                     skip_next_blank_line = True
                     out: list[bytes] = []
                     for evt in converted_events:
+                        # 统一使用 SSE 格式输出（Gemini streamGenerateContent 也使用 SSE）
+                        # 参考: https://ai.google.dev/api/generate-content
                         out.append(
                             f"data: {json.dumps(evt, ensure_ascii=False)}\n\n".encode("utf-8")
                         )

src/core/api_format/__init__.py

@@ -13,6 +13,7 @@ API 格式核心模块
 """
 
 from src.core.api_format.conversion import (
+    ClaudeStreamConversionState,
     ClaudeToGeminiConverter,
     ClaudeToOpenAIConverter,
     FormatConversionError,
@@ -20,6 +21,7 @@ from src.core.api_format.conversion import (
     GeminiStreamConversionState,
     GeminiToClaudeConverter,
     GeminiToOpenAIConverter,
+    OpenAIStreamConversionState,
     OpenAIToClaudeConverter,
     OpenAIToGeminiConverter,
     RequestConverter,
@@ -137,6 +139,8 @@ __all__ = [
     # State
     "StreamConversionState",
     "GeminiStreamConversionState",
+    "ClaudeStreamConversionState",
+    "OpenAIStreamConversionState",
     # Exceptions
     "FormatConversionError",
     # Compatibility

src/core/api_format/conversion/__init__.py

@@ -33,7 +33,9 @@ from src.core.api_format.conversion.registry import (
     converter_registry,
 )
 from src.core.api_format.conversion.state import (
+    ClaudeStreamConversionState,
     GeminiStreamConversionState,
+    OpenAIStreamConversionState,
     StreamConversionState,
 )
 from src.core.logger import logger
@@ -72,6 +74,8 @@ __all__ = [
     # State
     "StreamConversionState",
     "GeminiStreamConversionState",
+    "ClaudeStreamConversionState",
+    "OpenAIStreamConversionState",
     # Exceptions
     "FormatConversionError",
     # Compatibility

src/core/api_format/conversion/converters/claude_to_openai.py

@@ -370,7 +370,7 @@ class ClaudeToOpenAIConverter:
             return None
 
         if event_type == "content_block_delta":
-            delta_payload = event.get("delta", {})
+            delta_payload = event.get("delta") or {}
             delta_type = delta_payload.get("type")
 
             if delta_type == "text_delta":
@@ -390,7 +390,7 @@ class ClaudeToOpenAIConverter:
             return None
 
         if event_type == "message_delta":
-            delta = event.get("delta", {})
+            delta = event.get("delta") or {}
             stop_reason = delta.get("stop_reason")
             finish_reason = self.STOP_REASON_MAP.get(stop_reason, stop_reason)
             return self._base_chunk(chunk_id, model, {}, finish_reason=finish_reason)

src/core/api_format/conversion/converters/gemini.py

@@ -11,7 +11,11 @@ import time
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 if TYPE_CHECKING:
-    from src.core.api_format.conversion.state import GeminiStreamConversionState
+    from src.core.api_format.conversion.state import (
+        ClaudeStreamConversionState,
+        GeminiStreamConversionState,
+        OpenAIStreamConversionState,
+    )
 
 
 class ClaudeToGeminiConverter:
@@ -169,14 +173,335 @@ class ClaudeToGeminiConverter:
 
         return [{"function_declarations": function_declarations}]
 
+    def convert_response(self, claude_response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 Claude 响应转换为 Gemini 响应
+
+        Args:
+            claude_response: Claude 格式的响应字典
+
+        Returns:
+            Gemini 格式的响应字典
+        """
+        content_blocks = claude_response.get("content", [])
+        parts = self._convert_response_content_to_parts(content_blocks)
+
+        # 转换停止原因
+        stop_reason = claude_response.get("stop_reason")
+        finish_reason = self._convert_stop_reason_to_gemini(stop_reason)
+
+        # 转换使用量
+        usage = claude_response.get("usage", {})
+
+        return {
+            "candidates": [
+                {
+                    "content": {
+                        "parts": parts,
+                        "role": "model",
+                    },
+                    "finishReason": finish_reason,
+                    "index": 0,
+                }
+            ],
+            "usageMetadata": {
+                "promptTokenCount": usage.get("input_tokens", 0),
+                "candidatesTokenCount": usage.get("output_tokens", 0),
+                "totalTokenCount": usage.get("input_tokens", 0) + usage.get("output_tokens", 0),
+            },
+            "modelVersion": claude_response.get("model", "claude"),
+        }
+
+    def _convert_response_content_to_parts(
+        self, content_blocks: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """将 Claude content blocks 转换为 Gemini parts"""
+        parts = []
+        for block in content_blocks:
+            block_type = block.get("type")
+            if block_type == "text":
+                parts.append({"text": block.get("text", "")})
+            elif block_type == "tool_use":
+                parts.append(
+                    {
+                        "functionCall": {
+                            "name": block.get("name", ""),
+                            "args": block.get("input", {}),
+                        }
+                    }
+                )
+        return parts if parts else [{"text": ""}]
+
+    def _convert_stop_reason_to_gemini(self, stop_reason: Optional[str]) -> str:
+        """转换停止原因为 Gemini 格式"""
+        mapping = {
+            "end_turn": "STOP",
+            "max_tokens": "MAX_TOKENS",
+            "stop_sequence": "STOP",
+            "tool_use": "STOP",
+        }
+        return mapping.get(stop_reason or "", "STOP")
+
+    # ==================== 流式转换 ====================
+
+    def convert_stream_chunk(
+        self,
+        chunk: Dict[str, Any],
+        state: Optional["ClaudeStreamConversionState"] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        将 Claude 流式响应转换为 Gemini 流式响应
+
+        Args:
+            chunk: Claude SSE 事件
+            state: 流式转换状态
+
+        Returns:
+            Gemini 流式响应列表
+        """
+        from src.core.api_format.conversion.state import ClaudeStreamConversionState
+
+        if state is None:
+            state = ClaudeStreamConversionState()
+
+        events: List[Dict[str, Any]] = []
+        event_type = chunk.get("type")
+
+        if event_type == "message_start":
+            # 初始化状态
+            message = chunk.get("message", {})
+            state.model = message.get("model", "claude")
+            state.message_id = message.get("id", "msg_claude")
+
+        elif event_type == "content_block_start":
+            # 记录内容块开始
+            content_block = chunk.get("content_block", {})
+            state.current_block_type = content_block.get("type", "text")
+            state.current_block_index = chunk.get("index", 0)
+            if state.current_block_type == "tool_use":
+                state.current_tool_name = content_block.get("name", "")
+                state.current_tool_id = content_block.get("id", "")
+                state.accumulated_tool_input = ""
+
+        elif event_type == "content_block_delta":
+            delta = chunk.get("delta") or {}
+            delta_type = delta.get("type")
+
+            if delta_type == "text_delta":
+                text = delta.get("text", "")
+                if text:
+                    # 发送 Gemini 流式响应
+                    events.append(
+                        {
+                            "candidates": [
+                                {
+                                    "content": {
+                                        "parts": [{"text": text}],
+                                        "role": "model",
+                                    },
+                                    "index": 0,
+                                }
+                            ],
+                            "modelVersion": state.model,
+                        }
+                    )
+            elif delta_type == "input_json_delta":
+                # 累积工具输入
+                state.accumulated_tool_input += delta.get("partial_json", "")
+
+        elif event_type == "content_block_stop":
+            # 如果是工具调用块结束，发送工具调用
+            if state.current_block_type == "tool_use" and state.current_tool_name:
+                try:
+                    args = json.loads(state.accumulated_tool_input) if state.accumulated_tool_input else {}
+                except json.JSONDecodeError:
+                    args = {}
+                events.append(
+                    {
+                        "candidates": [
+                            {
+                                "content": {
+                                    "parts": [
+                                        {
+                                            "functionCall": {
+                                                "name": state.current_tool_name,
+                                                "args": args,
+                                            }
+                                        }
+                                    ],
+                                    "role": "model",
+                                },
+                                "index": 0,
+                            }
+                        ],
+                        "modelVersion": state.model,
+                    }
+                )
+                state.current_tool_name = ""
+                state.accumulated_tool_input = ""
+
+        elif event_type == "message_delta":
+            # 消息结束
+            delta = chunk.get("delta") or {}
+            stop_reason = delta.get("stop_reason")
+            if stop_reason:
+                finish_reason = self._convert_stop_reason_to_gemini(stop_reason)
+                events.append(
+                    {
+                        "candidates": [
+                            {
+                                "content": {"parts": [], "role": "model"},
+                                "finishReason": finish_reason,
+                                "index": 0,
+                            }
+                        ],
+                        "usageMetadata": chunk.get("usage", {}),
+                        "modelVersion": state.model,
+                    }
+                )
+
+        return events
+
 
 class GeminiToClaudeConverter:
     """
-    Gemini -> Claude 响应转换器
+    Gemini -> Claude 转换器
 
-    将 Gemini generateContent 响应转换为 Claude Messages API 格式
+    - 请求转换：将 Gemini generateContent 请求转换为 Claude Messages API 格式
+    - 响应转换：将 Gemini generateContent 响应转换为 Claude Messages API 格式
     """
 
+    def convert_request(self, gemini_request: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 Gemini 请求转换为 Claude 请求
+
+        Args:
+            gemini_request: Gemini 格式的请求字典
+
+        Returns:
+            Claude 格式的请求字典
+        """
+        claude_request: Dict[str, Any] = {
+            "messages": self._convert_contents_to_messages(gemini_request.get("contents", [])),
+        }
+
+        # 转换 system instruction
+        system_instruction = gemini_request.get("system_instruction")
+        if system_instruction:
+            parts = system_instruction.get("parts", [])
+            system_text = "".join(p.get("text", "") for p in parts if "text" in p)
+            if system_text:
+                claude_request["system"] = system_text
+
+        # 转换生成配置
+        generation_config = gemini_request.get("generation_config", {})
+        if "max_output_tokens" in generation_config:
+            claude_request["max_tokens"] = generation_config["max_output_tokens"]
+        else:
+            claude_request["max_tokens"] = 4096  # Claude 需要 max_tokens
+        if "temperature" in generation_config:
+            claude_request["temperature"] = generation_config["temperature"]
+        if "top_p" in generation_config:
+            claude_request["top_p"] = generation_config["top_p"]
+        if "top_k" in generation_config:
+            claude_request["top_k"] = generation_config["top_k"]
+        if "stop_sequences" in generation_config:
+            claude_request["stop_sequences"] = generation_config["stop_sequences"]
+
+        # 转换工具
+        tools = gemini_request.get("tools", [])
+        if tools:
+            claude_request["tools"] = self._convert_tools_to_claude(tools)
+
+        return claude_request
+
+    def _convert_contents_to_messages(
+        self, contents: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """转换 Gemini contents 为 Claude messages"""
+        messages = []
+        for content in contents:
+            role = content.get("role", "user")
+            # Gemini 使用 "model"，Claude 使用 "assistant"
+            claude_role = "assistant" if role == "model" else "user"
+
+            parts = content.get("parts", [])
+            claude_content = self._convert_parts_to_claude_content(parts)
+
+            messages.append(
+                {
+                    "role": claude_role,
+                    "content": claude_content,
+                }
+            )
+        return messages
+
+    def _convert_parts_to_claude_content(
+        self, parts: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """将 Gemini parts 转换为 Claude content blocks"""
+        content = []
+        for part in parts:
+            if "text" in part:
+                content.append(
+                    {
+                        "type": "text",
+                        "text": part["text"],
+                    }
+                )
+            elif "inline_data" in part:
+                # 转换图片
+                inline_data = part["inline_data"]
+                content.append(
+                    {
+                        "type": "image",
+                        "source": {
+                            "type": "base64",
+                            "media_type": inline_data.get("mime_type", "image/png"),
+                            "data": inline_data.get("data", ""),
+                        },
+                    }
+                )
+            elif "function_call" in part:
+                # 转换工具调用
+                func_call = part["function_call"]
+                content.append(
+                    {
+                        "type": "tool_use",
+                        "id": f"toolu_{func_call.get('name', '')}",
+                        "name": func_call.get("name", ""),
+                        "input": func_call.get("args", {}),
+                    }
+                )
+            elif "function_response" in part:
+                # 转换工具结果
+                func_response = part["function_response"]
+                result = func_response.get("response", {}).get("result", "")
+                content.append(
+                    {
+                        "type": "tool_result",
+                        "tool_use_id": func_response.get("name", ""),
+                        "content": result if isinstance(result, str) else json.dumps(result),
+                    }
+                )
+        return content if content else [{"type": "text", "text": ""}]
+
+    def _convert_tools_to_claude(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """转换 Gemini 工具为 Claude 格式"""
+        claude_tools = []
+        for tool in tools:
+            function_declarations = tool.get("function_declarations", [])
+            for func_decl in function_declarations:
+                claude_tool = {
+                    "name": func_decl.get("name", ""),
+                }
+                if "description" in func_decl:
+                    claude_tool["description"] = func_decl["description"]
+                if "parameters" in func_decl:
+                    claude_tool["input_schema"] = func_decl["parameters"]
+                claude_tools.append(claude_tool)
+        return claude_tools
+
     def convert_response(self, gemini_response: Dict[str, Any]) -> Dict[str, Any]:
         """
         将 Gemini 响应转换为 Claude 响应
@@ -303,13 +628,13 @@ class GeminiToClaudeConverter:
             state = GeminiStreamConversionState()
 
         events: List[Dict[str, Any]] = []
-        candidates = chunk.get("candidates", [])
+        candidates = chunk.get("candidates") or []
         if not candidates:
             return events
 
         candidate = candidates[0]
-        content = candidate.get("content", {})
+        content = candidate.get("content") or {}
-        parts = content.get("parts", [])
+        parts = content.get("parts") or []
 
         # 发送 message_start（首次）
         if not state.message_started:
@@ -582,14 +907,348 @@ class OpenAIToGeminiConverter:
 
         return [{"function_declarations": function_declarations}]
 
+    def convert_response(self, openai_response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 OpenAI 响应转换为 Gemini 响应
+
+        Args:
+            openai_response: OpenAI 格式的响应字典
+
+        Returns:
+            Gemini 格式的响应字典
+        """
+        choices = openai_response.get("choices", [])
+        candidates = []
+
+        for i, choice in enumerate(choices):
+            message = choice.get("message", {})
+            parts = []
+
+            # 转换文本内容
+            content = message.get("content")
+            if content:
+                parts.append({"text": content})
+
+            # 转换工具调用
+            tool_calls = message.get("tool_calls", [])
+            for tc in tool_calls:
+                if tc.get("type") == "function":
+                    func = tc.get("function", {})
+                    try:
+                        args = json.loads(func.get("arguments", "{}"))
+                    except json.JSONDecodeError:
+                        args = {}
+                    parts.append(
+                        {
+                            "functionCall": {
+                                "name": func.get("name", ""),
+                                "args": args,
+                            }
+                        }
+                    )
+
+            # 转换停止原因
+            finish_reason = self._convert_finish_reason_to_gemini(choice.get("finish_reason"))
+
+            candidates.append(
+                {
+                    "content": {
+                        "parts": parts if parts else [{"text": ""}],
+                        "role": "model",
+                    },
+                    "finishReason": finish_reason,
+                    "index": i,
+                }
+            )
+
+        # 转换使用量
+        usage = openai_response.get("usage", {})
+
+        return {
+            "candidates": candidates,
+            "usageMetadata": {
+                "promptTokenCount": usage.get("prompt_tokens", 0),
+                "candidatesTokenCount": usage.get("completion_tokens", 0),
+                "totalTokenCount": usage.get("total_tokens", 0),
+            },
+            "modelVersion": openai_response.get("model", "gpt"),
+        }
+
+    def _convert_finish_reason_to_gemini(self, finish_reason: Optional[str]) -> str:
+        """转换停止原因为 Gemini 格式"""
+        mapping = {
+            "stop": "STOP",
+            "length": "MAX_TOKENS",
+            "content_filter": "SAFETY",
+            "tool_calls": "STOP",
+            "function_call": "STOP",
+        }
+        return mapping.get(finish_reason or "", "STOP")
+
+    # ==================== 流式转换 ====================
+
+    def convert_stream_chunk(
+        self,
+        chunk: Dict[str, Any],
+        state: Optional["OpenAIStreamConversionState"] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        将 OpenAI 流式响应转换为 Gemini 流式响应
+
+        Args:
+            chunk: OpenAI chat.completion.chunk
+            state: 流式转换状态
+
+        Returns:
+            Gemini 流式响应列表
+        """
+        from src.core.api_format.conversion.state import OpenAIStreamConversionState
+
+        if state is None:
+            state = OpenAIStreamConversionState()
+
+        events: List[Dict[str, Any]] = []
+        choices = chunk.get("choices") or []
+
+        if not choices:
+            return events
+
+        choice = choices[0]
+        delta = choice.get("delta") or {}
+        finish_reason = choice.get("finish_reason")
+
+        # 记录模型
+        if chunk.get("model"):
+            state.model = chunk["model"]
+
+        # 处理文本增量
+        content = delta.get("content")
+        if content:
+            events.append(
+                {
+                    "candidates": [
+                        {
+                            "content": {
+                                "parts": [{"text": content}],
+                                "role": "model",
+                            },
+                            "index": 0,
+                        }
+                    ],
+                    "modelVersion": state.model,
+                }
+            )
+
+        # 处理工具调用
+        tool_calls = delta.get("tool_calls") or []
+        for tc in tool_calls:
+            if tc.get("function"):
+                func = tc["function"]
+                # 工具名称
+                if func.get("name"):
+                    state.current_tool_name = func["name"]
+                    state.accumulated_tool_args = ""
+                # 工具参数
+                if func.get("arguments"):
+                    state.accumulated_tool_args += func["arguments"]
+
+        # 处理结束
+        if finish_reason:
+            # 如果有累积的工具调用，先发送
+            if state.current_tool_name and state.accumulated_tool_args:
+                try:
+                    args = json.loads(state.accumulated_tool_args)
+                except json.JSONDecodeError:
+                    args = {}
+                events.append(
+                    {
+                        "candidates": [
+                            {
+                                "content": {
+                                    "parts": [
+                                        {
+                                            "functionCall": {
+                                                "name": state.current_tool_name,
+                                                "args": args,
+                                            }
+                                        }
+                                    ],
+                                    "role": "model",
+                                },
+                                "index": 0,
+                            }
+                        ],
+                        "modelVersion": state.model,
+                    }
+                )
+                state.current_tool_name = ""
+                state.accumulated_tool_args = ""
+
+            # 发送结束标记
+            gemini_finish_reason = self._convert_finish_reason_to_gemini(finish_reason)
+            events.append(
+                {
+                    "candidates": [
+                        {
+                            "content": {"parts": [], "role": "model"},
+                            "finishReason": gemini_finish_reason,
+                            "index": 0,
+                        }
+                    ],
+                    "modelVersion": state.model,
+                }
+            )
+
+        return events
+
 
 class GeminiToOpenAIConverter:
     """
-    Gemini -> OpenAI 响应转换器
+    Gemini -> OpenAI 转换器
 
-    将 Gemini generateContent 响应转换为 OpenAI Chat Completions API 格式
+    - 请求转换：将 Gemini generateContent 请求转换为 OpenAI Chat Completions API 格式
+    - 响应转换：将 Gemini generateContent 响应转换为 OpenAI Chat Completions API 格式
     """
 
+    def convert_request(self, gemini_request: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 Gemini 请求转换为 OpenAI 请求
+
+        Args:
+            gemini_request: Gemini 格式的请求字典
+
+        Returns:
+            OpenAI 格式的请求字典
+        """
+        openai_request: Dict[str, Any] = {
+            "messages": self._convert_contents_to_messages(gemini_request),
+        }
+
+        # 注意：stream 参数由调用方根据请求类型设置
+        # Gemini 通过 URL 端点区分流式/非流式（streamGenerateContent vs generateContent）
+        # OpenAI 通过请求体中的 stream 字段区分
+        # 调用方（chat_handler_base）会在格式转换后设置 stream 参数
+
+        # 转换生成配置
+        generation_config = gemini_request.get("generation_config", {})
+        if "max_output_tokens" in generation_config:
+            openai_request["max_tokens"] = generation_config["max_output_tokens"]
+        if "temperature" in generation_config:
+            openai_request["temperature"] = generation_config["temperature"]
+        if "top_p" in generation_config:
+            openai_request["top_p"] = generation_config["top_p"]
+        if "stop_sequences" in generation_config:
+            openai_request["stop"] = generation_config["stop_sequences"]
+        if "candidate_count" in generation_config:
+            openai_request["n"] = generation_config["candidate_count"]
+
+        # 转换工具
+        tools = gemini_request.get("tools", [])
+        if tools:
+            openai_request["tools"] = self._convert_tools_to_openai(tools)
+
+        return openai_request
+
+    def _convert_contents_to_messages(
+        self, gemini_request: Dict[str, Any]
+    ) -> List[Dict[str, Any]]:
+        """转换 Gemini contents 为 OpenAI messages"""
+        messages = []
+
+        # 转换 system instruction
+        system_instruction = gemini_request.get("system_instruction")
+        if system_instruction:
+            parts = system_instruction.get("parts", [])
+            system_text = "".join(p.get("text", "") for p in parts if "text" in p)
+            if system_text:
+                messages.append({"role": "system", "content": system_text})
+
+        # 转换 contents
+        for content in gemini_request.get("contents", []):
+            role = content.get("role", "user")
+            # Gemini 使用 "model"，OpenAI 使用 "assistant"
+            openai_role = "assistant" if role == "model" else "user"
+
+            parts = content.get("parts", [])
+            openai_content, tool_calls = self._convert_parts_to_openai_content(parts)
+
+            message: Dict[str, Any] = {
+                "role": openai_role,
+            }
+
+            if openai_content:
+                message["content"] = openai_content
+            if tool_calls:
+                message["tool_calls"] = tool_calls
+
+            messages.append(message)
+
+        return messages
+
+    def _convert_parts_to_openai_content(
+        self, parts: List[Dict[str, Any]]
+    ) -> tuple[Any, List[Dict[str, Any]]]:
+        """将 Gemini parts 转换为 OpenAI content 和 tool_calls"""
+        content_parts: List[Any] = []
+        tool_calls: List[Dict[str, Any]] = []
+
+        for part in parts:
+            if "text" in part:
+                content_parts.append({"type": "text", "text": part["text"]})
+            elif "inline_data" in part:
+                # 转换图片
+                inline_data = part["inline_data"]
+                mime_type = inline_data.get("mime_type", "image/png")
+                data = inline_data.get("data", "")
+                content_parts.append(
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": f"data:{mime_type};base64,{data}"},
+                    }
+                )
+            elif "function_call" in part:
+                # 转换工具调用
+                func_call = part["function_call"]
+                tool_calls.append(
+                    {
+                        "id": f"call_{func_call.get('name', '')}",
+                        "type": "function",
+                        "function": {
+                            "name": func_call.get("name", ""),
+                            "arguments": json.dumps(func_call.get("args", {})),
+                        },
+                    }
+                )
+
+        # 简化内容格式
+        if len(content_parts) == 1 and content_parts[0].get("type") == "text":
+            content = content_parts[0]["text"]
+        elif content_parts:
+            content = content_parts
+        else:
+            content = None
+
+        return content, tool_calls
+
+    def _convert_tools_to_openai(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """转换 Gemini 工具为 OpenAI 格式"""
+        openai_tools = []
+        for tool in tools:
+            function_declarations = tool.get("function_declarations", [])
+            for func_decl in function_declarations:
+                openai_tool: Dict[str, Any] = {
+                    "type": "function",
+                    "function": {
+                        "name": func_decl.get("name", ""),
+                    },
+                }
+                if "description" in func_decl:
+                    openai_tool["function"]["description"] = func_decl["description"]
+                if "parameters" in func_decl:
+                    openai_tool["function"]["parameters"] = func_decl["parameters"]
+                openai_tools.append(openai_tool)
+        return openai_tools
+
     def convert_response(self, gemini_response: Dict[str, Any]) -> Dict[str, Any]:
         """
         将 Gemini 响应转换为 OpenAI 响应
@@ -707,13 +1366,13 @@ class GeminiToOpenAIConverter:
             state = GeminiStreamConversionState()
 
         events: List[Dict[str, Any]] = []
-        candidates = chunk.get("candidates", [])
+        candidates = chunk.get("candidates") or []
         if not candidates:
             return events
 
         candidate = candidates[0]
-        content = candidate.get("content", {})
+        content = candidate.get("content") or {}
-        parts = content.get("parts", [])
+        parts = content.get("parts") or []
         finish_reason = candidate.get("finishReason")
 
         chunk_id = f"chatcmpl-{state.message_id or 'gemini'}"

src/core/api_format/conversion/converters/openai_to_claude.py

@@ -389,12 +389,12 @@ class OpenAIToClaudeConverter:
 
         events: List[Dict[str, Any]] = []
 
-        choices = chunk.get("choices", [])
+        choices = chunk.get("choices") or []
         if not choices:
             return events
 
         choice = choices[0]
-        delta = choice.get("delta", {})
+        delta = choice.get("delta") or {}
         finish_reason = choice.get("finish_reason")
 
         # 处理角色（第一个 chunk）
@@ -429,7 +429,7 @@ class OpenAIToClaudeConverter:
             )
 
         # 处理工具调用
-        tool_calls = delta.get("tool_calls", [])
+        tool_calls = delta.get("tool_calls") or []
         for tool_call in tool_calls:
             index = tool_call.get("index", 0)
 

src/core/api_format/conversion/registry.py

@@ -23,7 +23,12 @@ from src.core.metrics import format_conversion_duration_seconds, format_conversi
 from .exceptions import FormatConversionError
 
 if TYPE_CHECKING:
-    from .state import GeminiStreamConversionState, StreamConversionState
+    from .state import (
+        ClaudeStreamConversionState,
+        GeminiStreamConversionState,
+        OpenAIStreamConversionState,
+        StreamConversionState,
+    )
 
 
 @contextmanager
@@ -390,7 +395,14 @@ class FormatConverterRegistry:
         chunk: Dict[str, Any],
         source_format: str,
         target_format: str,
-        state: Optional[Union["StreamConversionState", "GeminiStreamConversionState"]] = None,
+        state: Optional[
+            Union[
+                "StreamConversionState",
+                "GeminiStreamConversionState",
+                "ClaudeStreamConversionState",
+                "OpenAIStreamConversionState",
+            ]
+        ] = None,
     ) -> list[Dict[str, Any]]:
         """
         严格模式流式块转换 - 失败时抛出异常

src/core/api_format/conversion/state.py

@@ -63,7 +63,52 @@ class GeminiStreamConversionState:
         self.has_sent_usage = False
 
 
+@dataclass
+class ClaudeStreamConversionState:
+    """
+    Claude -> Gemini 流式转换状态
+
+    用于将 Claude SSE 事件流转换为 Gemini JSON 流式响应
+    """
+
+    message_id: str = ""
+    model: str = ""
+    current_block_type: str = ""  # 当前内容块类型（text/tool_use）
+    current_block_index: int = 0
+    current_tool_name: str = ""
+    current_tool_id: str = ""
+    accumulated_tool_input: str = ""  # 累积的工具输入 JSON
+
+    def reset(self) -> None:
+        """重置状态（重试时调用）"""
+        self.current_block_type = ""
+        self.current_block_index = 0
+        self.current_tool_name = ""
+        self.current_tool_id = ""
+        self.accumulated_tool_input = ""
+
+
+@dataclass
+class OpenAIStreamConversionState:
+    """
+    OpenAI -> Gemini 流式转换状态
+
+    用于将 OpenAI SSE 事件流转换为 Gemini JSON 流式响应
+    """
+
+    model: str = ""
+    current_tool_name: str = ""
+    accumulated_tool_args: str = ""  # 累积的工具参数 JSON
+
+    def reset(self) -> None:
+        """重置状态（重试时调用）"""
+        self.current_tool_name = ""
+        self.accumulated_tool_args = ""
+
+
 __all__ = [
     "StreamConversionState",
     "GeminiStreamConversionState",
+    "ClaudeStreamConversionState",
+    "OpenAIStreamConversionState",
 ]

src/core/api_format/metadata.py

@@ -36,6 +36,8 @@ class ApiFormatDefinition:
     - auth_type: 认证类型 ("header" 直接放值, "bearer" 加 Bearer 前缀)
     - extra_headers: 该格式必须携带的额外头部（如 anthropic-version）
     - protected_keys: 不应被 extra_headers 覆盖的头部（小写）
+    - model_in_body: 是否需要在请求体中包含 model 字段（Gemini 等格式通过 URL 传递模型名）
+    - stream_in_body: 是否需要在请求体中包含 stream 字段（Gemini 等格式通过 URL 端点区分流式）
     """
 
     api_format: APIFormat
@@ -46,6 +48,8 @@ class ApiFormatDefinition:
     auth_type: str = "bearer"  # "bearer" or "header"
     extra_headers: Mapping[str, str] = field(default_factory=dict)  # 格式必须的额外头部
     protected_keys: frozenset[str] = field(default_factory=frozenset)  # 受保护的头部 key（小写）
+    model_in_body: bool = True  # 是否需要在请求体中包含 model 字段
+    stream_in_body: bool = True  # 是否需要在请求体中包含 stream 字段
 
     def iter_aliases(self) -> Iterable[str]:
         """返回大小写统一后的别名集合，包含枚举名本身。"""
@@ -112,6 +116,8 @@ _DEFINITIONS: Dict[APIFormat, ApiFormatDefinition] = {
         auth_header="x-goog-api-key",
         auth_type="header",
         protected_keys=frozenset({"x-goog-api-key", "content-type"}),
+        model_in_body=False,  # Gemini 通过 URL 路径传递模型名
+        stream_in_body=False,  # Gemini 通过 URL 端点区分流式（streamGenerateContent vs generateContent）
     ),
     APIFormat.GEMINI_CLI: ApiFormatDefinition(
         api_format=APIFormat.GEMINI_CLI,
@@ -121,6 +127,8 @@ _DEFINITIONS: Dict[APIFormat, ApiFormatDefinition] = {
         auth_header="x-goog-api-key",
         auth_type="header",
         protected_keys=frozenset({"x-goog-api-key", "content-type"}),
+        model_in_body=False,  # Gemini 通过 URL 路径传递模型名
+        stream_in_body=False,  # Gemini 通过 URL 端点区分流式
     ),
 }