refactor: 重构 API 格式相关代码到统一的 api_format 模块

- 新增 src/core/api_format/ 模块，整合所有 API 格式相关功能
  - enums.py: APIFormat 枚举定义
  - metadata.py: API 格式元数据注册
  - headers.py: 请求头处理逻辑
  - detection.py: API 格式检测
  - conversion/: 格式转换器（Claude/OpenAI/Gemini 互转）
- 删除分散在各处的旧实现
  - src/core/api_format_metadata.py
  - src/core/headers.py
  - src/api/handlers/*/converter.py
  - src/api/handlers/base/format_converter_registry.py
- 更新所有引用路径，使用新的统一模块
- 新增 cli_handler_base 中的格式转换支持

src/core/api_format/__init__.py

@@ -0,0 +1,154 @@
+"""
+API 格式核心模块
+
+统一管理 API 格式相关的枚举、元数据、工具函数和格式转换功能。
+
+模块组成：
+- enums.py: APIFormat 枚举定义
+- metadata.py: 格式元数据定义（别名、路径、认证等）
+- headers.py: 请求头处理（构建、过滤、脱敏）
+- utils.py: 工具函数（is_cli_format, get_base_format 等）
+- detection.py: 格式检测（从请求头、响应内容检测格式）
+- conversion/: 格式转换子模块
+"""
+
+from src.core.api_format.conversion import (
+    ClaudeToGeminiConverter,
+    ClaudeToOpenAIConverter,
+    FormatConversionError,
+    FormatConverterRegistry,
+    GeminiStreamConversionState,
+    GeminiToClaudeConverter,
+    GeminiToOpenAIConverter,
+    OpenAIToClaudeConverter,
+    OpenAIToGeminiConverter,
+    RequestConverter,
+    ResponseConverter,
+    StreamChunkConverter,
+    StreamConversionState,
+    converter_registry,
+    is_format_compatible,
+    register_all_converters,
+)
+from src.core.api_format.detection import (
+    detect_cli_format_from_path,
+    detect_format_and_key_from_starlette,
+    detect_format_from_request,
+    detect_format_from_response,
+)
+from src.core.api_format.enums import APIFormat
+from src.core.api_format.headers import (
+    CORE_REDACT_HEADERS,
+    HOP_BY_HOP_HEADERS,
+    RESPONSE_DROP_HEADERS,
+    SENSITIVE_HEADERS,
+    UPSTREAM_DROP_HEADERS,
+    HeaderBuilder,
+    build_adapter_base_headers,
+    build_adapter_headers,
+    build_upstream_headers,
+    detect_capabilities,
+    extract_client_api_key,
+    extract_set_headers_from_rules,
+    filter_response_headers,
+    get_adapter_protected_keys,
+    get_extra_headers_from_endpoint,
+    get_header_value,
+    merge_headers_with_protection,
+    normalize_headers,
+    redact_headers_for_log,
+)
+from src.core.api_format.metadata import (
+    API_FORMAT_DEFINITIONS,
+    ApiFormatDefinition,
+    get_api_format_definition,
+    get_auth_config,
+    get_default_path,
+    get_extra_headers,
+    get_local_path,
+    get_protected_keys,
+    is_cli_api_format,
+    list_api_format_definitions,
+    register_api_format_definition,
+    resolve_api_format,
+    resolve_api_format_alias,
+)
+from src.core.api_format.utils import (
+    get_base_format,
+    is_cli_format,
+    is_convertible_format,
+    is_same_format,
+    normalize_format,
+)
+
+__all__ = [
+    # Enums
+    "APIFormat",
+    # Metadata
+    "ApiFormatDefinition",
+    "API_FORMAT_DEFINITIONS",
+    "get_api_format_definition",
+    "list_api_format_definitions",
+    "resolve_api_format",
+    "resolve_api_format_alias",
+    "register_api_format_definition",
+    "get_default_path",
+    "get_local_path",
+    "get_auth_config",
+    "get_extra_headers",
+    "get_protected_keys",
+    "is_cli_api_format",
+    # Utils
+    "is_cli_format",
+    "get_base_format",
+    "normalize_format",
+    "is_same_format",
+    "is_convertible_format",
+    # Headers
+    "UPSTREAM_DROP_HEADERS",
+    "CORE_REDACT_HEADERS",
+    "HOP_BY_HOP_HEADERS",
+    "RESPONSE_DROP_HEADERS",
+    "SENSITIVE_HEADERS",
+    "normalize_headers",
+    "get_header_value",
+    "extract_client_api_key",
+    "detect_capabilities",
+    "HeaderBuilder",
+    "build_upstream_headers",
+    "merge_headers_with_protection",
+    "filter_response_headers",
+    "redact_headers_for_log",
+    "build_adapter_base_headers",
+    "build_adapter_headers",
+    "get_adapter_protected_keys",
+    "extract_set_headers_from_rules",
+    "get_extra_headers_from_endpoint",
+    # Registry
+    "FormatConverterRegistry",
+    "converter_registry",
+    "register_all_converters",
+    # Protocols
+    "RequestConverter",
+    "ResponseConverter",
+    "StreamChunkConverter",
+    # State
+    "StreamConversionState",
+    "GeminiStreamConversionState",
+    # Exceptions
+    "FormatConversionError",
+    # Compatibility
+    "is_format_compatible",
+    # Detection
+    "detect_format_from_request",
+    "detect_format_and_key_from_starlette",
+    "detect_format_from_response",
+    "detect_cli_format_from_path",
+    # Converters
+    "OpenAIToClaudeConverter",
+    "ClaudeToOpenAIConverter",
+    "ClaudeToGeminiConverter",
+    "GeminiToClaudeConverter",
+    "OpenAIToGeminiConverter",
+    "GeminiToOpenAIConverter",
+]

src/core/api_format/conversion/__init__.py

@@ -0,0 +1,86 @@
+"""
+格式转换核心模块
+
+该目录用于承载与「API 格式转换」相关的核心能力（不依赖 FastAPI/Handler 层），
+以便 services/core 可复用并避免出现 services -> api 的反向依赖。
+
+模块组成：
+- registry.py: 转换器注册表，管理转换器实例和能力查询
+- protocols.py: 转换器协议定义（Protocol）
+- state.py: 流式转换状态类
+- exceptions.py: 转换异常定义
+- compatibility.py: 格式兼容性检查函数
+- converters/: 内置格式转换器实现
+"""
+
+from src.core.api_format.conversion.compatibility import is_format_compatible
+from src.core.api_format.conversion.converters import (
+    ClaudeToGeminiConverter,
+    ClaudeToOpenAIConverter,
+    GeminiToClaudeConverter,
+    GeminiToOpenAIConverter,
+    OpenAIToClaudeConverter,
+    OpenAIToGeminiConverter,
+)
+from src.core.api_format.conversion.exceptions import FormatConversionError
+from src.core.api_format.conversion.protocols import (
+    RequestConverter,
+    ResponseConverter,
+    StreamChunkConverter,
+)
+from src.core.api_format.conversion.registry import (
+    FormatConverterRegistry,
+    converter_registry,
+)
+from src.core.api_format.conversion.state import (
+    GeminiStreamConversionState,
+    StreamConversionState,
+)
+from src.core.logger import logger
+
+
+def register_all_converters() -> None:
+    """
+    注册所有内置的格式转换器
+
+    在应用启动时调用此函数
+    """
+    # Claude <-> OpenAI
+    converter_registry.register("OPENAI", "CLAUDE", OpenAIToClaudeConverter())
+    converter_registry.register("CLAUDE", "OPENAI", ClaudeToOpenAIConverter())
+
+    # Claude <-> Gemini
+    converter_registry.register("CLAUDE", "GEMINI", ClaudeToGeminiConverter())
+    converter_registry.register("GEMINI", "CLAUDE", GeminiToClaudeConverter())
+
+    # OpenAI <-> Gemini
+    converter_registry.register("OPENAI", "GEMINI", OpenAIToGeminiConverter())
+    converter_registry.register("GEMINI", "OPENAI", GeminiToOpenAIConverter())
+
+    logger.info(f"[ConverterRegistry] 已注册 {len(converter_registry.list_converters())} 个格式转换器")
+
+
+__all__ = [
+    # Registry
+    "FormatConverterRegistry",
+    "converter_registry",
+    "register_all_converters",
+    # Protocols
+    "RequestConverter",
+    "ResponseConverter",
+    "StreamChunkConverter",
+    # State
+    "StreamConversionState",
+    "GeminiStreamConversionState",
+    # Exceptions
+    "FormatConversionError",
+    # Compatibility
+    "is_format_compatible",
+    # Converters
+    "OpenAIToClaudeConverter",
+    "ClaudeToOpenAIConverter",
+    "ClaudeToGeminiConverter",
+    "GeminiToClaudeConverter",
+    "OpenAIToGeminiConverter",
+    "GeminiToOpenAIConverter",
+]

src/core/api_format/conversion/compatibility.py

@@ -0,0 +1,103 @@
+"""
+格式兼容性检查
+
+用于候选筛选时判断端点是否可以处理客户端请式。
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Optional, Tuple
+
+from src.core.api_format.utils import is_cli_format
+
+if TYPE_CHECKING:
+    from src.core.api_format.conversion.registry import FormatConverterRegistry
+
+logger = logging.getLogger(__name__)
+
+
+def is_format_compatible(
+    client_format: str,
+    endpoint_api_format: str,
+    endpoint_format_acceptance_config: Optional[dict],
+    is_stream: bool,
+    global_conversion_enabled: bool,
+    registry: Optional["FormatConverterRegistry"] = None,
+) -> Tuple[bool, bool, Optional[str]]:
+    """
+    检查端点是否兼容客户端格式
+
+    Args:
+        client_format: 客户端请求格式
+        endpoint_api_format: 端点的 API 格式
+        endpoint_format_acceptance_config: 端点的格式接受配置
+        is_stream: 是否是流式请求
+        global_conversion_enabled: 全局格式转换开关
+        registry: 转换器注册表（可选，默认使用全局单例）
+
+    Returns:
+        (is_compatible, needs_conversion, skip_reason)
+        - is_compatible: 是否兼容
+        - needs_conversion: 是否需要转换
+        - skip_reason: 不兼容时的原因
+    """
+    # 延迟导入避免循环依赖
+    if registry is None:
+        from src.core.api_format.conversion.registry import converter_registry
+
+        registry = converter_registry
+
+    provider_format = endpoint_api_format.upper()
+    client_format_upper = client_format.upper()
+
+    # 1. 格式完全匹配 -> 兼容，无需转换
+    if provider_format == client_format_upper:
+        return True, False, None
+
+    # 2. CLI 格式不参与转换
+    if is_cli_format(client_format_upper):
+        return False, False, "CLI 格式不支持转换"
+    if is_cli_format(provider_format):
+        return False, False, "Provider 为 CLI 格式，不支持转换"
+
+    # 3. 检查全局开关
+    if not global_conversion_enabled:
+        return False, False, "全局格式转换未启用"
+
+    # 4. 检查端点配置
+    if endpoint_format_acceptance_config is None:
+        return False, False, "端点未配置格式转换"
+
+    config = endpoint_format_acceptance_config
+    if not config.get("enabled", False):
+        return False, False, "端点格式转换未启用"
+
+    # 检查 reject_formats（优先）
+    reject_formats = config.get("reject_formats", [])
+    if client_format_upper in [f.upper() for f in reject_formats]:
+        return False, False, f"端点拒绝 {client_format} 格式"
+
+    # 检查 accept_formats
+    accept_formats = config.get("accept_formats", [])
+    if accept_formats and client_format_upper not in [f.upper() for f in accept_formats]:
+        return False, False, f"端点不接受 {client_format} 格式"
+
+    # 检查流式转换
+    if is_stream and not config.get("stream_conversion", True):
+        return False, False, "端点不支持流式格式转换"
+
+    # 5. 检查转换器能力
+    if not registry.can_convert_full(
+        client_format_upper,
+        provider_format,
+        require_stream=is_stream,
+    ):
+        return False, False, f"不存在 {client_format} <-> {provider_format} 的完整转换器"
+
+    return True, True, None
+
+
+__all__ = [
+    "is_format_compatible",
+]

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

@@ -0,0 +1,26 @@
+"""
+格式转换器集合
+
+将 Claude/OpenAI/Gemini 格式之间的转换器统一导出。
+"""
+
+from .claude_to_openai import ClaudeToOpenAIConverter
+from .gemini import (
+    ClaudeToGeminiConverter,
+    GeminiToClaudeConverter,
+    GeminiToOpenAIConverter,
+    OpenAIToGeminiConverter,
+)
+from .openai_to_claude import OpenAIToClaudeConverter
+
+__all__ = [
+    # OpenAI <-> Claude
+    "OpenAIToClaudeConverter",
+    "ClaudeToOpenAIConverter",
+    # Claude <-> Gemini
+    "ClaudeToGeminiConverter",
+    "GeminiToClaudeConverter",
+    # OpenAI <-> Gemini
+    "OpenAIToGeminiConverter",
+    "GeminiToOpenAIConverter",
+]

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

@@ -0,0 +1,450 @@
+"""
+Claude -> OpenAI 格式转换器
+
+将 Claude Messages API 格式转换为 OpenAI Chat Completions API 格式。
+"""
+
+from __future__ import annotations
+
+import json
+import time
+import uuid
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+
+if TYPE_CHECKING:
+    from src.core.api_format.conversion.state import StreamConversionState
+
+
+class ClaudeToOpenAIConverter:
+    """
+    Claude -> OpenAI 格式转换器
+
+    支持：
+    - 请求转换：Claude Request -> OpenAI Chat Request
+    - 响应转换：Claude Response -> OpenAI Chat Response
+    - 流式转换：Claude SSE -> OpenAI SSE
+    """
+
+    # 内容类型常量
+    CONTENT_TYPE_TEXT = "text"
+    CONTENT_TYPE_IMAGE = "image"
+    CONTENT_TYPE_TOOL_USE = "tool_use"
+    CONTENT_TYPE_TOOL_RESULT = "tool_result"
+
+    # 停止原因映射
+    STOP_REASON_MAP = {
+        "end_turn": "stop",
+        "max_tokens": "length",
+        "stop_sequence": "stop",
+        "tool_use": "tool_calls",
+    }
+
+    def __init__(self, model_mapping: Optional[Dict[str, str]] = None):
+        """
+        Args:
+            model_mapping: Claude 模型到 OpenAI 模型的映射
+        """
+        self._model_mapping = model_mapping or {}
+
+    # ==================== 请求转换 ====================
+
+    def convert_request(self, request: Union[Dict[str, Any], Any]) -> Dict[str, Any]:
+        """
+        将 Claude 请求转换为 OpenAI 格式
+
+        Args:
+            request: Claude 请求（Dict 或 Pydantic 模型）
+
+        Returns:
+            OpenAI 格式的请求字典
+        """
+        if hasattr(request, "model_dump"):
+            data = request.model_dump(exclude_none=True)
+        else:
+            data = dict(request)
+
+        # 模型映射
+        model = data.get("model", "")
+        openai_model = self._model_mapping.get(model, model)
+
+        # 构建消息列表
+        messages: List[Dict[str, Any]] = []
+
+        # 处理 system 消息
+        system_content = self._extract_text_content(data.get("system"))
+        if system_content:
+            messages.append({"role": "system", "content": system_content})
+
+        # 处理对话消息
+        for message in data.get("messages", []):
+            converted = self._convert_message(message)
+            if converted:
+                messages.append(converted)
+
+        # 构建 OpenAI 请求
+        result: Dict[str, Any] = {
+            "model": openai_model,
+            "messages": messages,
+        }
+
+        # 可选参数
+        if data.get("max_tokens"):
+            result["max_tokens"] = data["max_tokens"]
+        if data.get("temperature") is not None:
+            result["temperature"] = data["temperature"]
+        if data.get("top_p") is not None:
+            result["top_p"] = data["top_p"]
+        if data.get("stream"):
+            result["stream"] = data["stream"]
+        if data.get("stop_sequences"):
+            result["stop"] = data["stop_sequences"]
+
+        # 工具转换
+        tools = self._convert_tools(data.get("tools"))
+        if tools:
+            result["tools"] = tools
+
+        tool_choice = self._convert_tool_choice(data.get("tool_choice"))
+        if tool_choice:
+            result["tool_choice"] = tool_choice
+
+        return result
+
+    def _convert_message(self, message: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        """转换单条消息"""
+        role = message.get("role")
+
+        if role == "user":
+            return self._convert_user_message(message)
+        if role == "assistant":
+            return self._convert_assistant_message(message)
+
+        return None
+
+    def _convert_user_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        """转换用户消息"""
+        content = message.get("content")
+
+        if isinstance(content, str):
+            return {"role": "user", "content": content}
+
+        openai_content: List[Dict[str, Any]] = []
+        for block in content or []:
+            block_type = block.get("type")
+
+            if block_type == self.CONTENT_TYPE_TEXT:
+                openai_content.append({"type": "text", "text": block.get("text", "")})
+            elif block_type == self.CONTENT_TYPE_IMAGE:
+                source = block.get("source", {})
+                media_type = source.get("media_type", "image/jpeg")
+                data = source.get("data", "")
+                openai_content.append(
+                    {"type": "image_url", "image_url": {"url": f"data:{media_type};base64,{data}"}}
+                )
+            elif block_type == self.CONTENT_TYPE_TOOL_RESULT:
+                tool_content = block.get("content", "")
+                rendered = self._render_tool_content(tool_content)
+                openai_content.append({"type": "text", "text": f"Tool result: {rendered}"})
+
+        # 简化单文本内容
+        if len(openai_content) == 1 and openai_content[0]["type"] == "text":
+            return {"role": "user", "content": openai_content[0]["text"]}
+
+        return {"role": "user", "content": openai_content or ""}
+
+    def _convert_assistant_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        """转换助手消息"""
+        content = message.get("content")
+        text_parts: List[str] = []
+        tool_calls: List[Dict[str, Any]] = []
+
+        if isinstance(content, str):
+            text_parts.append(content)
+        else:
+            for idx, block in enumerate(content or []):
+                block_type = block.get("type")
+
+                if block_type == self.CONTENT_TYPE_TEXT:
+                    text_parts.append(block.get("text", ""))
+                elif block_type == self.CONTENT_TYPE_TOOL_USE:
+                    tool_calls.append(
+                        {
+                            "id": block.get("id", f"call_{idx}"),
+                            "type": "function",
+                            "function": {
+                                "name": block.get("name", ""),
+                                "arguments": json.dumps(block.get("input", {}), ensure_ascii=False),
+                            },
+                        }
+                    )
+
+        result: Dict[str, Any] = {"role": "assistant"}
+
+        message_content = "\n".join([p for p in text_parts if p]) or None
+        if message_content:
+            result["content"] = message_content
+
+        if tool_calls:
+            result["tool_calls"] = tool_calls
+
+        return result
+
+    def _convert_tools(
+        self, tools: Optional[List[Dict[str, Any]]]
+    ) -> Optional[List[Dict[str, Any]]]:
+        """转换工具定义"""
+        if not tools:
+            return None
+
+        result: List[Dict[str, Any]] = []
+        for tool in tools:
+            result.append(
+                {
+                    "type": "function",
+                    "function": {
+                        "name": tool.get("name", ""),
+                        "description": tool.get("description"),
+                        "parameters": tool.get("input_schema", {}),
+                    },
+                }
+            )
+        return result
+
+    def _convert_tool_choice(
+        self, tool_choice: Optional[Dict[str, Any]]
+    ) -> Optional[Union[str, Dict[str, Any]]]:
+        """转换工具选择"""
+        if tool_choice is None:
+            return None
+
+        choice_type = tool_choice.get("type")
+        if choice_type in ("tool", "tool_use"):
+            return {"type": "function", "function": {"name": tool_choice.get("name", "")}}
+        if choice_type == "any":
+            return "required"
+        if choice_type == "auto":
+            return "auto"
+
+        return tool_choice
+
+    # ==================== 响应转换 ====================
+
+    def convert_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 Claude 响应转换为 OpenAI 格式
+
+        Args:
+            response: Claude 响应字典
+
+        Returns:
+            OpenAI 格式的响应字典
+        """
+        # 提取内容
+        content_parts: List[str] = []
+        tool_calls: List[Dict[str, Any]] = []
+
+        for idx, block in enumerate(response.get("content", [])):
+            block_type = block.get("type")
+
+            if block_type == self.CONTENT_TYPE_TEXT:
+                content_parts.append(block.get("text", ""))
+            elif block_type == self.CONTENT_TYPE_TOOL_USE:
+                tool_calls.append(
+                    {
+                        "id": block.get("id", f"call_{idx}"),
+                        "type": "function",
+                        "function": {
+                            "name": block.get("name", ""),
+                            "arguments": json.dumps(block.get("input", {}), ensure_ascii=False),
+                        },
+                    }
+                )
+
+        # 构建消息
+        message: Dict[str, Any] = {"role": "assistant"}
+        text_content = "\n".join([p for p in content_parts if p]) or None
+        if text_content:
+            message["content"] = text_content
+        if tool_calls:
+            message["tool_calls"] = tool_calls
+
+        # 转换停止原因
+        stop_reason = response.get("stop_reason")
+        finish_reason = self.STOP_REASON_MAP.get(stop_reason, stop_reason) if stop_reason else None
+
+        # 转换 usage
+        usage = response.get("usage", {})
+        openai_usage = {
+            "prompt_tokens": usage.get("input_tokens", 0),
+            "completion_tokens": usage.get("output_tokens", 0),
+            "total_tokens": (usage.get("input_tokens", 0) + usage.get("output_tokens", 0)),
+        }
+
+        return {
+            "id": f"chatcmpl-{response.get('id', uuid.uuid4().hex[:8])}",
+            "object": "chat.completion",
+            "created": int(time.time()),
+            "model": response.get("model", ""),
+            "choices": [
+                {
+                    "index": 0,
+                    "message": message,
+                    "finish_reason": finish_reason,
+                }
+            ],
+            "usage": openai_usage,
+        }
+
+    # ==================== 流式转换 ====================
+
+    def convert_stream_chunk(
+        self,
+        chunk: Dict[str, Any],
+        state: Optional["StreamConversionState"] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        将 Claude SSE 事件转换为 OpenAI 格式
+
+        Args:
+            chunk: Claude SSE 事件
+            state: 流式转换状态
+
+        Returns:
+            OpenAI 格式的 SSE chunk 列表
+        """
+        from src.core.api_format.conversion.state import StreamConversionState
+
+        if state is None:
+            state = StreamConversionState()
+
+        result = self._convert_single_event(chunk, state.model, state.message_id)
+        if result is None:
+            return []
+        return [result]
+
+    def _convert_single_event(
+        self,
+        event: Dict[str, Any],
+        model: str = "",
+        message_id: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        转换单个 Claude SSE 事件为 OpenAI 格式
+
+        Args:
+            event: Claude SSE 事件
+            model: 模型名称
+            message_id: 消息 ID
+
+        Returns:
+            OpenAI 格式的 SSE chunk，如果无法转换返回 None
+        """
+        event_type = event.get("type")
+        chunk_id = f"chatcmpl-{(message_id or 'stream')[-8:]}"
+
+        if event_type == "message_start":
+            message = event.get("message", {})
+            return self._base_chunk(
+                chunk_id,
+                model or message.get("model", ""),
+                {"role": "assistant"},
+            )
+
+        if event_type == "content_block_start":
+            content_block = event.get("content_block", {})
+            if content_block.get("type") == self.CONTENT_TYPE_TOOL_USE:
+                delta = {
+                    "tool_calls": [
+                        {
+                            "index": event.get("index", 0),
+                            "id": content_block.get("id", ""),
+                            "type": "function",
+                            "function": {
+                                "name": content_block.get("name", ""),
+                                "arguments": "",
+                            },
+                        }
+                    ]
+                }
+                return self._base_chunk(chunk_id, model, delta)
+            return None
+
+        if event_type == "content_block_delta":
+            delta_payload = event.get("delta", {})
+            delta_type = delta_payload.get("type")
+
+            if delta_type == "text_delta":
+                delta = {"content": delta_payload.get("text", "")}
+                return self._base_chunk(chunk_id, model, delta)
+
+            if delta_type == "input_json_delta":
+                delta = {
+                    "tool_calls": [
+                        {
+                            "index": event.get("index", 0),
+                            "function": {"arguments": delta_payload.get("partial_json", "")},
+                        }
+                    ]
+                }
+                return self._base_chunk(chunk_id, model, delta)
+            return None
+
+        if event_type == "message_delta":
+            delta = event.get("delta", {})
+            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)
+
+        if event_type == "message_stop":
+            return self._base_chunk(chunk_id, model, {}, finish_reason="stop")
+
+        return None
+
+    def _base_chunk(
+        self,
+        chunk_id: str,
+        model: str,
+        delta: Dict[str, Any],
+        finish_reason: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """构建基础 OpenAI chunk"""
+        return {
+            "id": chunk_id,
+            "object": "chat.completion.chunk",
+            "created": int(time.time()),
+            "model": model,
+            "system_fingerprint": None,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": delta,
+                    "finish_reason": finish_reason,
+                }
+            ],
+        }
+
+    # ==================== 工具方法 ====================
+
+    def _extract_text_content(
+        self, content: Optional[Union[str, List[Dict[str, Any]]]]
+    ) -> Optional[str]:
+        """提取文本内容"""
+        if isinstance(content, str):
+            return content
+        if isinstance(content, list):
+            parts = [
+                block.get("text", "")
+                for block in content
+                if block.get("type") == self.CONTENT_TYPE_TEXT
+            ]
+            return "\n\n".join(filter(None, parts)) or None
+        return None
+
+    def _render_tool_content(self, tool_content: Any) -> str:
+        """渲染工具内容"""
+        if isinstance(tool_content, list):
+            return json.dumps(tool_content, ensure_ascii=False)
+        return str(tool_content)
+
+
+__all__ = ["ClaudeToOpenAIConverter"]

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

@@ -0,0 +1,856 @@
+"""
+Gemini 格式转换器
+
+提供 Gemini 与其他 API 格式（Claude、OpenAI）之间的转换
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+
+if TYPE_CHECKING:
+    from src.core.api_format.conversion.state import GeminiStreamConversionState
+
+
+class ClaudeToGeminiConverter:
+    """
+    Claude -> Gemini 请求转换器
+
+    将 Claude Messages API 格式转换为 Gemini generateContent 格式
+    """
+
+    def convert_request(self, claude_request: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 Claude 请求转换为 Gemini 请求
+
+        Args:
+            claude_request: Claude 格式的请求字典
+
+        Returns:
+            Gemini 格式的请求字典
+        """
+        gemini_request: Dict[str, Any] = {
+            "contents": self._convert_messages(claude_request.get("messages", [])),
+        }
+
+        # 转换 system prompt
+        system = claude_request.get("system")
+        if system:
+            gemini_request["system_instruction"] = self._convert_system(system)
+
+        # 转换生成配置
+        generation_config = self._build_generation_config(claude_request)
+        if generation_config:
+            gemini_request["generation_config"] = generation_config
+
+        # 转换工具
+        tools = claude_request.get("tools")
+        if tools:
+            gemini_request["tools"] = self._convert_tools(tools)
+
+        return gemini_request
+
+    def _convert_messages(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """转换消息列表"""
+        contents = []
+        for msg in messages:
+            role = msg.get("role", "user")
+            # Gemini 使用 "model" 而不是 "assistant"
+            gemini_role = "model" if role == "assistant" else "user"
+
+            content = msg.get("content", "")
+            parts = self._convert_content_to_parts(content)
+
+            contents.append(
+                {
+                    "role": gemini_role,
+                    "parts": parts,
+                }
+            )
+        return contents
+
+    def _convert_content_to_parts(self, content: Any) -> List[Dict[str, Any]]:
+        """将 Claude 内容转换为 Gemini parts"""
+        if isinstance(content, str):
+            return [{"text": content}]
+
+        if isinstance(content, list):
+            parts: List[Dict[str, Any]] = []
+            for block in content:
+                if isinstance(block, str):
+                    parts.append({"text": block})
+                elif isinstance(block, dict):
+                    block_type = block.get("type")
+                    if block_type == "text":
+                        parts.append({"text": block.get("text", "")})
+                    elif block_type == "image":
+                        # 转换图片
+                        source = block.get("source", {})
+                        if source.get("type") == "base64":
+                            parts.append(
+                                {
+                                    "inline_data": {
+                                        "mime_type": source.get("media_type", "image/png"),
+                                        "data": source.get("data", ""),
+                                    }
+                                }
+                            )
+                    elif block_type == "tool_use":
+                        # 转换工具调用
+                        parts.append(
+                            {
+                                "function_call": {
+                                    "name": block.get("name", ""),
+                                    "args": block.get("input", {}),
+                                }
+                            }
+                        )
+                    elif block_type == "tool_result":
+                        # 转换工具结果
+                        parts.append(
+                            {
+                                "function_response": {
+                                    "name": block.get("tool_use_id", ""),
+                                    "response": {"result": block.get("content", "")},
+                                }
+                            }
+                        )
+            return parts
+
+        return [{"text": str(content)}]
+
+    def _convert_system(self, system: Any) -> Dict[str, Any]:
+        """转换 system prompt"""
+        if isinstance(system, str):
+            return {"parts": [{"text": system}]}
+
+        if isinstance(system, list):
+            parts = []
+            for item in system:
+                if isinstance(item, str):
+                    parts.append({"text": item})
+                elif isinstance(item, dict) and item.get("type") == "text":
+                    parts.append({"text": item.get("text", "")})
+            return {"parts": parts}
+
+        return {"parts": [{"text": str(system)}]}
+
+    def _build_generation_config(self, claude_request: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        """构建生成配置"""
+        config: Dict[str, Any] = {}
+
+        if "max_tokens" in claude_request:
+            config["max_output_tokens"] = claude_request["max_tokens"]
+        if "temperature" in claude_request:
+            config["temperature"] = claude_request["temperature"]
+        if "top_p" in claude_request:
+            config["top_p"] = claude_request["top_p"]
+        if "top_k" in claude_request:
+            config["top_k"] = claude_request["top_k"]
+        if "stop_sequences" in claude_request:
+            config["stop_sequences"] = claude_request["stop_sequences"]
+
+        return config if config else None
+
+    def _convert_tools(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """转换工具定义"""
+        function_declarations = []
+        for tool in tools:
+            func_decl = {
+                "name": tool.get("name", ""),
+            }
+            if "description" in tool:
+                func_decl["description"] = tool["description"]
+            if "input_schema" in tool:
+                func_decl["parameters"] = tool["input_schema"]
+            function_declarations.append(func_decl)
+
+        return [{"function_declarations": function_declarations}]
+
+
+class GeminiToClaudeConverter:
+    """
+    Gemini -> Claude 响应转换器
+
+    将 Gemini generateContent 响应转换为 Claude Messages API 格式
+    """
+
+    def convert_response(self, gemini_response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 Gemini 响应转换为 Claude 响应
+
+        Args:
+            gemini_response: Gemini 格式的响应字典
+
+        Returns:
+            Claude 格式的响应字典
+        """
+        candidates = gemini_response.get("candidates", [])
+        if not candidates:
+            return self._create_empty_response()
+
+        candidate = candidates[0]
+        content = candidate.get("content", {})
+        parts = content.get("parts", [])
+
+        # 转换内容块
+        claude_content = self._convert_parts_to_content(parts)
+
+        # 转换使用量
+        usage = self._convert_usage(gemini_response.get("usageMetadata", {}))
+
+        # 转换停止原因
+        stop_reason = self._convert_finish_reason(candidate.get("finishReason"))
+
+        return {
+            "id": f"msg_{gemini_response.get('modelVersion', 'gemini')}",
+            "type": "message",
+            "role": "assistant",
+            "content": claude_content,
+            "model": gemini_response.get("modelVersion", "gemini"),
+            "stop_reason": stop_reason,
+            "stop_sequence": None,
+            "usage": usage,
+        }
+
+    def _convert_parts_to_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 "functionCall" in part:
+                func_call = part["functionCall"]
+                content.append(
+                    {
+                        "type": "tool_use",
+                        "id": f"toolu_{func_call.get('name', '')}",
+                        "name": func_call.get("name", ""),
+                        "input": func_call.get("args", {}),
+                    }
+                )
+        return content
+
+    def _convert_usage(self, usage_metadata: Dict[str, Any]) -> Dict[str, int]:
+        """转换使用量信息"""
+        return {
+            "input_tokens": usage_metadata.get("promptTokenCount", 0),
+            "output_tokens": usage_metadata.get("candidatesTokenCount", 0),
+            "cache_creation_input_tokens": 0,
+            "cache_read_input_tokens": usage_metadata.get("cachedContentTokenCount", 0),
+        }
+
+    def _convert_finish_reason(self, finish_reason: Optional[str]) -> Optional[str]:
+        """转换停止原因"""
+        mapping = {
+            "STOP": "end_turn",
+            "MAX_TOKENS": "max_tokens",
+            "SAFETY": "content_filtered",
+            "RECITATION": "content_filtered",
+            "OTHER": "stop_sequence",
+        }
+        if finish_reason is None:
+            return "end_turn"
+        return mapping.get(finish_reason, "end_turn")
+
+    def _create_empty_response(self) -> Dict[str, Any]:
+        """创建空响应"""
+        return {
+            "id": "msg_empty",
+            "type": "message",
+            "role": "assistant",
+            "content": [],
+            "model": "gemini",
+            "stop_reason": "end_turn",
+            "stop_sequence": None,
+            "usage": {
+                "input_tokens": 0,
+                "output_tokens": 0,
+            },
+        }
+
+    # ==================== 流式转换 ====================
+
+    def convert_stream_chunk(
+        self,
+        chunk: Dict[str, Any],
+        state: Optional["GeminiStreamConversionState"] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        将 Gemini 流式响应转换为 Claude SSE 事件
+
+        Gemini 流式格式与 Claude/OpenAI 不同：
+        - Gemini 返回完整累积文本，需要计算增量
+        - 需要生成 message_start/content_block_delta/message_delta 事件序列
+
+        Args:
+            chunk: Gemini 流式响应块
+            state: 流式转换状态（跨 chunk 追踪）
+
+        Returns:
+            Claude SSE 事件列表
+        """
+        from src.core.api_format.conversion.state import GeminiStreamConversionState
+
+        if state is None:
+            state = GeminiStreamConversionState()
+
+        events: List[Dict[str, Any]] = []
+        candidates = chunk.get("candidates", [])
+        if not candidates:
+            return events
+
+        candidate = candidates[0]
+        content = candidate.get("content", {})
+        parts = content.get("parts", [])
+
+        # 发送 message_start（首次）
+        if not state.message_started:
+            events.append(
+                {
+                    "type": "message_start",
+                    "message": {
+                        "id": state.message_id or "msg_gemini",
+                        "type": "message",
+                        "role": "assistant",
+                        "model": state.model,
+                        "content": [],
+                        "stop_reason": None,
+                        "stop_sequence": None,
+                    },
+                }
+            )
+            state.message_started = True
+
+        # 处理文本增量
+        for part in parts:
+            if "text" in part:
+                full_text = part["text"]
+                # Gemini 返回累积文本，计算增量
+                delta_text = full_text[len(state.accumulated_text) :]
+                if delta_text:
+                    state.accumulated_text = full_text
+                    # 首次文本需要发送 content_block_start
+                    if not state.content_block_started:
+                        events.append(
+                            {
+                                "type": "content_block_start",
+                                "index": state.current_block_index,
+                                "content_block": {"type": "text", "text": ""},
+                            }
+                        )
+                        state.content_block_started = True
+                    events.append(
+                        {
+                            "type": "content_block_delta",
+                            "index": state.current_block_index,
+                            "delta": {"type": "text_delta", "text": delta_text},
+                        }
+                    )
+            elif "functionCall" in part:
+                # 工具调用
+                func_call = part["functionCall"]
+                tool_id = f"toolu_{func_call.get('name', '')}_{state.tool_call_index}"
+                state.tool_call_index += 1
+                state.current_block_index += 1
+
+                events.append(
+                    {
+                        "type": "content_block_start",
+                        "index": state.current_block_index,
+                        "content_block": {
+                            "type": "tool_use",
+                            "id": tool_id,
+                            "name": func_call.get("name", ""),
+                            "input": {},
+                        },
+                    }
+                )
+                # 发送完整的 input 作为 JSON delta
+                args = func_call.get("args", {})
+                if args:
+                    events.append(
+                        {
+                            "type": "content_block_delta",
+                            "index": state.current_block_index,
+                            "delta": {
+                                "type": "input_json_delta",
+                                "partial_json": json.dumps(args, ensure_ascii=False),
+                            },
+                        }
+                    )
+                events.append(
+                    {
+                        "type": "content_block_stop",
+                        "index": state.current_block_index,
+                    }
+                )
+
+        # 处理结束
+        finish_reason = candidate.get("finishReason")
+        if finish_reason:
+            # 关闭当前内容块
+            if state.content_block_started:
+                events.append(
+                    {
+                        "type": "content_block_stop",
+                        "index": state.current_block_index,
+                    }
+                )
+
+            stop_reason = self._convert_finish_reason(finish_reason)
+            events.append(
+                {
+                    "type": "message_delta",
+                    "delta": {"stop_reason": stop_reason},
+                    "usage": {"output_tokens": 0},
+                }
+            )
+
+        return events
+
+
+class OpenAIToGeminiConverter:
+    """
+    OpenAI -> Gemini 请求转换器
+
+    将 OpenAI Chat Completions API 格式转换为 Gemini generateContent 格式
+    """
+
+    def convert_request(self, openai_request: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 OpenAI 请求转换为 Gemini 请求
+
+        Args:
+            openai_request: OpenAI 格式的请求字典
+
+        Returns:
+            Gemini 格式的请求字典
+        """
+        messages = openai_request.get("messages", [])
+
+        # 分离 system 消息和其他消息
+        system_messages = []
+        other_messages = []
+        for msg in messages:
+            if msg.get("role") == "system":
+                system_messages.append(msg)
+            else:
+                other_messages.append(msg)
+
+        gemini_request: Dict[str, Any] = {
+            "contents": self._convert_messages(other_messages),
+        }
+
+        # 转换 system messages
+        if system_messages:
+            system_text = "\n".join(msg.get("content", "") for msg in system_messages)
+            gemini_request["system_instruction"] = {"parts": [{"text": system_text}]}
+
+        # 转换生成配置
+        generation_config = self._build_generation_config(openai_request)
+        if generation_config:
+            gemini_request["generation_config"] = generation_config
+
+        # 转换工具
+        tools = openai_request.get("tools")
+        if tools:
+            gemini_request["tools"] = self._convert_tools(tools)
+
+        return gemini_request
+
+    def _convert_messages(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """转换消息列表"""
+        contents = []
+        for msg in messages:
+            role = msg.get("role", "user")
+            gemini_role = "model" if role == "assistant" else "user"
+
+            content = msg.get("content", "")
+            parts = self._convert_content_to_parts(content)
+
+            # 处理工具调用
+            tool_calls = msg.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(
+                        {
+                            "function_call": {
+                                "name": func.get("name", ""),
+                                "args": args,
+                            }
+                        }
+                    )
+
+            if parts:
+                contents.append(
+                    {
+                        "role": gemini_role,
+                        "parts": parts,
+                    }
+                )
+        return contents
+
+    def _convert_content_to_parts(self, content: Any) -> List[Dict[str, Any]]:
+        """将 OpenAI 内容转换为 Gemini parts"""
+        if content is None:
+            return []
+
+        if isinstance(content, str):
+            return [{"text": content}]
+
+        if isinstance(content, list):
+            parts: List[Dict[str, Any]] = []
+            for item in content:
+                if isinstance(item, str):
+                    parts.append({"text": item})
+                elif isinstance(item, dict):
+                    item_type = item.get("type")
+                    if item_type == "text":
+                        parts.append({"text": item.get("text", "")})
+                    elif item_type == "image_url":
+                        # OpenAI 图片 URL 格式
+                        image_url = item.get("image_url", {})
+                        url = image_url.get("url", "")
+                        if url.startswith("data:"):
+                            # base64 数据 URL
+                            # 格式: data:image/png;base64,xxxxx
+                            try:
+                                header, data = url.split(",", 1)
+                                mime_type = header.split(":")[1].split(";")[0]
+                                parts.append(
+                                    {
+                                        "inline_data": {
+                                            "mime_type": mime_type,
+                                            "data": data,
+                                        }
+                                    }
+                                )
+                            except (ValueError, IndexError):
+                                pass
+            return parts
+
+        return [{"text": str(content)}]
+
+    def _build_generation_config(self, openai_request: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        """构建生成配置"""
+        config: Dict[str, Any] = {}
+
+        if "max_tokens" in openai_request:
+            config["max_output_tokens"] = openai_request["max_tokens"]
+        if "temperature" in openai_request:
+            config["temperature"] = openai_request["temperature"]
+        if "top_p" in openai_request:
+            config["top_p"] = openai_request["top_p"]
+        if "stop" in openai_request:
+            stop = openai_request["stop"]
+            if isinstance(stop, str):
+                config["stop_sequences"] = [stop]
+            elif isinstance(stop, list):
+                config["stop_sequences"] = stop
+        if "n" in openai_request:
+            config["candidate_count"] = openai_request["n"]
+
+        return config if config else None
+
+    def _convert_tools(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """转换工具定义"""
+        function_declarations = []
+        for tool in tools:
+            if tool.get("type") == "function":
+                func = tool.get("function", {})
+                func_decl = {
+                    "name": func.get("name", ""),
+                }
+                if "description" in func:
+                    func_decl["description"] = func["description"]
+                if "parameters" in func:
+                    func_decl["parameters"] = func["parameters"]
+                function_declarations.append(func_decl)
+
+        return [{"function_declarations": function_declarations}]
+
+
+class GeminiToOpenAIConverter:
+    """
+    Gemini -> OpenAI 响应转换器
+
+    将 Gemini generateContent 响应转换为 OpenAI Chat Completions API 格式
+    """
+
+    def convert_response(self, gemini_response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 Gemini 响应转换为 OpenAI 响应
+
+        Args:
+            gemini_response: Gemini 格式的响应字典
+
+        Returns:
+            OpenAI 格式的响应字典
+        """
+        candidates = gemini_response.get("candidates", [])
+        choices = []
+
+        for i, candidate in enumerate(candidates):
+            content = candidate.get("content", {})
+            parts = content.get("parts", [])
+
+            # 提取文本内容
+            text_parts = []
+            tool_calls = []
+
+            for part in parts:
+                if "text" in part:
+                    text_parts.append(part["text"])
+                elif "functionCall" in part:
+                    func_call = part["functionCall"]
+                    tool_calls.append(
+                        {
+                            "id": f"call_{func_call.get('name', '')}_{i}",
+                            "type": "function",
+                            "function": {
+                                "name": func_call.get("name", ""),
+                                "arguments": json.dumps(func_call.get("args", {})),
+                            },
+                        }
+                    )
+
+            message: Dict[str, Any] = {
+                "role": "assistant",
+                "content": "".join(text_parts) if text_parts else None,
+            }
+
+            if tool_calls:
+                message["tool_calls"] = tool_calls
+
+            finish_reason = self._convert_finish_reason(candidate.get("finishReason"))
+
+            choices.append(
+                {
+                    "index": i,
+                    "message": message,
+                    "finish_reason": finish_reason,
+                }
+            )
+
+        # 转换使用量
+        usage = self._convert_usage(gemini_response.get("usageMetadata", {}))
+
+        return {
+            "id": f"chatcmpl-{gemini_response.get('modelVersion', 'gemini')}",
+            "object": "chat.completion",
+            "created": int(time.time()),
+            "model": gemini_response.get("modelVersion", "gemini"),
+            "choices": choices,
+            "usage": usage,
+        }
+
+    def _convert_usage(self, usage_metadata: Dict[str, Any]) -> Dict[str, int]:
+        """转换使用量信息"""
+        prompt_tokens = usage_metadata.get("promptTokenCount", 0)
+        completion_tokens = usage_metadata.get("candidatesTokenCount", 0)
+        return {
+            "prompt_tokens": prompt_tokens,
+            "completion_tokens": completion_tokens,
+            "total_tokens": prompt_tokens + completion_tokens,
+        }
+
+    def _convert_finish_reason(self, finish_reason: Optional[str]) -> str:
+        """转换停止原因"""
+        mapping = {
+            "STOP": "stop",
+            "MAX_TOKENS": "length",
+            "SAFETY": "content_filter",
+            "RECITATION": "content_filter",
+            "OTHER": "stop",
+        }
+        if finish_reason is None:
+            return "stop"
+        return mapping.get(finish_reason, "stop")
+
+    # ==================== 流式转换 ====================
+
+    def convert_stream_chunk(
+        self,
+        chunk: Dict[str, Any],
+        state: Optional["GeminiStreamConversionState"] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        将 Gemini 流式响应转换为 OpenAI SSE chunk
+
+        Gemini 流式格式与 OpenAI 不同：
+        - Gemini 返回完整累积文本，需要计算增量
+        - 需要生成 OpenAI chat.completion.chunk 格式
+
+        Args:
+            chunk: Gemini 流式响应块
+            state: 流式转换状态（跨 chunk 追踪）
+
+        Returns:
+            OpenAI SSE chunk 列表
+        """
+        from src.core.api_format.conversion.state import GeminiStreamConversionState
+
+        if state is None:
+            state = GeminiStreamConversionState()
+
+        events: List[Dict[str, Any]] = []
+        candidates = chunk.get("candidates", [])
+        if not candidates:
+            return events
+
+        candidate = candidates[0]
+        content = candidate.get("content", {})
+        parts = content.get("parts", [])
+        finish_reason = candidate.get("finishReason")
+
+        chunk_id = f"chatcmpl-{state.message_id or 'gemini'}"
+
+        # 发送首个 chunk（带 role）
+        if not state.message_started:
+            events.append(
+                {
+                    "id": chunk_id,
+                    "object": "chat.completion.chunk",
+                    "created": int(time.time()),
+                    "model": state.model,
+                    "choices": [
+                        {
+                            "index": 0,
+                            "delta": {"role": "assistant", "content": ""},
+                            "finish_reason": None,
+                        }
+                    ],
+                }
+            )
+            state.message_started = True
+
+        # 处理文本增量
+        for part in parts:
+            if "text" in part:
+                full_text = part["text"]
+                # Gemini 返回累积文本，计算增量
+                delta_text = full_text[len(state.accumulated_text) :]
+                if delta_text:
+                    state.accumulated_text = full_text
+                    events.append(
+                        {
+                            "id": chunk_id,
+                            "object": "chat.completion.chunk",
+                            "created": int(time.time()),
+                            "model": state.model,
+                            "choices": [
+                                {
+                                    "index": 0,
+                                    "delta": {"content": delta_text},
+                                    "finish_reason": None,
+                                }
+                            ],
+                        }
+                    )
+            elif "functionCall" in part:
+                # 工具调用
+                func_call = part["functionCall"]
+                tool_id = f"call_{func_call.get('name', '')}_{state.tool_call_index}"
+
+                # 工具调用开始
+                events.append(
+                    {
+                        "id": chunk_id,
+                        "object": "chat.completion.chunk",
+                        "created": int(time.time()),
+                        "model": state.model,
+                        "choices": [
+                            {
+                                "index": 0,
+                                "delta": {
+                                    "tool_calls": [
+                                        {
+                                            "index": state.tool_call_index,
+                                            "id": tool_id,
+                                            "type": "function",
+                                            "function": {
+                                                "name": func_call.get("name", ""),
+                                                "arguments": "",
+                                            },
+                                        }
+                                    ]
+                                },
+                                "finish_reason": None,
+                            }
+                        ],
+                    }
+                )
+
+                # 工具调用参数
+                args = func_call.get("args", {})
+                if args:
+                    events.append(
+                        {
+                            "id": chunk_id,
+                            "object": "chat.completion.chunk",
+                            "created": int(time.time()),
+                            "model": state.model,
+                            "choices": [
+                                {
+                                    "index": 0,
+                                    "delta": {
+                                        "tool_calls": [
+                                            {
+                                                "index": state.tool_call_index,
+                                                "function": {
+                                                    "arguments": json.dumps(
+                                                        args, ensure_ascii=False
+                                                    )
+                                                },
+                                            }
+                                        ]
+                                    },
+                                    "finish_reason": None,
+                                }
+                            ],
+                        }
+                    )
+
+                state.tool_call_index += 1
+
+        # 处理结束
+        if finish_reason:
+            openai_finish_reason = self._convert_finish_reason(finish_reason)
+            events.append(
+                {
+                    "id": chunk_id,
+                    "object": "chat.completion.chunk",
+                    "created": int(time.time()),
+                    "model": state.model,
+                    "choices": [
+                        {
+                            "index": 0,
+                            "delta": {},
+                            "finish_reason": openai_finish_reason,
+                        }
+                    ],
+                }
+            )
+
+        return events
+
+
+__all__ = [
+    "ClaudeToGeminiConverter",
+    "GeminiToClaudeConverter",
+    "OpenAIToGeminiConverter",
+    "GeminiToOpenAIConverter",
+]

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

@@ -0,0 +1,492 @@
+"""
+OpenAI -> Claude 格式转换器
+
+将 OpenAI Chat Completions API 格式转换为 Claude Messages API 格式。
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+
+if TYPE_CHECKING:
+    from src.core.api_format.conversion.state import StreamConversionState
+
+
+class OpenAIToClaudeConverter:
+    """
+    OpenAI -> Claude 格式转换器
+
+    支持：
+    - 请求转换：OpenAI Chat Request -> Claude Request
+    - 响应转换：OpenAI Chat Response -> Claude Response
+    - 流式转换：OpenAI SSE -> Claude SSE
+    """
+
+    # 内容类型常量
+    CONTENT_TYPE_TEXT = "text"
+    CONTENT_TYPE_IMAGE = "image"
+    CONTENT_TYPE_TOOL_USE = "tool_use"
+    CONTENT_TYPE_TOOL_RESULT = "tool_result"
+
+    # 停止原因映射（OpenAI -> Claude）
+    FINISH_REASON_MAP = {
+        "stop": "end_turn",
+        "length": "max_tokens",
+        "tool_calls": "tool_use",
+        "function_call": "tool_use",
+        "content_filter": "end_turn",
+    }
+
+    def __init__(self, model_mapping: Optional[Dict[str, str]] = None):
+        """
+        Args:
+            model_mapping: OpenAI 模型到 Claude 模型的映射
+        """
+        self._model_mapping = model_mapping or {}
+
+    # ==================== 请求转换 ====================
+
+    def convert_request(self, request: Union[Dict[str, Any], Any]) -> Dict[str, Any]:
+        """
+        将 OpenAI 请求转换为 Claude 格式
+
+        Args:
+            request: OpenAI 请求（Dict 或 Pydantic 模型）
+
+        Returns:
+            Claude 格式的请求字典
+        """
+        if hasattr(request, "model_dump"):
+            data = request.model_dump(exclude_none=True)
+        else:
+            data = dict(request)
+
+        # 模型映射
+        model = data.get("model", "")
+        claude_model = self._model_mapping.get(model, model)
+
+        # 处理消息
+        system_content: Optional[str] = None
+        claude_messages: List[Dict[str, Any]] = []
+
+        for message in data.get("messages", []):
+            role = message.get("role")
+
+            # 提取 system 消息
+            if role == "system":
+                system_content = self._collapse_content(message.get("content"))
+                continue
+
+            # 转换其他消息
+            converted = self._convert_message(message)
+            if converted:
+                claude_messages.append(converted)
+
+        # 构建 Claude 请求
+        result: Dict[str, Any] = {
+            "model": claude_model,
+            "messages": claude_messages,
+            "max_tokens": data.get("max_tokens") or 4096,
+        }
+
+        # 可选参数
+        if data.get("temperature") is not None:
+            result["temperature"] = data["temperature"]
+        if data.get("top_p") is not None:
+            result["top_p"] = data["top_p"]
+        if data.get("stream"):
+            result["stream"] = data["stream"]
+        if data.get("stop"):
+            result["stop_sequences"] = self._convert_stop(data["stop"])
+        if system_content:
+            result["system"] = system_content
+
+        # 工具转换
+        tools = self._convert_tools(data.get("tools"))
+        if tools:
+            result["tools"] = tools
+
+        tool_choice = self._convert_tool_choice(data.get("tool_choice"))
+        if tool_choice:
+            result["tool_choice"] = tool_choice
+
+        return result
+
+    def _convert_message(self, message: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        """转换单条消息"""
+        role = message.get("role")
+
+        if role == "user":
+            return self._convert_user_message(message)
+        if role == "assistant":
+            return self._convert_assistant_message(message)
+        if role == "tool":
+            return self._convert_tool_message(message)
+
+        return None
+
+    def _convert_user_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        """转换用户消息"""
+        content = message.get("content")
+
+        if isinstance(content, str) or content is None:
+            return {"role": "user", "content": content or ""}
+
+        # 转换内容数组
+        claude_content: List[Dict[str, Any]] = []
+        for item in content:
+            item_type = item.get("type")
+
+            if item_type == "text":
+                claude_content.append(
+                    {"type": self.CONTENT_TYPE_TEXT, "text": item.get("text", "")}
+                )
+            elif item_type == "image_url":
+                image_url = (item.get("image_url") or {}).get("url", "")
+                claude_content.append(self._convert_image_url(image_url))
+
+        return {"role": "user", "content": claude_content}
+
+    def _convert_assistant_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        """转换助手消息"""
+        content_blocks: List[Dict[str, Any]] = []
+
+        # 处理文本内容
+        content = message.get("content")
+        if isinstance(content, str):
+            content_blocks.append({"type": self.CONTENT_TYPE_TEXT, "text": content})
+        elif isinstance(content, list):
+            for part in content:
+                if part.get("type") == "text":
+                    content_blocks.append(
+                        {"type": self.CONTENT_TYPE_TEXT, "text": part.get("text", "")}
+                    )
+
+        # 处理工具调用
+        for tool_call in message.get("tool_calls") or []:
+            if tool_call.get("type") == "function":
+                function = tool_call.get("function", {})
+                arguments = function.get("arguments", "{}")
+                try:
+                    input_data = json.loads(arguments)
+                except json.JSONDecodeError:
+                    input_data = {"raw": arguments}
+
+                content_blocks.append(
+                    {
+                        "type": self.CONTENT_TYPE_TOOL_USE,
+                        "id": tool_call.get("id", ""),
+                        "name": function.get("name", ""),
+                        "input": input_data,
+                    }
+                )
+
+        # 简化单文本内容
+        if not content_blocks:
+            return {"role": "assistant", "content": ""}
+        if len(content_blocks) == 1 and content_blocks[0]["type"] == self.CONTENT_TYPE_TEXT:
+            return {"role": "assistant", "content": content_blocks[0]["text"]}
+
+        return {"role": "assistant", "content": content_blocks}
+
+    def _convert_tool_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        """转换工具结果消息"""
+        tool_content = message.get("content", "")
+
+        # 尝试解析 JSON
+        parsed_content = tool_content
+        if isinstance(tool_content, str):
+            try:
+                parsed_content = json.loads(tool_content)
+            except json.JSONDecodeError:
+                pass
+
+        tool_block = {
+            "type": self.CONTENT_TYPE_TOOL_RESULT,
+            "tool_use_id": message.get("tool_call_id", ""),
+            "content": parsed_content,
+        }
+
+        return {"role": "user", "content": [tool_block]}
+
+    def _convert_tools(
+        self, tools: Optional[List[Dict[str, Any]]]
+    ) -> Optional[List[Dict[str, Any]]]:
+        """转换工具定义"""
+        if not tools:
+            return None
+
+        result: List[Dict[str, Any]] = []
+        for tool in tools:
+            if tool.get("type") != "function":
+                continue
+
+            function = tool.get("function", {})
+            result.append(
+                {
+                    "name": function.get("name", ""),
+                    "description": function.get("description"),
+                    "input_schema": function.get("parameters") or {},
+                }
+            )
+
+        return result if result else None
+
+    def _convert_tool_choice(
+        self, tool_choice: Optional[Union[str, Dict[str, Any]]]
+    ) -> Optional[Dict[str, Any]]:
+        """转换工具选择"""
+        if tool_choice is None:
+            return None
+        if tool_choice == "none":
+            return {"type": "none"}
+        if tool_choice == "auto":
+            return {"type": "auto"}
+        if tool_choice == "required":
+            return {"type": "any"}
+        if isinstance(tool_choice, dict) and tool_choice.get("type") == "function":
+            function = tool_choice.get("function", {})
+            return {"type": "tool_use", "name": function.get("name", "")}
+
+        return {"type": "auto"}
+
+    def _convert_image_url(self, image_url: str) -> Dict[str, Any]:
+        """转换图片 URL"""
+        if image_url.startswith("data:"):
+            header, _, data = image_url.partition(",")
+            media_type = "image/jpeg"
+            if ";" in header:
+                media_type = header.split(";")[0].split(":")[-1]
+
+            return {
+                "type": self.CONTENT_TYPE_IMAGE,
+                "source": {
+                    "type": "base64",
+                    "media_type": media_type,
+                    "data": data,
+                },
+            }
+
+        return {"type": self.CONTENT_TYPE_TEXT, "text": f"[Image: {image_url}]"}
+
+    def _convert_stop(self, stop: Optional[Union[str, List[str]]]) -> Optional[List[str]]:
+        """转换停止序列"""
+        if stop is None:
+            return None
+        if isinstance(stop, str):
+            return [stop]
+        return stop
+
+    # ==================== 响应转换 ====================
+
+    def convert_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        将 OpenAI 响应转换为 Claude 格式
+
+        Args:
+            response: OpenAI 响应字典
+
+        Returns:
+            Claude 格式的响应字典
+        """
+        choices = response.get("choices", [])
+        if not choices:
+            return self._empty_claude_response(response)
+
+        choice = choices[0]
+        message = choice.get("message", {})
+
+        # 构建 content 数组
+        content: List[Dict[str, Any]] = []
+
+        # 处理文本
+        text_content = message.get("content")
+        if text_content:
+            content.append(
+                {
+                    "type": self.CONTENT_TYPE_TEXT,
+                    "text": text_content,
+                }
+            )
+
+        # 处理工具调用
+        for tool_call in message.get("tool_calls") or []:
+            if tool_call.get("type") == "function":
+                function = tool_call.get("function", {})
+                arguments = function.get("arguments", "{}")
+                try:
+                    input_data = json.loads(arguments)
+                except json.JSONDecodeError:
+                    input_data = {"raw": arguments}
+
+                content.append(
+                    {
+                        "type": self.CONTENT_TYPE_TOOL_USE,
+                        "id": tool_call.get("id", ""),
+                        "name": function.get("name", ""),
+                        "input": input_data,
+                    }
+                )
+
+        # 转换 finish_reason
+        finish_reason = choice.get("finish_reason")
+        stop_reason = self.FINISH_REASON_MAP.get(finish_reason, "end_turn")
+
+        # 转换 usage
+        usage = response.get("usage", {})
+        claude_usage = {
+            "input_tokens": usage.get("prompt_tokens", 0),
+            "output_tokens": usage.get("completion_tokens", 0),
+        }
+
+        return {
+            "id": f"msg_{response.get('id', uuid.uuid4().hex[:8])}",
+            "type": "message",
+            "role": "assistant",
+            "model": response.get("model", ""),
+            "content": content,
+            "stop_reason": stop_reason,
+            "stop_sequence": None,
+            "usage": claude_usage,
+        }
+
+    def _empty_claude_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
+        """构建空的 Claude 响应"""
+        return {
+            "id": f"msg_{response.get('id', uuid.uuid4().hex[:8])}",
+            "type": "message",
+            "role": "assistant",
+            "model": response.get("model", ""),
+            "content": [],
+            "stop_reason": "end_turn",
+            "stop_sequence": None,
+            "usage": {"input_tokens": 0, "output_tokens": 0},
+        }
+
+    # ==================== 流式转换 ====================
+
+    def convert_stream_chunk(
+        self,
+        chunk: Dict[str, Any],
+        state: Optional["StreamConversionState"] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        将 OpenAI SSE chunk 转换为 Claude SSE 事件
+
+        Args:
+            chunk: OpenAI SSE chunk
+            state: 流式转换状态
+
+        Returns:
+            Claude SSE 事件列表
+        """
+        from src.core.api_format.conversion.state import StreamConversionState
+
+        if state is None:
+            state = StreamConversionState()
+
+        events: List[Dict[str, Any]] = []
+
+        choices = chunk.get("choices", [])
+        if not choices:
+            return events
+
+        choice = choices[0]
+        delta = choice.get("delta", {})
+        finish_reason = choice.get("finish_reason")
+
+        # 处理角色（第一个 chunk）
+        role = delta.get("role")
+        if role and not state.message_started:
+            msg_id = state.message_id or f"msg_{uuid.uuid4().hex[:8]}"
+            events.append(
+                {
+                    "type": "message_start",
+                    "message": {
+                        "id": msg_id,
+                        "type": "message",
+                        "role": role,
+                        "model": state.model,
+                        "content": [],
+                        "stop_reason": None,
+                        "stop_sequence": None,
+                    },
+                }
+            )
+            state.message_started = True
+
+        # 处理文本内容
+        content_delta = delta.get("content")
+        if isinstance(content_delta, str):
+            events.append(
+                {
+                    "type": "content_block_delta",
+                    "index": 0,
+                    "delta": {"type": "text_delta", "text": content_delta},
+                }
+            )
+
+        # 处理工具调用
+        tool_calls = delta.get("tool_calls", [])
+        for tool_call in tool_calls:
+            index = tool_call.get("index", 0)
+
+            # 工具调用开始
+            if "id" in tool_call:
+                function = tool_call.get("function", {})
+                events.append(
+                    {
+                        "type": "content_block_start",
+                        "index": index,
+                        "content_block": {
+                            "type": self.CONTENT_TYPE_TOOL_USE,
+                            "id": tool_call["id"],
+                            "name": function.get("name", ""),
+                        },
+                    }
+                )
+
+            # 工具调用参数增量
+            function = tool_call.get("function", {})
+            if "arguments" in function:
+                events.append(
+                    {
+                        "type": "content_block_delta",
+                        "index": index,
+                        "delta": {
+                            "type": "input_json_delta",
+                            "partial_json": function.get("arguments", ""),
+                        },
+                    }
+                )
+
+        # 处理结束
+        if finish_reason:
+            stop_reason = self.FINISH_REASON_MAP.get(finish_reason, "end_turn")
+            events.append(
+                {
+                    "type": "message_delta",
+                    "delta": {"stop_reason": stop_reason},
+                }
+            )
+
+        return events
+
+    # ==================== 工具方法 ====================
+
+    def _collapse_content(
+        self, content: Optional[Union[str, List[Dict[str, Any]]]]
+    ) -> Optional[str]:
+        """折叠内容为字符串"""
+        if isinstance(content, str):
+            return content
+        if not content:
+            return None
+
+        text_parts = [part.get("text", "") for part in content if part.get("type") == "text"]
+        return "\n\n".join(filter(None, text_parts)) or None
+
+
+__all__ = ["OpenAIToClaudeConverter"]

src/core/api_format/conversion/exceptions.py

@@ -0,0 +1,27 @@
+"""
+格式转换异常
+
+用于严格模式转换失败时抛出，让编排器可以尝试下一个候选。
+"""
+
+from __future__ import annotations
+
+
+class FormatConversionError(Exception):
+    """
+    格式转换失败异常
+
+    在严格模式下，转换失败会抛出此异常，
+    让 Orchestrator 可以捕获并尝试下一个候选。
+    """
+
+    def __init__(self, source_format: str, target_format: str, message: str) -> None:
+        self.source_format = source_format
+        self.target_format = target_format
+        self.message = message
+        super().__init__(f"格式转换失败 ({source_format} -> {target_format}): {message}")
+
+
+__all__ = [
+    "FormatConversionError",
+]

src/core/api_format/conversion/protocols.py

@@ -0,0 +1,54 @@
+"""
+转换器协议定义
+
+定义转换器必须实现的方法签名，用于类型检查和文档说明。
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class RequestConverter(Protocol):
+    """请求转换器协议"""
+
+    def convert_request(self, request: Dict[str, Any]) -> Dict[str, Any]: ...
+
+
+@runtime_checkable
+class ResponseConverter(Protocol):
+    """响应转换器协议"""
+
+    def convert_response(self, response: Dict[str, Any]) -> Dict[str, Any]: ...
+
+
+@runtime_checkable
+class StreamChunkConverter(Protocol):
+    """
+    流式响应块转换器协议
+
+    统一签名：(chunk, state) -> List[Dict]
+
+    说明：
+    - chunk: 单个流式事件/块
+    - state: 跨 chunk 的转换状态（StreamConversionState 或 GeminiStreamConversionState）
+    - 返回: 转换后的事件列表（可能 0-N 个）
+
+    注意：
+    - 所有流式转换器必须实现此签名
+    - state 参数用于跨 chunk 维护状态（如累积文本、消息 ID 等）
+    """
+
+    def convert_stream_chunk(
+        self,
+        chunk: Dict[str, Any],
+        state: Any,
+    ) -> List[Dict[str, Any]]: ...
+
+
+__all__ = [
+    "RequestConverter",
+    "ResponseConverter",
+    "StreamChunkConverter",
+]

src/core/api_format/conversion/registry.py

@@ -0,0 +1,390 @@
+"""
+格式转换器注册表（核心层）
+
+自动管理不同 API 格式之间的转换器，支持：
+- 请求转换：客户端格式 → Provider 格式
+- 响应转换：Provider 格式 → 客户端格式
+
+说明：
+- 该注册表位于 core 层，避免 services 依赖 api/handlers。
+- 具体转换器的注册（例如 Claude/OpenAI/Gemini）应由应用启动层完成，
+  或由 api 层的 bootstrap 逻辑完成，以保持依赖方向：api -> core。
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+
+from src.core.logger import logger
+
+from .exceptions import FormatConversionError
+
+if TYPE_CHECKING:
+    from .state import GeminiStreamConversionState, StreamConversionState
+
+
+class FormatConverterRegistry:
+    """
+    格式转换器注册表
+
+    管理不同 API 格式之间的双向转换器
+    """
+
+    def __init__(self) -> None:
+        # key: (source_format, target_format), value: converter instance
+        self._converters: Dict[Tuple[str, str], Any] = {}
+
+    def register(
+        self,
+        source_format: str,
+        target_format: str,
+        converter: Any,
+    ) -> None:
+        """
+        注册格式转换器
+
+        Args:
+            source_format: 源格式（如 "CLAUDE", "OPENAI", "GEMINI"）
+            target_format: 目标格式
+            converter: 转换器实例（需要有 convert_request/convert_response 方法）
+        """
+        key = (source_format.upper(), target_format.upper())
+        self._converters[key] = converter
+        logger.info(f"[ConverterRegistry] 注册转换器: {source_format} -> {target_format}")
+
+    def get_converter(
+        self,
+        source_format: str,
+        target_format: str,
+    ) -> Optional[Any]:
+        """
+        获取转换器
+
+        Args:
+            source_format: 源格式
+            target_format: 目标格式
+
+        Returns:
+            转换器实例，如果不存在返回 None
+        """
+        key = (source_format.upper(), target_format.upper())
+        return self._converters.get(key)
+
+    def has_converter(
+        self,
+        source_format: str,
+        target_format: str,
+    ) -> bool:
+        """检查是否存在转换器"""
+        key = (source_format.upper(), target_format.upper())
+        return key in self._converters
+
+    def convert_request(
+        self,
+        request: Dict[str, Any],
+        source_format: str,
+        target_format: str,
+    ) -> Dict[str, Any]:
+        """
+        转换请求
+
+        Args:
+            request: 原始请求字典
+            source_format: 源格式（客户端格式）
+            target_format: 目标格式（Provider 格式）
+
+        Returns:
+            转换后的请求字典，如果无需转换或没有转换器则返回原始请求
+        """
+        # 同格式无需转换
+        if source_format.upper() == target_format.upper():
+            return request
+
+        converter = self.get_converter(source_format, target_format)
+        if converter is None:
+            logger.warning(
+                f"[ConverterRegistry] 未找到请求转换器: {source_format} -> {target_format}，返回原始请求"
+            )
+            return request
+
+        if not hasattr(converter, "convert_request"):
+            logger.warning(
+                f"[ConverterRegistry] 转换器缺少 convert_request 方法: {source_format} -> {target_format}"
+            )
+            return request
+
+        try:
+            converted: Dict[str, Any] = converter.convert_request(request)
+            logger.debug(f"[ConverterRegistry] 请求转换成功: {source_format} -> {target_format}")
+            return converted
+        except Exception as e:
+            logger.error(f"[ConverterRegistry] 请求转换失败: {source_format} -> {target_format}: {e}")
+            return request
+
+    def convert_response(
+        self,
+        response: Dict[str, Any],
+        source_format: str,
+        target_format: str,
+    ) -> Dict[str, Any]:
+        """
+        转换响应
+
+        Args:
+            response: 原始响应字典
+            source_format: 源格式（Provider 格式）
+            target_format: 目标格式（客户端格式）
+
+        Returns:
+            转换后的响应字典，如果无需转换或没有转换器则返回原始响应
+        """
+        # 同格式无需转换
+        if source_format.upper() == target_format.upper():
+            return response
+
+        converter = self.get_converter(source_format, target_format)
+        if converter is None:
+            logger.warning(
+                f"[ConverterRegistry] 未找到响应转换器: {source_format} -> {target_format}，返回原始响应"
+            )
+            return response
+
+        if not hasattr(converter, "convert_response"):
+            logger.warning(
+                f"[ConverterRegistry] 转换器缺少 convert_response 方法: {source_format} -> {target_format}"
+            )
+            return response
+
+        try:
+            converted: Dict[str, Any] = converter.convert_response(response)
+            logger.debug(f"[ConverterRegistry] 响应转换成功: {source_format} -> {target_format}")
+            return converted
+        except Exception as e:
+            logger.error(f"[ConverterRegistry] 响应转换失败: {source_format} -> {target_format}: {e}")
+            return response
+
+    def convert_stream_chunk(
+        self,
+        chunk: Dict[str, Any],
+        source_format: str,
+        target_format: str,
+        state: Optional[Union["StreamConversionState", "GeminiStreamConversionState"]] = None,
+    ) -> list[Dict[str, Any]]:
+        """
+        转换流式响应块
+
+        Args:
+            chunk: 原始流式响应块
+            source_format: 源格式（Provider 格式）
+            target_format: 目标格式（客户端格式）
+            state: 流式转换状态（StreamConversionState 或 GeminiStreamConversionState）
+
+        Returns:
+            转换后的事件列表（可能 0-N 个），失败时返回原始 chunk 的单元素列表
+        """
+        # 同格式无需转换
+        if source_format.upper() == target_format.upper():
+            return [chunk]
+
+        converter = self.get_converter(source_format, target_format)
+        if converter is None:
+            return [chunk]
+
+        # 使用流式转换方法
+        if hasattr(converter, "convert_stream_chunk"):
+            try:
+                result: list[Dict[str, Any]] = converter.convert_stream_chunk(chunk, state)
+                return result
+            except Exception as e:
+                logger.error(f"[ConverterRegistry] 流式块转换失败: {source_format} -> {target_format}: {e}")
+                return [chunk]
+
+        # 降级到普通响应转换（作为单个事件返回）
+        if hasattr(converter, "convert_response"):
+            try:
+                result = converter.convert_response(chunk)
+                return [result]
+            except Exception:
+                return [chunk]
+
+        return [chunk]
+
+    def list_converters(self) -> list[Tuple[str, str]]:
+        """列出所有已注册的转换器"""
+        return list(self._converters.keys())
+
+    # ========== 能力查询方法 ==========
+
+    def can_convert_request(self, source: str, target: str) -> bool:
+        """检查是否支持请求转换"""
+        converter = self.get_converter(source, target)
+        return converter is not None and hasattr(converter, "convert_request")
+
+    def can_convert_response(self, source: str, target: str) -> bool:
+        """检查是否支持响应转换"""
+        converter = self.get_converter(source, target)
+        return converter is not None and hasattr(converter, "convert_response")
+
+    def can_convert_stream(self, source: str, target: str) -> bool:
+        """检查是否支持流式转换"""
+        converter = self.get_converter(source, target)
+        if converter is None:
+            return False
+        return hasattr(converter, "convert_stream_chunk")
+
+    def can_convert_full(
+        self,
+        source: str,
+        target: str,
+        require_stream: bool = False,
+    ) -> bool:
+        """
+        检查是否支持完整的双向转换
+
+        对于跨格式请求，需要：
+        1. 请求转换：source -> target
+        2. 响应转换：target -> source（注意方向相反）
+        3. 流式转换（如果 require_stream=True）：target -> source
+
+        Args:
+            source: 客户端格式
+            target: Provider 格式
+            require_stream: 是否要求支持流式转换
+        """
+        # 请求：client -> provider
+        if not self.can_convert_request(source, target):
+            return False
+        # 响应：provider -> client（方向相反）
+        if not self.can_convert_response(target, source):
+            return False
+        # 流式：provider -> client（方向相反）
+        if require_stream and not self.can_convert_stream(target, source):
+            return False
+        return True
+
+    def get_supported_targets(self, source: str) -> list[str]:
+        """获取指定源格式支持转换到的目标格式列表"""
+        source_upper = source.upper()
+        return [target for (src, target) in self._converters.keys() if src == source_upper]
+
+    # ========== 严格模式方法 ==========
+
+    def convert_request_strict(
+        self,
+        request: Dict[str, Any],
+        source_format: str,
+        target_format: str,
+    ) -> Dict[str, Any]:
+        """
+        严格模式请求转换 - 失败时抛出异常
+
+        用于需要故障转移的场景：转换失败会抛出 FormatConversionError，
+        让 Orchestrator 可以尝试下一个候选。
+
+        Raises:
+            FormatConversionError: 转换失败时抛出
+        """
+        # 同格式无需转换
+        if source_format.upper() == target_format.upper():
+            return request
+
+        converter = self.get_converter(source_format, target_format)
+        if converter is None:
+            raise FormatConversionError(source_format, target_format, "未找到转换器")
+
+        if not hasattr(converter, "convert_request"):
+            raise FormatConversionError(source_format, target_format, "转换器缺少 convert_request 方法")
+
+        try:
+            converted: Dict[str, Any] = converter.convert_request(request)
+            logger.debug(f"[ConverterRegistry] 请求转换成功: {source_format} -> {target_format}")
+            return converted
+        except FormatConversionError:
+            raise
+        except Exception as e:
+            raise FormatConversionError(source_format, target_format, str(e)) from e
+
+    def convert_response_strict(
+        self,
+        response: Dict[str, Any],
+        source_format: str,
+        target_format: str,
+    ) -> Dict[str, Any]:
+        """
+        严格模式响应转换 - 失败时抛出异常
+
+        Raises:
+            FormatConversionError: 转换失败时抛出
+        """
+        if source_format.upper() == target_format.upper():
+            return response
+
+        converter = self.get_converter(source_format, target_format)
+        if converter is None:
+            raise FormatConversionError(source_format, target_format, "未找到转换器")
+
+        if not hasattr(converter, "convert_response"):
+            raise FormatConversionError(source_format, target_format, "转换器缺少 convert_response 方法")
+
+        try:
+            converted: Dict[str, Any] = converter.convert_response(response)
+            logger.debug(f"[ConverterRegistry] 响应转换成功: {source_format} -> {target_format}")
+            return converted
+        except FormatConversionError:
+            raise
+        except Exception as e:
+            raise FormatConversionError(source_format, target_format, str(e)) from e
+
+    def convert_stream_chunk_strict(
+        self,
+        chunk: Dict[str, Any],
+        source_format: str,
+        target_format: str,
+        state: Optional[Union["StreamConversionState", "GeminiStreamConversionState"]] = None,
+    ) -> list[Dict[str, Any]]:
+        """
+        严格模式流式块转换 - 失败时抛出异常
+
+        Args:
+            chunk: 流式响应块
+            source_format: 源格式
+            target_format: 目标格式
+            state: 流式转换状态（StreamConversionState 或 GeminiStreamConversionState）
+
+        Returns:
+            转换后的事件列表（可能 0-N 个）
+
+        Raises:
+            FormatConversionError: 转换失败时抛出
+        """
+        if source_format.upper() == target_format.upper():
+            return [chunk]
+
+        converter = self.get_converter(source_format, target_format)
+        if converter is None:
+            raise FormatConversionError(source_format, target_format, "未找到转换器")
+
+        if not hasattr(converter, "convert_stream_chunk"):
+            raise FormatConversionError(
+                source_format, target_format, "转换器缺少 convert_stream_chunk 方法"
+            )
+
+        try:
+            result: list[Dict[str, Any]] = converter.convert_stream_chunk(chunk, state)
+            return result
+        except FormatConversionError:
+            raise
+        except Exception as e:
+            raise FormatConversionError(source_format, target_format, f"流式块转换失败: {e}") from e
+
+
+# 全局单例
+converter_registry = FormatConverterRegistry()
+
+
+__all__ = [
+    "FormatConverterRegistry",
+    "converter_registry",
+    "FormatConversionError",
+]
+

src/core/api_format/conversion/state.py

@@ -0,0 +1,69 @@
+"""
+流式转换状态类
+
+用于在多个 chunk 之间维护转换上下文，例如：
+- 是否已发送 message_start 事件
+- 累积的文本（用于计算增量）
+- 当前内容块索引
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class StreamConversionState:
+    """
+    SSE 流式转换状态（Claude <-> OpenAI）
+
+    用于跨 chunk 维护转换状态，确保生成正确的事件序列。
+    """
+
+    message_id: str = ""
+    model: str = ""
+    message_started: bool = False
+    content_block_started: bool = False
+    current_tool_index: int = 0
+
+    def reset(self) -> None:
+        """重置状态（重试时调用）"""
+        self.message_started = False
+        self.content_block_started = False
+        self.current_tool_index = 0
+
+
+@dataclass
+class GeminiStreamConversionState:
+    """
+    Gemini 流式转换状态（JSON 数组格式）
+
+    与 Claude/OpenAI 的 SSE 不同，Gemini 需要追踪额外状态：
+    1. 累积文本（用于计算真正的增量）
+    2. 内容块索引（用于工具调用）
+    3. 是否已发送 message_start
+    """
+
+    message_id: str = ""
+    model: str = ""
+    accumulated_text: str = ""  # 累积的文本（用于计算增量）
+    message_started: bool = False
+    content_block_started: bool = False
+    current_block_index: int = 0
+    tool_call_index: int = 0  # 工具调用计数
+    has_sent_usage: bool = False  # 是否已发送 usage
+
+    def reset(self) -> None:
+        """重置状态（重试时调用）"""
+        self.accumulated_text = ""
+        self.message_started = False
+        self.content_block_started = False
+        self.current_block_index = 0
+        self.tool_call_index = 0
+        self.has_sent_usage = False
+
+
+__all__ = [
+    "StreamConversionState",
+    "GeminiStreamConversionState",
+]

src/core/api_format/detection.py

@@ -0,0 +1,183 @@
+"""
+API 格式检测
+
+提供从请求头、响应内容等检测 API 格式的函数。
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Dict, Optional, Tuple
+
+if TYPE_CHECKING:
+    from starlette.requests import Request
+
+from src.core.api_format.enums import APIFormat
+from src.core.api_format.metadata import API_FORMAT_DEFINITIONS, ApiFormatDefinition
+
+
+def _extract_api_key_by_definition(
+    headers: Dict[str, str],
+    query_params: Optional[Dict[str, str]],
+    definition: ApiFormatDefinition,
+) -> Optional[str]:
+    """
+    根据格式定义从请求中提取 API Key
+
+    Args:
+        headers: 请求头字典（key 小写）
+        query_params: 查询参数字典（可选）
+        definition: API 格式定义
+
+    Returns:
+        提取到的 API Key，或 None
+    """
+    auth_header = definition.auth_header.lower()
+    auth_type = definition.auth_type
+
+    header_value = headers.get(auth_header)
+    if not header_value:
+        # Gemini 还支持 ?key= 参数
+        if definition.api_format in (APIFormat.GEMINI, APIFormat.GEMINI_CLI):
+            return query_params.get("key") if query_params else None
+        return None
+
+    if auth_type == "bearer":
+        # Bearer token: "Bearer xxx"
+        if header_value.lower().startswith("bearer "):
+            return header_value[7:].strip()
+        return None
+    else:
+        # header 类型: 直接使用值
+        return header_value
+
+
+def detect_format_from_request(
+    headers: Dict[str, str],
+    query_params: Optional[Dict[str, str]] = None,
+) -> Tuple[APIFormat, Optional[str]]:
+    """
+    从请求头检测 API 格式和 API Key
+
+    检测优先级：
+    1. x-api-key + anthropic-version -> Claude
+    2. x-goog-api-key 或 ?key= -> Gemini
+    3. Authorization: Bearer -> OpenAI (默认)
+
+    Args:
+        headers: 请求头字典（key 应为小写）
+        query_params: 查询参数字典（可选）
+
+    Returns:
+        (APIFormat, api_key) 元组
+    """
+    # Claude: x-api-key + anthropic-version (必须同时存在)
+    claude_def = API_FORMAT_DEFINITIONS[APIFormat.CLAUDE]
+    claude_key = _extract_api_key_by_definition(headers, query_params, claude_def)
+    if claude_key and headers.get("anthropic-version"):
+        return APIFormat.CLAUDE, claude_key
+
+    # Gemini: x-goog-api-key (header 类型) 或 ?key=
+    gemini_def = API_FORMAT_DEFINITIONS[APIFormat.GEMINI]
+    gemini_key = _extract_api_key_by_definition(headers, query_params, gemini_def)
+    if gemini_key:
+        return APIFormat.GEMINI, gemini_key
+
+    # OpenAI: Authorization: Bearer (默认)
+    # 注意: 如果只有 x-api-key 但没有 anthropic-version，也走 OpenAI 格式
+    openai_def = API_FORMAT_DEFINITIONS[APIFormat.OPENAI]
+    openai_key = _extract_api_key_by_definition(headers, query_params, openai_def)
+    # 如果 OpenAI 格式没有 key，但有 x-api-key，也用它（兼容）
+    if not openai_key and claude_key:
+        openai_key = claude_key
+    return APIFormat.OPENAI, openai_key
+
+
+def detect_format_and_key_from_starlette(
+    request: "Request",
+) -> Tuple[str, Optional[str]]:
+    """
+    从 Starlette Request 对象检测 API 格式和 API Key
+
+    这是一个便捷函数，用于直接处理 Starlette/FastAPI 请求对象。
+
+    Args:
+        request: Starlette Request 对象
+
+    Returns:
+        (format_name, api_key) 元组，format_name 为小写字符串
+    """
+    # 规范化 headers 为小写
+    headers = {k.lower(): v for k, v in request.headers.items()}
+    query_params = dict(request.query_params)
+
+    api_format, api_key = detect_format_from_request(headers, query_params)
+
+    # 返回小写格式名
+    format_name = api_format.value.lower()
+    return format_name, api_key
+
+
+def detect_format_from_response(
+    response_data: dict,
+) -> Optional[APIFormat]:
+    """
+    从响应内容检测 API 格式
+
+    Args:
+        response_data: 响应 JSON 字典
+
+    Returns:
+        检测到的格式，或 None
+    """
+    # Claude: 有 type="message" 或特定的 content 结构
+    if response_data.get("type") == "message":
+        return APIFormat.CLAUDE
+    if "content" in response_data and isinstance(response_data["content"], list):
+        first_content = response_data["content"][0] if response_data["content"] else {}
+        if first_content.get("type") in ("text", "tool_use"):
+            return APIFormat.CLAUDE
+
+    # OpenAI: 有 choices 数组
+    if "choices" in response_data:
+        return APIFormat.OPENAI
+
+    # Gemini: 有 candidates 数组
+    if "candidates" in response_data:
+        return APIFormat.GEMINI
+
+    return None
+
+
+def detect_cli_format_from_path(
+    path: str,
+    base_format: APIFormat,
+) -> bool:
+    """
+    根据请求路径检测是否为 CLI 模式
+
+    CLI 模式的特征：
+    - OpenAI CLI: 请求 /responses 路径
+    - Claude CLI: 有特定的路径模式
+    - Gemini CLI: 有特定的路径模式
+
+    Args:
+        path: 请求路径
+        base_format: 基础格式
+
+    Returns:
+        True 如果是 CLI 模式
+    """
+    # OpenAI CLI 特征: /v1/responses 路径
+    if base_format == APIFormat.OPENAI and "/responses" in path:
+        return True
+
+    # 其他 CLI 模式通常由 Adapter 层根据具体业务逻辑判断
+    return False
+
+
+__all__ = [
+    "detect_format_from_request",
+    "detect_format_and_key_from_starlette",
+    "detect_format_from_response",
+    "detect_cli_format_from_path",
+]

src/core/api_format/enums.py

@@ -0,0 +1,21 @@
+"""
+API 格式枚举定义
+
+定义所有支持的 API 格式，决定请求/响应的处理方式。
+"""
+
+from enum import Enum
+
+
+class APIFormat(Enum):
+    """API 格式枚举 - 决定请求/响应的处理方式"""
+
+    CLAUDE = "CLAUDE"  # Claude API 格式
+    CLAUDE_CLI = "CLAUDE_CLI"  # Claude CLI API 格式（使用 authorization: Bearer）
+    OPENAI = "OPENAI"  # OpenAI API 格式
+    OPENAI_CLI = "OPENAI_CLI"  # OpenAI CLI/Responses API 格式（用于 Claude Code 等客户端）
+    GEMINI = "GEMINI"  # Google Gemini API 格式
+    GEMINI_CLI = "GEMINI_CLI"  # Gemini CLI API 格式
+
+
+__all__ = ["APIFormat"]

src/core/api_format/headers.py

@@ -0,0 +1,578 @@
+"""
+统一的请求头处理模块
+
+职责：
+1. 请求头规范化（大小写统一）
+2. 客户端 API Key 提取
+3. 能力需求检测
+4. 上游请求头构建
+5. 响应头过滤
+6. 日志脱敏
+"""
+
+from __future__ import annotations
+
+from typing import AbstractSet, Any, Dict, FrozenSet, Optional, Set
+
+from src.core.api_format.enums import APIFormat
+from src.core.api_format.metadata import get_auth_config, get_extra_headers, get_protected_keys
+
+
+# =============================================================================
+# 头部常量定义
+# =============================================================================
+
+# 转发给上游时需要剔除的头部（系统管理 + 认证替换）
+UPSTREAM_DROP_HEADERS: FrozenSet[str] = frozenset(
+    {
+        # 认证头 - 会被替换为 Provider 的认证
+        "authorization",
+        "x-api-key",
+        "x-goog-api-key",
+        # 系统管理头 - 由 HTTP 客户端重新生成
+        "host",
+        "content-length",
+        "transfer-encoding",
+        "connection",
+        # 编码头 - 避免客户端请求 brotli/zstd 但 httpx 不支持
+        "accept-encoding",
+    }
+)
+
+# 最小必脱敏集合（编译时常量，用于快速路径）
+# 完整脱敏应使用 SystemConfigService.get_sensitive_headers()
+CORE_REDACT_HEADERS: FrozenSet[str] = frozenset(
+    {
+        "authorization",
+        "x-api-key",
+        "x-goog-api-key",
+    }
+)
+
+# Hop-by-hop 头部 (RFC 7230)
+HOP_BY_HOP_HEADERS: FrozenSet[str] = frozenset(
+    {
+        "connection",
+        "keep-alive",
+        "proxy-authenticate",
+        "proxy-authorization",
+        "te",
+        "trailer",
+        "transfer-encoding",
+        "upgrade",
+    }
+)
+
+# 响应时需要过滤的头部（body-dependent + hop-by-hop）
+RESPONSE_DROP_HEADERS: FrozenSet[str] = (
+    frozenset(
+        {
+            "content-length",
+            "content-encoding",
+            "transfer-encoding",
+            "content-type",
+        }
+    )
+    | HOP_BY_HOP_HEADERS
+)
+
+
+# =============================================================================
+# 请求头规范化
+# =============================================================================
+
+
+def normalize_headers(headers: Dict[str, str]) -> Dict[str, str]:
+    """
+    将请求头 key 统一为小写
+
+    用于处理 context.original_headers 的大小写敏感问题。
+    """
+
+    return {k.lower(): v for k, v in headers.items()}
+
+
+def get_header_value(headers: Dict[str, str], key: str, default: str = "") -> str:
+    """
+    大小写不敏感地获取请求头值
+
+    Args:
+        headers: 原始请求头（可能大小写不一致）
+        key: 要获取的 key（任意大小写）
+        default: 未找到时的默认值
+
+    Returns:
+        头部值，未找到返回 default
+    """
+
+    key_lower = key.lower()
+    for k, v in headers.items():
+        if k.lower() == key_lower:
+            return v
+    return default
+
+
+# =============================================================================
+# 客户端 API Key 提取
+# =============================================================================
+
+
+def extract_client_api_key(headers: Dict[str, str], api_format: APIFormat) -> Optional[str]:
+    """
+    从客户端请求头提取 API Key
+
+    自动处理大小写，根据 API 格式使用正确的认证头和类型。
+
+    Args:
+        headers: 原始请求头（自动处理大小写）
+        api_format: API 格式
+
+    Returns:
+        提取的 API Key，未找到返回 None
+    """
+
+    auth_header, auth_type = get_auth_config(api_format)
+    value = get_header_value(headers, auth_header)
+    if not value:
+        return None
+
+    if auth_type == "bearer":
+        # Bearer token 格式: "Bearer <token>"
+        if value.lower().startswith("bearer "):
+            return value[7:]  # 移除 "Bearer " 前缀
+        return None
+
+    # 直接 header 格式
+    return value
+
+
+# =============================================================================
+# 能力需求检测
+# =============================================================================
+
+
+def detect_capabilities(
+    headers: Dict[str, str],
+    api_format: APIFormat,
+    request_body: Optional[Dict[str, Any]] = None,  # noqa: ARG001 - 预留给部分格式使用
+) -> Dict[str, bool]:
+    """
+    从请求头检测能力需求
+
+    当前支持:
+    - Claude/Claude CLI: anthropic-beta 头中的 context-1m
+
+    Args:
+        headers: 原始请求头（自动处理大小写）
+        api_format: API 格式
+        request_body: 请求体（部分格式可能需要）
+
+    Returns:
+        能力需求字典，如 {"context_1m": True}
+    """
+
+    requirements: Dict[str, bool] = {}
+
+    if api_format in (APIFormat.CLAUDE, APIFormat.CLAUDE_CLI):
+        beta_header = get_header_value(headers, "anthropic-beta")
+        if "context-1m" in beta_header.lower():
+            requirements["context_1m"] = True
+
+    return requirements
+
+
+# =============================================================================
+# 上游请求头构建
+# =============================================================================
+
+
+class HeaderBuilder:
+    """
+    请求头构建器
+
+    使用 lower-case key 索引确保唯一性和确定的优先级。
+    优先级（后者覆盖前者）：原始头部 < endpoint 头部 < extra 头部 < 认证头
+    """
+
+    def __init__(self) -> None:
+        # key: (original_case_key, value)
+        self._headers: Dict[str, tuple[str, str]] = {}
+
+    def add(self, key: str, value: str) -> "HeaderBuilder":
+        """添加单个头部（会覆盖同名头部）"""
+        self._headers[key.lower()] = (key, value)
+        return self
+
+    def add_many(self, headers: Dict[str, str]) -> "HeaderBuilder":
+        """批量添加头部"""
+        for k, v in headers.items():
+            self.add(k, v)
+        return self
+
+    def add_protected(self, headers: Dict[str, str], protected_keys: AbstractSet[str]) -> "HeaderBuilder":
+        """
+        添加头部但保护指定的 key 不被覆盖
+
+        用于 endpoint 额外请求头不能覆盖认证头的场景。
+        """
+        protected_lower = {k.lower() for k in protected_keys}
+        for k, v in headers.items():
+            if k.lower() not in protected_lower:
+                self.add(k, v)
+        return self
+
+    def remove(self, keys: FrozenSet[str]) -> "HeaderBuilder":
+        """移除指定的头部"""
+        for k in keys:
+            self._headers.pop(k.lower(), None)
+        return self
+
+    def rename(self, from_key: str, to_key: str) -> "HeaderBuilder":
+        """
+        重命名头部（保留原值）
+
+        如果 from_key 不存在，则不做任何操作。
+        """
+        from_lower = from_key.lower()
+        if from_lower in self._headers:
+            _, value = self._headers.pop(from_lower)
+            self._headers[to_key.lower()] = (to_key, value)
+        return self
+
+    def apply_rules(
+        self,
+        rules: list[Dict[str, Any]],
+        protected_keys: Optional[AbstractSet[str]] = None,
+    ) -> "HeaderBuilder":
+        """
+        应用请求头规则
+
+        支持的规则类型：
+        - set: 设置/覆盖头部 {"action": "set", "key": "X-Custom", "value": "fixed"}
+        - drop: 删除头部 {"action": "drop", "key": "X-Unwanted"}
+        - rename: 重命名头部 {"action": "rename", "from": "X-Old", "to": "X-New"}
+
+        Args:
+            rules: 规则列表
+            protected_keys: 受保护的 key（不能被 set/drop/rename 修改）
+        """
+        protected_lower = {k.lower() for k in protected_keys} if protected_keys else set()
+
+        for rule in rules:
+            action = rule.get("action")
+
+            if action == "set":
+                key = rule.get("key", "")
+                value = rule.get("value", "")
+                if key and key.lower() not in protected_lower:
+                    self.add(key, value)
+
+            elif action == "drop":
+                key = rule.get("key", "")
+                if key and key.lower() not in protected_lower:
+                    self._headers.pop(key.lower(), None)
+
+            elif action == "rename":
+                from_key = rule.get("from", "")
+                to_key = rule.get("to", "")
+                if from_key and to_key:
+                    # 两个 key 都不能是受保护的
+                    if from_key.lower() not in protected_lower and to_key.lower() not in protected_lower:
+                        self.rename(from_key, to_key)
+
+        return self
+
+    def build(self) -> Dict[str, str]:
+        """构建最终的头部字典"""
+        return {original_key: value for original_key, value in self._headers.values()}
+
+
+def build_upstream_headers(
+    original_headers: Dict[str, str],
+    api_format: APIFormat,
+    provider_api_key: str,
+    *,
+    endpoint_headers: Optional[Dict[str, str]] = None,
+    extra_headers: Optional[Dict[str, str]] = None,
+    drop_headers: Optional[FrozenSet[str]] = None,
+) -> Dict[str, str]:
+    """
+    构建发送给上游 Provider 的请求头
+
+    优先级（后者覆盖前者）：
+    1. 原始头部（排除 drop_headers）
+    2. endpoint 配置头部
+    3. extra_headers
+    4. 认证头（最高优先级，始终设置）
+
+    Args:
+        original_headers: 客户端原始请求头
+        api_format: API 格式
+        provider_api_key: Provider 的 API Key（已解密）
+        endpoint_headers: Endpoint 配置的额外头部
+        extra_headers: 调用方传入的额外头部
+        drop_headers: 需要剔除的头部集合（None 使用默认值，空集合表示不剔除）
+
+    Returns:
+        构建好的请求头字典
+    """
+
+    # 使用 is None 判断，允许显式传空集合
+    if drop_headers is None:
+        drop_headers = UPSTREAM_DROP_HEADERS
+
+    auth_header, auth_type = get_auth_config(api_format)
+    auth_value = f"Bearer {provider_api_key}" if auth_type == "bearer" else provider_api_key
+
+    # 认证头是受保护的，不能被 endpoint_headers 覆盖
+    protected_keys = {auth_header.lower(), "content-type"}
+
+    builder = HeaderBuilder()
+
+    # 1. 添加原始头部（排除 drop_headers）
+    for k, v in original_headers.items():
+        if k.lower() not in drop_headers:
+            builder.add(k, v)
+
+    # 2. 添加 endpoint 头部（保护认证头）
+    if endpoint_headers:
+        builder.add_protected(endpoint_headers, protected_keys)
+
+    # 3. 添加 extra_headers
+    if extra_headers:
+        builder.add_many(extra_headers)
+
+    # 4. 设置认证头（最高优先级）
+    builder.add(auth_header, auth_value)
+
+    # 5. 确保 Content-Type
+    result = builder.build()
+    if not any(k.lower() == "content-type" for k in result):
+        result["Content-Type"] = "application/json"
+
+    return result
+
+
+def merge_headers_with_protection(
+    base_headers: Dict[str, str],
+    extra_headers: Optional[Dict[str, str]],
+    protected_keys: FrozenSet[str] | Set[str],
+) -> Dict[str, str]:
+    """
+    合并头部但保护指定的 key 不被覆盖
+
+    等价于原 build_safe_headers 的功能。
+
+    Args:
+        base_headers: 基础头部
+        extra_headers: 要合并的额外头部
+        protected_keys: 受保护的 key 集合
+
+    Returns:
+        合并后的头部
+    """
+    if not extra_headers:
+        return dict(base_headers)
+
+    builder = HeaderBuilder()
+    builder.add_many(base_headers)
+    builder.add_protected(extra_headers, protected_keys)
+    return builder.build()
+
+
+# =============================================================================
+# 响应头过滤
+# =============================================================================
+
+
+def filter_response_headers(
+    headers: Optional[Dict[str, str]],
+    drop_headers: Optional[FrozenSet[str]] = None,
+) -> Dict[str, str]:
+    """
+    过滤上游响应头中不应透传给客户端的字段
+
+    Args:
+        headers: 上游响应头
+        drop_headers: 要剔除的头部集合（None 使用默认值）
+
+    Returns:
+        过滤后的头部
+    """
+    if not headers:
+        return {}
+
+    if drop_headers is None:
+        drop_headers = RESPONSE_DROP_HEADERS
+
+    return {k: v for k, v in headers.items() if k.lower() not in drop_headers}
+
+
+# =============================================================================
+# 日志脱敏
+# =============================================================================
+
+
+def redact_headers_for_log(
+    headers: Dict[str, str],
+    redact_keys: Optional[FrozenSet[str]] = None,
+) -> Dict[str, str]:
+    """
+    将敏感头部值替换为 *** 用于日志记录
+
+    Args:
+        headers: 原始头部
+        redact_keys: 要脱敏的 key 集合（None 使用 CORE_REDACT_HEADERS）
+
+    Returns:
+        脱敏后的头部
+
+    Note:
+        完整的脱敏应该使用 SystemConfigService.get_sensitive_headers()
+        来获取用户配置的敏感头列表。
+    """
+    if redact_keys is None:
+        redact_keys = CORE_REDACT_HEADERS
+
+    return {k: "***" if k.lower() in redact_keys else v for k, v in headers.items()}
+
+
+# =============================================================================
+# 兼容层（向后兼容，逐步废弃）
+# =============================================================================
+
+# 兼容 request_builder.py 的 SENSITIVE_HEADERS
+SENSITIVE_HEADERS = UPSTREAM_DROP_HEADERS
+
+
+# =============================================================================
+# Adapter 统一接口
+# =============================================================================
+
+
+def build_adapter_base_headers(
+    api_format: APIFormat,
+    api_key: str,
+    *,
+    include_extra: bool = True,
+) -> Dict[str, str]:
+    """
+    根据 API 格式构建基础请求头
+
+    包含：认证头 + Content-Type + 格式特定的额外头部（如 anthropic-version）
+
+    Args:
+        api_format: API 格式
+        api_key: API Key（已解密）
+        include_extra: 是否包含格式特定的额外头部（默认 True）
+
+    Returns:
+        基础请求头字典
+    """
+    auth_header, auth_type = get_auth_config(api_format)
+    auth_value = f"Bearer {api_key}" if auth_type == "bearer" else api_key
+
+    headers: Dict[str, str] = {
+        auth_header: auth_value,
+        "Content-Type": "application/json",
+    }
+
+    if include_extra:
+        extra = get_extra_headers(api_format)
+        if extra:
+            headers.update(extra)
+
+    return headers
+
+
+def build_adapter_headers(
+    api_format: APIFormat,
+    api_key: str,
+    extra_headers: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+    """
+    构建完整的 Adapter 请求头
+
+    在基础头部上合并 extra_headers，同时保护关键头部不被覆盖。
+
+    Args:
+        api_format: API 格式
+        api_key: API Key（已解密）
+        extra_headers: 调用方传入的额外头部
+
+    Returns:
+        完整的请求头字典
+    """
+    base = build_adapter_base_headers(api_format, api_key)
+
+    if not extra_headers:
+        return base
+
+    protected = get_protected_keys(api_format)
+    return merge_headers_with_protection(base, extra_headers, protected)
+
+
+def get_adapter_protected_keys(api_format: APIFormat) -> tuple[str, ...]:
+    """
+    获取 Adapter 的受保护头部 key
+
+    用于 get_protected_header_keys() 方法返回值。
+
+    Args:
+        api_format: API 格式
+
+    Returns:
+        受保护的头部 key 元组
+    """
+    return tuple(get_protected_keys(api_format))
+
+
+# =============================================================================
+# Header Rules 工具函数
+# =============================================================================
+
+
+def extract_set_headers_from_rules(
+    header_rules: Optional[list[Dict[str, Any]]],
+) -> Optional[Dict[str, str]]:
+    """
+    从 header_rules 中提取 set 操作生成的头部字典
+
+    用于需要构造额外请求头的场景（如模型列表查询、模型测试等）。
+    注意：drop 和 rename 操作在这里不适用，因为它们用于修改已存在的头部。
+
+    Args:
+        header_rules: 请求头规则列表 [{"action": "set", "key": "X-Custom", "value": "val"}, ...]
+
+    Returns:
+        set 操作生成的头部字典，如果没有则返回 None
+    """
+    if not header_rules:
+        return None
+
+    headers: Dict[str, str] = {}
+    for rule in header_rules:
+        if rule.get("action") == "set":
+            key = rule.get("key", "")
+            value = rule.get("value", "")
+            if key:
+                headers[key] = value
+
+    return headers if headers else None
+
+
+def get_extra_headers_from_endpoint(endpoint: Any) -> Optional[Dict[str, str]]:
+    """
+    从 endpoint 提取额外请求头
+
+    用于需要构造额外请求头的场景（如模型列表查询、模型测试等）。
+
+    Args:
+        endpoint: ProviderEndpoint 对象
+
+    Returns:
+        额外请求头字典，如果没有则返回 None
+    """
+    header_rules = getattr(endpoint, "header_rules", None)
+    return extract_set_headers_from_rules(header_rules)
+

src/core/api_format/metadata.py

@@ -0,0 +1,299 @@
+"""
+API 格式元数据定义
+
+集中维护 API 格式的元数据，避免新增格式时到处修改常量。
+
+使用方式：
+    # 解析格式别名
+    from src.core.api_format import resolve_api_format
+    api_format = resolve_api_format("claude")  # -> APIFormat.CLAUDE
+
+    # 获取格式定义
+    from src.core.api_format import get_api_format_definition
+    definition = get_api_format_definition(APIFormat.CLAUDE)
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from functools import lru_cache
+from types import MappingProxyType
+from typing import Dict, Iterable, List, Mapping, MutableMapping, Optional, Sequence, Union
+
+from .enums import APIFormat
+
+
+@dataclass(frozen=True)
+class ApiFormatDefinition:
+    """
+    描述一个 API 格式的所有通用信息。
+
+    - aliases: 用于 detect_api_format 的 provider 别名或快捷名称
+    - default_path: 上游默认请求路径（如 /v1/messages），可通过 Endpoint.custom_path 覆盖
+    - path_prefix: 本站路径前缀（如 /claude, /openai），为空表示无前缀
+    - auth_header: 认证头名称 (如 "x-api-key", "x-goog-api-key")
+    - auth_type: 认证类型 ("header" 直接放值, "bearer" 加 Bearer 前缀)
+    - extra_headers: 该格式必须携带的额外头部（如 anthropic-version）
+    - protected_keys: 不应被 extra_headers 覆盖的头部（小写）
+    """
+
+    api_format: APIFormat
+    aliases: Sequence[str] = field(default_factory=tuple)
+    default_path: str = "/"  # 上游默认请求路径
+    path_prefix: str = ""  # 本站路径前缀，为空表示无前缀
+    auth_header: str = "Authorization"
+    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（小写）
+
+    def iter_aliases(self) -> Iterable[str]:
+        """返回大小写统一后的别名集合，包含枚举名本身。"""
+        yield normalize_alias_value(self.api_format.value)
+        for alias in self.aliases:
+            normalized = normalize_alias_value(alias)
+            if normalized:
+                yield normalized
+
+
+_DEFINITIONS: Dict[APIFormat, ApiFormatDefinition] = {
+    APIFormat.CLAUDE: ApiFormatDefinition(
+        api_format=APIFormat.CLAUDE,
+        aliases=("claude", "anthropic", "claude_compatible"),
+        default_path="/v1/messages",
+        path_prefix="",  # 通过请求头区分格式，不使用路径前缀
+        auth_header="x-api-key",
+        auth_type="header",
+        extra_headers={"anthropic-version": "2023-06-01"},
+        protected_keys=frozenset({"x-api-key", "content-type", "anthropic-version"}),
+    ),
+    APIFormat.CLAUDE_CLI: ApiFormatDefinition(
+        api_format=APIFormat.CLAUDE_CLI,
+        aliases=("claude_cli", "claude-cli"),
+        default_path="/v1/messages",
+        path_prefix="",  # 与 CLAUDE 共享入口，通过 header 区分
+        auth_header="Authorization",
+        auth_type="bearer",
+        protected_keys=frozenset({"authorization", "content-type"}),
+    ),
+    APIFormat.OPENAI: ApiFormatDefinition(
+        api_format=APIFormat.OPENAI,
+        aliases=(
+            "openai",
+            "deepseek",
+            "grok",
+            "moonshot",
+            "zhipu",
+            "qwen",
+            "baichuan",
+            "minimax",
+            "openai_compatible",
+        ),
+        default_path="/v1/chat/completions",
+        path_prefix="",  # 默认格式
+        auth_header="Authorization",
+        auth_type="bearer",
+        protected_keys=frozenset({"authorization", "content-type"}),
+    ),
+    APIFormat.OPENAI_CLI: ApiFormatDefinition(
+        api_format=APIFormat.OPENAI_CLI,
+        aliases=("openai_cli", "responses"),
+        default_path="/responses",
+        path_prefix="",  # 与 OPENAI 共享入口
+        auth_header="Authorization",
+        auth_type="bearer",
+        protected_keys=frozenset({"authorization", "content-type"}),
+    ),
+    APIFormat.GEMINI: ApiFormatDefinition(
+        api_format=APIFormat.GEMINI,
+        aliases=("gemini", "google", "vertex"),
+        default_path="/v1beta/models/{model}:{action}",
+        path_prefix="",  # 通过请求头区分格式
+        auth_header="x-goog-api-key",
+        auth_type="header",
+        protected_keys=frozenset({"x-goog-api-key", "content-type"}),
+    ),
+    APIFormat.GEMINI_CLI: ApiFormatDefinition(
+        api_format=APIFormat.GEMINI_CLI,
+        aliases=("gemini_cli", "gemini-cli"),
+        default_path="/v1beta/models/{model}:{action}",
+        path_prefix="",  # 与 GEMINI 共享入口
+        auth_header="x-goog-api-key",
+        auth_type="header",
+        protected_keys=frozenset({"x-goog-api-key", "content-type"}),
+    ),
+}
+
+# 对外只暴露只读视图，避免被随意修改
+API_FORMAT_DEFINITIONS: Mapping[APIFormat, ApiFormatDefinition] = MappingProxyType(_DEFINITIONS)
+
+
+def get_api_format_definition(api_format: APIFormat) -> ApiFormatDefinition:
+    """获取指定格式的定义，不存在时抛出 KeyError。"""
+    return API_FORMAT_DEFINITIONS[api_format]
+
+
+def list_api_format_definitions() -> List[ApiFormatDefinition]:
+    """返回所有定义的浅拷贝列表，供遍历使用。"""
+    return list(API_FORMAT_DEFINITIONS.values())
+
+
+def build_alias_lookup() -> Dict[str, APIFormat]:
+    """
+    构建 alias -> APIFormat 的查找表。
+    每次调用都会返回新的 dict，避免可变全局引发并发问题。
+    """
+    lookup: MutableMapping[str, APIFormat] = {}
+    for definition in API_FORMAT_DEFINITIONS.values():
+        for alias in definition.iter_aliases():
+            lookup.setdefault(alias, definition.api_format)
+    return dict(lookup)
+
+
+def get_default_path(api_format: APIFormat) -> str:
+    """
+    获取该格式的上游默认请求路径。
+
+    可通过 Endpoint.custom_path 覆盖。
+    """
+    definition = API_FORMAT_DEFINITIONS.get(api_format)
+    return definition.default_path if definition else "/"
+
+
+def get_local_path(api_format: APIFormat) -> str:
+    """
+    获取该格式的本站入口路径。
+
+    本站入口路径 = path_prefix + default_path
+    例如：path_prefix="/openai" + default_path="/v1/chat/completions" -> "/openai/v1/chat/completions"
+    """
+    definition = API_FORMAT_DEFINITIONS.get(api_format)
+    if definition:
+        prefix = definition.path_prefix or ""
+        return prefix + definition.default_path
+    return "/"
+
+
+def get_auth_config(api_format: APIFormat) -> tuple[str, str]:
+    """
+    获取该格式的认证配置。
+
+    Returns:
+        (auth_header, auth_type) 元组
+        - auth_header: 认证头名称
+        - auth_type: "bearer" 或 "header"
+    """
+    definition = API_FORMAT_DEFINITIONS.get(api_format)
+    if definition:
+        return definition.auth_header, definition.auth_type
+    return "Authorization", "bearer"
+
+
+def get_extra_headers(api_format: APIFormat) -> Mapping[str, str]:
+    """
+    获取该格式必须携带的额外头部。
+
+    例如 Claude 需要 anthropic-version 头部。
+
+    Returns:
+        额外头部字典（只读）
+    """
+    definition = API_FORMAT_DEFINITIONS.get(api_format)
+    if definition:
+        return definition.extra_headers
+    return {}
+
+
+def get_protected_keys(api_format: APIFormat) -> frozenset[str]:
+    """
+    获取该格式的受保护头部 key（小写）。
+
+    这些头部不应被 extra_headers 覆盖。
+
+    Returns:
+        受保护的头部 key 集合
+    """
+    definition = API_FORMAT_DEFINITIONS.get(api_format)
+    if definition:
+        return definition.protected_keys
+    return frozenset({"authorization", "content-type"})
+
+
+@lru_cache(maxsize=1)
+def _alias_lookup_cache() -> Dict[str, APIFormat]:
+    """缓存 alias -> APIFormat 查找表，减少重复构建。"""
+    return build_alias_lookup()
+
+
+def resolve_api_format_alias(value: str) -> Optional[APIFormat]:
+    """根据别名查找 APIFormat，找不到时返回 None。"""
+    if not value:
+        return None
+    normalized = normalize_alias_value(value)
+    if not normalized:
+        return None
+    return _alias_lookup_cache().get(normalized)
+
+
+def resolve_api_format(
+    value: Union[str, APIFormat, None],
+    default: Optional[APIFormat] = None,
+) -> Optional[APIFormat]:
+    """
+    将任意字符串/枚举值解析为 APIFormat。
+
+    Args:
+        value: 可以是 APIFormat 或任意字符串/别名
+        default: 未解析成功时返回的默认值
+    """
+    if isinstance(value, APIFormat):
+        return value
+    if isinstance(value, str):
+        stripped = value.strip()
+        if not stripped:
+            return default
+        upper = stripped.upper()
+        if upper in APIFormat.__members__:
+            return APIFormat[upper]
+        alias = resolve_api_format_alias(stripped)
+        if alias:
+            return alias
+    return default
+
+
+def register_api_format_definition(definition: ApiFormatDefinition, *, override: bool = False) -> None:
+    """
+    注册或覆盖 API 格式定义，允许运行时扩展。
+
+    Args:
+        definition: 要注册的定义
+        override: 若目标枚举已存在，是否允许覆盖
+    """
+    existing = _DEFINITIONS.get(definition.api_format)
+    if existing and not override:
+        raise ValueError(f"{definition.api_format.value} 已存在，如需覆盖请设置 override=True")
+    _DEFINITIONS[definition.api_format] = definition
+    _refresh_metadata_cache()
+
+
+def _refresh_metadata_cache() -> None:
+    """更新别名缓存，供注册函数调用。"""
+    _alias_lookup_cache.cache_clear()
+
+
+def normalize_alias_value(value: str) -> str:
+    """统一别名格式：去空白、转小写，并将非字母数字转为单个下划线。"""
+    if value is None:
+        return ""
+    text = value.strip().lower()
+    # 将所有非字母数字字符替换为下划线，并折叠连续的下划线
+    text = re.sub(r"[^a-z0-9]+", "_", text)
+    return text.strip("_")
+
+
+# is_cli_format 和 is_cli_api_format 已移至 utils.py
+# 为保持兼容性，从 utils 重新导出
+from src.core.api_format.utils import is_cli_format  # noqa: E402
+
+# is_cli_api_format 是 is_cli_format 的别名（接受 APIFormat 枚举）
+is_cli_api_format = is_cli_format

src/core/api_format/utils.py

@@ -0,0 +1,106 @@
+"""
+API 格式工具函数
+
+提供格式判断、规范化等工具函数，供整个项目使用。
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional, Union
+
+if TYPE_CHECKING:
+    from src.core.api_format.enums import APIFormat
+
+
+def is_cli_format(format_id: Union[str, "APIFormat", None]) -> bool:
+    """
+    判断是否为 CLI 透传格式
+
+    CLI 格式以 _CLI 结尾，不参与格式转换，请求直接透传。
+
+    Args:
+        format_id: 格式标识符（字符串或 APIFormat 枚举）
+
+    Returns:
+        True 如果是 CLI 格式
+
+    Examples:
+        >>> is_cli_format("CLAUDE_CLI")
+        True
+        >>> is_cli_format("CLAUDE")
+        False
+        >>> is_cli_format(APIFormat.OPENAI_CLI)
+        True
+    """
+    if format_id is None:
+        return False
+    if hasattr(format_id, "value"):
+        format_id = format_id.value
+    return str(format_id).upper().endswith("_CLI")
+
+
+def get_base_format(format_id: Union[str, "APIFormat", None]) -> Optional[str]:
+    """
+    获取基础格式（去除 _CLI 后缀）
+
+    Args:
+        format_id: 格式标识符
+
+    Returns:
+        基础格式字符串，或 None
+
+    Examples:
+        >>> get_base_format("CLAUDE_CLI")
+        "CLAUDE"
+        >>> get_base_format("OPENAI")
+        "OPENAI"
+    """
+    if format_id is None:
+        return None
+    if hasattr(format_id, "value"):
+        format_id = format_id.value
+    format_str = str(format_id).upper()
+    if format_str.endswith("_CLI"):
+        return format_str[:-4]
+    return format_str
+
+
+def normalize_format(format_id: Union[str, "APIFormat", None]) -> Optional[str]:
+    """
+    规范化格式标识符
+
+    Args:
+        format_id: 格式标识符（可能是字符串、枚举或 None）
+
+    Returns:
+        大写的格式字符串，或 None
+    """
+    if format_id is None:
+        return None
+    if hasattr(format_id, "value"):
+        return str(format_id.value).upper()
+    return str(format_id).upper()
+
+
+def is_same_format(
+    format1: Union[str, "APIFormat", None],
+    format2: Union[str, "APIFormat", None],
+) -> bool:
+    """
+    判断两个格式是否相同
+
+    忽略大小写和枚举/字符串差异。
+    """
+    return normalize_format(format1) == normalize_format(format2)
+
+
+def is_convertible_format(format_id: Union[str, "APIFormat", None]) -> bool:
+    """
+    判断是否为可转换格式（非 CLI）
+
+    可转换格式可以与其他格式进行双向转换。
+    CLI 格式为透传模式，不参与转换。
+    """
+    if format_id is None:
+        return False
+    return not is_cli_format(format_id)