refactor(api_format): 重构为结构化的 (ApiFamily, EndpointKind) 标识体系

将扁平的 APIFormat 枚举 (CLAUDE, OPENAI, GEMINI, *_CLI) 重构为二维结构：
- ApiFamily: 协议族 (openai, claude, gemini)
- EndpointKind: 端点变体 (chat, cli, video, image)

新增 EndpointSignature 数据类和 `family:kind` 签名键工具函数，
统一 JSON dict/metrics/logs 中的字符串标识格式。

frontend/src/features/usage/components/UsageApiFormatTable.vue

@@ -76,6 +76,7 @@ import TableRow from '@/components/ui/table-row.vue'
 import TableHead from '@/components/ui/table-head.vue'
 import TableCell from '@/components/ui/table-cell.vue'
 import { formatTokens, formatCurrency } from '@/utils/format'
+import { API_FORMAT_LABELS } from '@/api/endpoints/types'
 import type { ApiFormatStatsItem } from '../types'
 
 defineProps<{
@@ -85,15 +86,13 @@ defineProps<{
 
 // 格式化 API 格式显示名称
 function formatApiFormat(format: string): string {
-  const formatMap: Record<string, string> = {
-    'CLAUDE': 'Claude',
-    'CLAUDE_CLI': 'Claude CLI',
-    'OPENAI': 'OpenAI',
-    'OPENAI_CLI': 'OpenAI CLI',
-    'GEMINI': 'Gemini',
-    'GEMINI_CLI': 'Gemini CLI',
-  }
-  return formatMap[format.toUpperCase()] || format
+  const raw = (format || '').trim()
+  return (
+    API_FORMAT_LABELS[raw] ||
+    API_FORMAT_LABELS[raw.toLowerCase()] ||
+    API_FORMAT_LABELS[raw.toUpperCase()] ||
+    raw
+  )
 }
 
 </script>

src/core/api_format/__init__.py

@@ -1,14 +1,8 @@
 """
-API 格式核心模块
+API 格式核心模块（新模式）。
 
-统一管理 API 格式相关的枚举、元数据、工具函数等。
-
-模块组成：
-- enums.py: APIFormat 枚举定义
-- metadata.py: 格式元数据定义（别名、路径、认证等）
-- headers.py: 请求头处理（构建、过滤、脱敏）
-- utils.py: 工具函数（is_cli_format, get_base_format 等）
-- detection.py: 格式检测（从请求头、响应内容检测格式）
+系统内部统一使用 endpoint signature key 作为“格式”标识：
+    `<api_family>:<endpoint_kind>`（全小写，例如 "openai:chat"）。
 """
 
 from src.core.api_format.auth import (
@@ -19,7 +13,7 @@ from src.core.api_format.auth import (
     OAuth2AuthHandler,
     QueryKeyAuthHandler,
     get_auth_handler,
-    get_default_auth_method,
+    get_default_auth_method_for_endpoint,
 )
 from src.core.api_format.detection import (
     RequestContext,
@@ -29,23 +23,22 @@ from src.core.api_format.detection import (
     detect_format_from_response,
     detect_request_context,
 )
-from src.core.api_format.enums import APIFormat, AuthMethod, EndpointType
+from src.core.api_format.enums import ApiFamily, AuthMethod, EndpointKind, EndpointType
 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_client_api_key_with_query,
+    build_adapter_base_headers_for_endpoint,
+    build_adapter_headers_for_endpoint,
+    build_upstream_headers_for_endpoint,
+    detect_capabilities_for_endpoint,
+    extract_client_api_key_for_endpoint,
+    extract_client_api_key_for_endpoint_with_query,
     extract_set_headers_from_rules,
     filter_response_headers,
-    get_adapter_protected_keys,
+    get_adapter_protected_keys_for_endpoint,
     get_extra_headers_from_endpoint,
     get_header_value,
     merge_headers_with_protection,
@@ -53,19 +46,25 @@ from src.core.api_format.headers import (
     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,
+    ENDPOINT_DEFINITIONS,
+    EndpointDefinition,
+    can_passthrough_endpoint,
+    get_auth_config_for_endpoint,
+    get_data_format_id_for_endpoint,
+    get_default_path_for_endpoint,
+    get_endpoint_definition,
+    get_extra_headers_for_endpoint,
+    get_local_path_for_endpoint,
+    get_protected_keys_for_endpoint,
+    list_endpoint_definitions,
+    make_endpoint_signature,
+    resolve_endpoint_definition,
+)
+from src.core.api_format.signature import (
+    EndpointSignature,
+    make_signature_key,
+    normalize_signature_key,
+    parse_signature_key,
 )
 from src.core.api_format.utils import (
     get_base_format,
@@ -77,23 +76,29 @@ from src.core.api_format.utils import (
 
 __all__ = [
     # Enums
-    "APIFormat",
+    "ApiFamily",
+    "EndpointKind",
     "AuthMethod",
     "EndpointType",
+    # Signature
+    "EndpointSignature",
+    "make_signature_key",
+    "parse_signature_key",
+    "normalize_signature_key",
     # 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",
+    "EndpointDefinition",
+    "ENDPOINT_DEFINITIONS",
+    "list_endpoint_definitions",
+    "get_endpoint_definition",
+    "resolve_endpoint_definition",
+    "make_endpoint_signature",
+    "get_default_path_for_endpoint",
+    "get_local_path_for_endpoint",
+    "get_auth_config_for_endpoint",
+    "get_extra_headers_for_endpoint",
+    "get_protected_keys_for_endpoint",
+    "get_data_format_id_for_endpoint",
+    "can_passthrough_endpoint",
     # Utils
     "is_cli_format",
     "get_base_format",
@@ -105,20 +110,19 @@ __all__ = [
     "CORE_REDACT_HEADERS",
     "HOP_BY_HOP_HEADERS",
     "RESPONSE_DROP_HEADERS",
-    "SENSITIVE_HEADERS",
     "normalize_headers",
     "get_header_value",
-    "extract_client_api_key",
-    "extract_client_api_key_with_query",
-    "detect_capabilities",
+    "extract_client_api_key_for_endpoint",
+    "extract_client_api_key_for_endpoint_with_query",
+    "detect_capabilities_for_endpoint",
     "HeaderBuilder",
-    "build_upstream_headers",
+    "build_upstream_headers_for_endpoint",
     "merge_headers_with_protection",
     "filter_response_headers",
     "redact_headers_for_log",
-    "build_adapter_base_headers",
-    "build_adapter_headers",
-    "get_adapter_protected_keys",
+    "build_adapter_base_headers_for_endpoint",
+    "build_adapter_headers_for_endpoint",
+    "get_adapter_protected_keys_for_endpoint",
     "extract_set_headers_from_rules",
     "get_extra_headers_from_endpoint",
     # Detection
@@ -136,5 +140,5 @@ __all__ = [
     "OAuth2AuthHandler",
     "QueryKeyAuthHandler",
     "get_auth_handler",
-    "get_default_auth_method",
+    "get_default_auth_method_for_endpoint",
 ]

src/core/api_format/auth.py

@@ -9,7 +9,9 @@ from __future__ import annotations
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING
 
-from src.core.api_format.enums import APIFormat, AuthMethod
+from src.core.api_format.enums import AuthMethod
+from src.core.api_format.metadata import resolve_endpoint_definition
+from src.core.api_format.signature import EndpointSignature
 
 if TYPE_CHECKING:
     from starlette.requests import Request
@@ -104,17 +106,16 @@ def get_auth_handler(auth_method: AuthMethod) -> AuthHandler:
     return handler
 
 
-def get_default_auth_method(api_format: APIFormat) -> AuthMethod:
-    """从 APIFormat 推断默认 AuthMethod（兼容旧逻辑）"""
-    mapping = {
-        APIFormat.OPENAI: AuthMethod.BEARER,
-        APIFormat.OPENAI_CLI: AuthMethod.BEARER,
-        APIFormat.CLAUDE: AuthMethod.API_KEY,
-        APIFormat.CLAUDE_CLI: AuthMethod.BEARER,
-        APIFormat.GEMINI: AuthMethod.GOOG_API_KEY,
-        APIFormat.GEMINI_CLI: AuthMethod.GOOG_API_KEY,
-    }
-    return mapping.get(api_format, AuthMethod.BEARER)
+def get_default_auth_method_for_endpoint(
+    value: str | EndpointSignature | tuple,  # tuple[ApiFamily, EndpointKind]
+) -> AuthMethod:
+    """
+    新模式：从 endpoint signature 推断默认 AuthMethod。
+
+    只接受 `family:kind` / EndpointSignature / (ApiFamily, EndpointKind)。
+    """
+    definition = resolve_endpoint_definition(value)
+    return definition.auth_method if definition else AuthMethod.BEARER
 
 
 __all__ = [
@@ -125,5 +126,5 @@ __all__ = [
     "OAuth2AuthHandler",
     "QueryKeyAuthHandler",
     "get_auth_handler",
-    "get_default_auth_method",
+    "get_default_auth_method_for_endpoint",
 ]

src/core/api_format/conversion/compatibility.py

@@ -12,14 +12,13 @@
 
 from __future__ import annotations
 
-
 import logging
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from src.core.api_format.conversion.registry import FormatConversionRegistry
 
-from src.core.api_format.metadata import can_passthrough
+from src.core.api_format.metadata import can_passthrough_endpoint
 
 logger = logging.getLogger(__name__)
 
@@ -59,15 +58,16 @@ def is_format_compatible(
         register_default_normalizers()
         registry = format_conversion_registry
 
-    provider_format = endpoint_api_format.upper()
-    client_format_upper = client_format.upper()
+    # 统一大写用于比较和 registry 查找（registry 以大写 key 索引 normalizer）
+    client_key = client_format.upper()
+    provider_key = endpoint_api_format.upper()
 
     # 1. 格式完全匹配 -> 透传（无需转换）
-    if provider_format == client_format_upper:
+    if provider_key == client_key:
         return True, False, None
 
     # 2. 格式不同 -> 需要检查全局格式转换开关
-    # 即使 data_format_id 相同（如 CLAUDE/CLAUDE_CLI），也需要全局开关启用
+    # 即使 data_format_id 相同（如 claude:chat / claude:cli），也需要全局开关启用
     if not global_conversion_enabled:
         return False, False, "全局格式转换未启用（环境变量 FORMAT_CONVERSION_ENABLED=false）"
 
@@ -83,18 +83,18 @@ def is_format_compatible(
 
     # 检查 reject_formats（优先）
     reject_formats = config.get("reject_formats", [])
-    if client_format_upper in [f.upper() for f in reject_formats]:
+    if client_key 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]:
+    if accept_formats and client_key not in [f.upper() for f in accept_formats]:
         return False, False, f"端点不接受 {client_format} 格式"
 
     # 4. 检查是否可以透传（data_format_id 相同）
-    # 例如：CLAUDE/CLAUDE_CLI 的 data_format_id 都是 "claude"，数据格式相同可透传
-    #      OPENAI 是 "openai_chat"，OPENAI_CLI 是 "openai_responses"，需要转换
-    if can_passthrough(client_format_upper, provider_format):
+    # 例如：claude:chat / claude:cli 的 data_format_id 都是 "claude"，数据格式相同可透传
+    #      openai:chat 是 "openai_chat"，openai:cli 是 "openai_responses"，需要转换
+    if can_passthrough_endpoint(client_key, provider_key):
         # data_format_id 相同，可透传（无需数据转换）
         return True, False, None
 
@@ -105,11 +105,11 @@ def is_format_compatible(
 
     # 6. 检查转换器能力
     if not registry.can_convert_full(
-        client_format_upper,
-        provider_format,
+        client_key,
+        provider_key,
         require_stream=is_stream,
     ):
-        return False, False, f"不存在 {client_format} <-> {provider_format} 的完整转换器"
+        return False, False, f"不存在 {client_format} <-> {endpoint_api_format} 的完整转换器"
 
     return True, True, None
 

src/core/api_format/conversion/exceptions.py

@@ -5,7 +5,6 @@
 """
 
 
-
 class FormatConversionError(Exception):
     """
     格式转换失败异常

src/core/api_format/conversion/field_mappings.py

@@ -9,9 +9,6 @@
   这些应复用 `src/core/api_format/metadata.py`（API_FORMAT_DEFINITIONS）作为单一事实来源。
 """
 
-
-
-
 # 角色映射（仅作为辅助；system/tool 的具体落点以 Normalizer 规则为准）
 ROLE_MAPPINGS: dict[str, dict[str, str]] = {
     "OPENAI": {
@@ -128,4 +125,3 @@ __all__ = [
     "ERROR_TYPE_MAPPINGS",
     "RETRYABLE_ERROR_TYPES",
 ]
-

src/core/api_format/conversion/internal.py

@@ -10,7 +10,6 @@
 - 兼容优先：UnknownBlock 在内部保留，但默认在输出阶段丢弃（可观测、可随时调整策略）
 """
 
-
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any
@@ -293,4 +292,3 @@ __all__ = [
     "InternalError",
     "FormatCapabilities",
 ]
-

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

@@ -6,6 +6,4 @@ Normalizers
 本目录在 Phase 1 仅创建结构；具体实现将在 Phase 2+ 补齐。
 """
 
-
 __all__: list[str] = []
-

src/core/api_format/conversion/normalizers/claude.py

@@ -7,7 +7,6 @@ Claude Messages API Normalizer
 - 可选：Claude error <-> InternalError
 """
 
-
 import json
 from typing import Any
 
@@ -54,7 +53,7 @@ from src.core.api_format.conversion.stream_state import StreamState
 
 
 class ClaudeNormalizer(FormatNormalizer):
-    FORMAT_ID = "CLAUDE"
+    FORMAT_ID = "claude:chat"
     capabilities = FormatCapabilities(
         supports_stream=True,
         supports_error_conversion=True,
@@ -161,7 +160,9 @@ class ClaudeNormalizer(FormatNormalizer):
         # Claude Messages API: messages[] 仅允许 user/assistant，且需要交替；这里做最小修复
         fixed_messages = self._coerce_claude_message_sequence(internal.messages)
 
-        out_messages: list[dict[str, Any]] = [self._internal_message_to_claude(m) for m in fixed_messages]
+        out_messages: list[dict[str, Any]] = [
+            self._internal_message_to_claude(m) for m in fixed_messages
+        ]
 
         result: dict[str, Any] = {
             "model": internal.model,
@@ -285,7 +286,9 @@ class ClaudeNormalizer(FormatNormalizer):
 
         stop_reason = None
         if internal.stop_reason is not None:
-            stop_reason = STOP_REASON_MAPPINGS.get("CLAUDE", {}).get(internal.stop_reason.value, "end_turn")
+            stop_reason = STOP_REASON_MAPPINGS.get("CLAUDE", {}).get(
+                internal.stop_reason.value, "end_turn"
+            )
 
         usage: dict[str, Any] = {"input_tokens": 0, "output_tokens": 0}
         if internal.usage:
@@ -353,7 +356,9 @@ class ClaudeNormalizer(FormatNormalizer):
             btype = str(block.get("type") or "unknown")
 
             if btype == "text":
-                events.append(ContentBlockStartEvent(block_index=index, block_type=ContentType.TEXT))
+                events.append(
+                    ContentBlockStartEvent(block_index=index, block_type=ContentType.TEXT)
+                )
                 return events
 
             if btype == "tool_use":
@@ -402,7 +407,9 @@ class ClaudeNormalizer(FormatNormalizer):
                 tool_id = ""
                 if isinstance(mapping, dict):
                     tool_id = str(mapping.get(index) or "")
-                events.append(ToolCallDeltaEvent(block_index=index, tool_id=tool_id, input_delta=str(partial)))
+                events.append(
+                    ToolCallDeltaEvent(block_index=index, tool_id=tool_id, input_delta=str(partial))
+                )
                 return events
 
             return events
@@ -528,7 +535,9 @@ class ClaudeNormalizer(FormatNormalizer):
         if isinstance(event, MessageStopEvent):
             stop_reason = None
             if event.stop_reason is not None:
-                stop_reason = STOP_REASON_MAPPINGS.get("CLAUDE", {}).get(event.stop_reason.value, "end_turn")
+                stop_reason = STOP_REASON_MAPPINGS.get("CLAUDE", {}).get(
+                    event.stop_reason.value, "end_turn"
+                )
 
             msg_delta: dict[str, Any] = {
                 "type": "message_delta",
@@ -592,7 +601,9 @@ class ClaudeNormalizer(FormatNormalizer):
     # Helpers
     # =========================
 
-    def _claude_message_to_internal(self, msg: dict[str, Any]) -> tuple[InternalMessage | None, dict[str, int]]:
+    def _claude_message_to_internal(
+        self, msg: dict[str, Any]
+    ) -> tuple[InternalMessage | None, dict[str, int]]:
         dropped: dict[str, int] = {}
         role_raw = str(msg.get("role") or "unknown")
 
@@ -635,7 +646,9 @@ class ClaudeNormalizer(FormatNormalizer):
             if btype == "text":
                 text = str(block.get("text") or "")
                 if text:
-                    blocks.append(TextBlock(text=text, extra=self._extract_extra(block, {"type", "text"})))
+                    blocks.append(
+                        TextBlock(text=text, extra=self._extract_extra(block, {"type", "text"}))
+                    )
                 continue
 
             if btype == "image":
@@ -645,7 +658,12 @@ class ClaudeNormalizer(FormatNormalizer):
                 if stype == "base64":
                     data = src.get("data")
                     media_type = src.get("media_type")
-                    if isinstance(data, str) and data and isinstance(media_type, str) and media_type:
+                    if (
+                        isinstance(data, str)
+                        and data
+                        and isinstance(media_type, str)
+                        and media_type
+                    ):
                         blocks.append(ImageBlock(data=data, media_type=media_type))
                         continue
                 dropped["claude_image_unsupported"] = dropped.get("claude_image_unsupported", 0) + 1
@@ -663,7 +681,9 @@ class ClaudeNormalizer(FormatNormalizer):
                         tool_id=tool_id,
                         tool_name=tool_name,
                         tool_input=tool_input,
-                        extra={"claude": self._extract_extra(block, {"type", "id", "name", "input"})},
+                        extra={
+                            "claude": self._extract_extra(block, {"type", "id", "name", "input"})
+                        },
                     )
                 )
                 continue
@@ -672,7 +692,9 @@ class ClaudeNormalizer(FormatNormalizer):
                 tool_use_id = str(block.get("tool_use_id") or "")
                 is_error = bool(block.get("is_error") or False)
                 raw_content = block.get("content")
-                blocks.append(self._tool_result_from_claude(tool_use_id, raw_content, is_error, block))
+                blocks.append(
+                    self._tool_result_from_claude(tool_use_id, raw_content, is_error, block)
+                )
                 continue
 
             dropped_key = f"claude_block:{btype}"
@@ -757,7 +779,9 @@ class ClaudeNormalizer(FormatNormalizer):
             texts: list[str] = []
             for item in system_value:
                 if not isinstance(item, dict):
-                    dropped["claude_system_item_non_dict"] = dropped.get("claude_system_item_non_dict", 0) + 1
+                    dropped["claude_system_item_non_dict"] = (
+                        dropped.get("claude_system_item_non_dict", 0) + 1
+                    )
                     continue
                 if item.get("type") == "text":
                     text = item.get("text")
@@ -792,8 +816,14 @@ class ClaudeNormalizer(FormatNormalizer):
                 ToolDefinition(
                     name=name,
                     description=tool.get("description"),
-                    parameters=tool.get("input_schema") if isinstance(tool.get("input_schema"), dict) else None,
-                    extra={"claude": self._extract_extra(tool, {"name", "description", "input_schema"})},
+                    parameters=(
+                        tool.get("input_schema")
+                        if isinstance(tool.get("input_schema"), dict)
+                        else None
+                    ),
+                    extra={
+                        "claude": self._extract_extra(tool, {"name", "description", "input_schema"})
+                    },
                 )
             )
         return out or None
@@ -813,7 +843,9 @@ class ClaudeNormalizer(FormatNormalizer):
             return ToolChoice(type=ToolChoiceType.REQUIRED, extra={"claude": tool_choice})
         if ctype in ("tool_use", "tool"):
             name = str(tool_choice.get("name") or "")
-            return ToolChoice(type=ToolChoiceType.TOOL, tool_name=name, extra={"claude": tool_choice})
+            return ToolChoice(
+                type=ToolChoiceType.TOOL, tool_name=name, extra={"claude": tool_choice}
+            )
 
         return ToolChoice(type=ToolChoiceType.AUTO, extra={"claude": tool_choice})
 
@@ -855,7 +887,11 @@ class ClaudeNormalizer(FormatNormalizer):
                     blocks.append(
                         {
                             "type": "image",
-                            "source": {"type": "base64", "media_type": b.media_type, "data": b.data},
+                            "source": {
+                                "type": "base64",
+                                "media_type": b.media_type,
+                                "data": b.data,
+                            },
                         }
                     )
                 elif b.url:
@@ -901,7 +937,9 @@ class ClaudeNormalizer(FormatNormalizer):
 
         return {"role": role, "content": blocks}
 
-    def _coerce_claude_message_sequence(self, messages: list[InternalMessage]) -> list[InternalMessage]:
+    def _coerce_claude_message_sequence(
+        self, messages: list[InternalMessage]
+    ) -> list[InternalMessage]:
         normalized: list[InternalMessage] = []
         for m in messages:
             role = m.role
@@ -940,7 +978,9 @@ class ClaudeNormalizer(FormatNormalizer):
                     continue
 
         if "total_tokens" not in fields:
-            fields["total_tokens"] = int(fields.get("input_tokens", 0) + fields.get("output_tokens", 0))
+            fields["total_tokens"] = int(
+                fields.get("input_tokens", 0) + fields.get("output_tokens", 0)
+            )
 
         return UsageInfo(
             input_tokens=int(fields.get("input_tokens", 0)),

src/core/api_format/conversion/normalizers/claude_cli.py

@@ -7,13 +7,11 @@ CLAUDE_CLI 的请求/响应 body 与 CLAUDE 一致（Anthropic Messages API）
 如需 CLI 特殊处理，可覆盖 request_from_internal / request_to_internal 等方法。
 """
 
-
 from src.core.api_format.conversion.normalizers.claude import ClaudeNormalizer
 
 
 class ClaudeCliNormalizer(ClaudeNormalizer):
-    FORMAT_ID = "CLAUDE_CLI"
+    FORMAT_ID = "claude:cli"
 
 
 __all__ = ["ClaudeCliNormalizer"]
-

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

@@ -63,7 +63,7 @@ from src.core.api_format.conversion.stream_state import StreamState
 
 
 class GeminiNormalizer(FormatNormalizer):
-    FORMAT_ID = "GEMINI"
+    FORMAT_ID = "gemini:chat"
     capabilities = FormatCapabilities(
         supports_stream=True,
         supports_error_conversion=True,

src/core/api_format/conversion/normalizers/gemini_cli.py

@@ -7,13 +7,11 @@ GEMINI_CLI 的请求/响应 body 与 GEMINI 一致（Google Gemini API），差
 如需 CLI 特殊处理，可覆盖 request_from_internal / request_to_internal 等方法。
 """
 
-
 from src.core.api_format.conversion.normalizers.gemini import GeminiNormalizer
 
 
 class GeminiCliNormalizer(GeminiNormalizer):
-    FORMAT_ID = "GEMINI_CLI"
+    FORMAT_ID = "gemini:cli"
 
 
 __all__ = ["GeminiCliNormalizer"]
-

src/core/api_format/conversion/normalizers/openai.py

@@ -60,7 +60,8 @@ from src.core.logger import logger
 
 
 class OpenAINormalizer(FormatNormalizer):
-    FORMAT_ID = "OPENAI"
+    # 新模式：ApiFamily + EndpointKind 的 signature key
+    FORMAT_ID = "openai:chat"
     capabilities = FormatCapabilities(
         supports_stream=True,
         supports_error_conversion=True,

src/core/api_format/conversion/normalizers/openai_cli.py

@@ -10,7 +10,6 @@ OpenAI CLI / Responses Normalizer (OPENAI_CLI)
 - 未识别的字段会进入 extra/raw，未知内容块保留在 internal，但默认输出阶段会丢弃。
 """
 
-
 import json
 import time
 from typing import Any
@@ -56,7 +55,7 @@ from src.core.api_format.conversion.stream_state import StreamState
 
 
 class OpenAICliNormalizer(FormatNormalizer):
-    FORMAT_ID = "OPENAI_CLI"
+    FORMAT_ID = "openai:cli"
     capabilities = FormatCapabilities(
         supports_stream=True,
         supports_error_conversion=True,
@@ -96,9 +95,7 @@ class OpenAICliNormalizer(FormatNormalizer):
         tools = self._tools_to_internal(request.get("tools"))
         tool_choice = self._tool_choice_to_internal(request.get("tool_choice"))
 
-        max_tokens = self._optional_int(
-            request.get("max_output_tokens", request.get("max_tokens"))
-        )
+        max_tokens = self._optional_int(request.get("max_output_tokens", request.get("max_tokens")))
 
         internal = InternalRequest(
             model=model,
@@ -275,7 +272,9 @@ class OpenAICliNormalizer(FormatNormalizer):
             if delta_text:
                 if not ss.get("text_block_started"):
                     ss["text_block_started"] = True
-                    events.append(ContentBlockStartEvent(block_index=0, block_type=ContentType.TEXT))
+                    events.append(
+                        ContentBlockStartEvent(block_index=0, block_type=ContentType.TEXT)
+                    )
                 events.append(ContentDeltaEvent(block_index=0, text_delta=delta_text))
             return events
 
@@ -331,11 +330,16 @@ class OpenAICliNormalizer(FormatNormalizer):
                         ss["tool_block_started"] = True
                         ss["current_tool_id"] = item.get("call_id") or item.get("id") or ""
                         ss["current_tool_name"] = item.get("name") or ""
-                        events.append(ContentBlockStartEvent(
-                            block_index=ss.get("block_index", 0),
-                            block_type=ContentType.TOOL_USE,
-                            extra={"tool_id": ss["current_tool_id"], "tool_name": ss["current_tool_name"]},
-                        ))
+                        events.append(
+                            ContentBlockStartEvent(
+                                block_index=ss.get("block_index", 0),
+                                block_type=ContentType.TOOL_USE,
+                                extra={
+                                    "tool_id": ss["current_tool_id"],
+                                    "tool_name": ss["current_tool_name"],
+                                },
+                            )
+                        )
                         ss["block_index"] = ss.get("block_index", 0) + 1
                 # message 输出项
                 elif item_type == "message":
@@ -357,11 +361,13 @@ class OpenAICliNormalizer(FormatNormalizer):
         if etype == "response.function_call_arguments.delta":
             delta = chunk.get("delta") or ""
             if delta:
-                events.append(ToolCallDeltaEvent(
-                    block_index=ss.get("block_index", 1) - 1,
-                    tool_id=ss.get("current_tool_id", ""),
-                    input_delta=delta,
-                ))
+                events.append(
+                    ToolCallDeltaEvent(
+                        block_index=ss.get("block_index", 1) - 1,
+                        tool_id=ss.get("current_tool_id", ""),
+                        input_delta=delta,
+                    )
+                )
             return events
 
         # response.function_call_arguments.done：工具调用参数完成
@@ -505,7 +511,9 @@ class OpenAICliNormalizer(FormatNormalizer):
             return resp_inner
         return response
 
-    def _extract_output_text_blocks(self, payload: dict[str, Any]) -> tuple[list[ContentBlock], dict[str, Any]]:
+    def _extract_output_text_blocks(
+        self, payload: dict[str, Any]
+    ) -> tuple[list[ContentBlock], dict[str, Any]]:
         text_parts: list[str] = []
 
         output = payload.get("output")
@@ -520,11 +528,15 @@ class OpenAICliNormalizer(FormatNormalizer):
                             if not isinstance(part, dict):
                                 continue
                             ptype = str(part.get("type") or "")
-                            if ptype in ("output_text", "text") and isinstance(part.get("text"), str):
+                            if ptype in ("output_text", "text") and isinstance(
+                                part.get("text"), str
+                            ):
                                 text_parts.append(part.get("text") or "")
                     continue
 
-                if item.get("type") in ("output_text", "text") and isinstance(item.get("text"), str):
+                if item.get("type") in ("output_text", "text") and isinstance(
+                    item.get("text"), str
+                ):
                     text_parts.append(item.get("text") or "")
 
         # 兼容：部分实现可能直接给 output_text
@@ -572,7 +584,12 @@ class OpenAICliNormalizer(FormatNormalizer):
             input_data = input_data.get("messages")
 
         if not isinstance(input_data, list):
-            return [InternalMessage(role=Role.USER, content=[UnknownBlock(raw_type="input", payload={"input": input_data})])]
+            return [
+                InternalMessage(
+                    role=Role.USER,
+                    content=[UnknownBlock(raw_type="input", payload={"input": input_data})],
+                )
+            ]
 
         messages: list[InternalMessage] = []
         for item in input_data:
@@ -585,7 +602,13 @@ class OpenAICliNormalizer(FormatNormalizer):
             if item_type == "message" or item.get("role"):
                 role = self._role_from_value(item.get("role"))
                 blocks = self._responses_content_to_blocks(item.get("content"))
-                messages.append(InternalMessage(role=role, content=blocks, extra=self._extract_extra(item, {"type", "role", "content"})))
+                messages.append(
+                    InternalMessage(
+                        role=role,
+                        content=blocks,
+                        extra=self._extract_extra(item, {"type", "role", "content"}),
+                    )
+                )
                 continue
 
             # function_call -> assistant 消息 + ToolUseBlock
@@ -594,14 +617,22 @@ class OpenAICliNormalizer(FormatNormalizer):
                 tool_name = str(item.get("name") or "")
                 args_raw = item.get("arguments") or "{}"
                 try:
-                    tool_input = json.loads(args_raw) if isinstance(args_raw, str) else (args_raw if isinstance(args_raw, dict) else {})
+                    tool_input = (
+                        json.loads(args_raw)
+                        if isinstance(args_raw, str)
+                        else (args_raw if isinstance(args_raw, dict) else {})
+                    )
                 except (json.JSONDecodeError, TypeError):
                     tool_input = {"_raw": args_raw}
                 tool_block = ToolUseBlock(
                     tool_id=tool_id,
                     tool_name=tool_name,
                     tool_input=tool_input,
-                    extra={"openai_cli": self._extract_extra(item, {"type", "call_id", "id", "name", "arguments"})},
+                    extra={
+                        "openai_cli": self._extract_extra(
+                            item, {"type", "call_id", "id", "name", "arguments"}
+                        )
+                    },
                 )
                 messages.append(InternalMessage(role=Role.ASSISTANT, content=[tool_block]))
                 continue
@@ -616,7 +647,9 @@ class OpenAICliNormalizer(FormatNormalizer):
                     tool_use_id=tool_use_id,
                     output=output,
                     content_text=content_text,
-                    extra={"openai_cli": self._extract_extra(item, {"type", "call_id", "id", "output"})},
+                    extra={
+                        "openai_cli": self._extract_extra(item, {"type", "call_id", "id", "output"})
+                    },
                 )
                 messages.append(InternalMessage(role=Role.TOOL, content=[result_block]))
                 continue
@@ -640,24 +673,30 @@ class OpenAICliNormalizer(FormatNormalizer):
                 reasoning_blocks: list[ContentBlock] = []
                 if summary_parts:
                     # 保留 reasoning 的 summary 作为 UnknownBlock，便于输出时决策
-                    reasoning_blocks.append(UnknownBlock(
-                        raw_type="reasoning",
-                        payload={"summary_text": "\n".join(summary_parts), "original": item},
-                    ))
+                    reasoning_blocks.append(
+                        UnknownBlock(
+                            raw_type="reasoning",
+                            payload={"summary_text": "\n".join(summary_parts), "original": item},
+                        )
+                    )
                 else:
                     reasoning_blocks.append(UnknownBlock(raw_type="reasoning", payload=item))
-                messages.append(InternalMessage(
-                    role=Role.ASSISTANT,
-                    content=reasoning_blocks,
-                    extra={"openai_cli": {"type": "reasoning"}},
-                ))
+                messages.append(
+                    InternalMessage(
+                        role=Role.ASSISTANT,
+                        content=reasoning_blocks,
+                        extra={"openai_cli": {"type": "reasoning"}},
+                    )
+                )
                 continue
 
             # 其他未知类型 -> 保留为 UnknownBlock
-            messages.append(InternalMessage(
-                role=Role.UNKNOWN,
-                content=[UnknownBlock(raw_type=item_type or "unknown", payload=item)],
-            ))
+            messages.append(
+                InternalMessage(
+                    role=Role.UNKNOWN,
+                    content=[UnknownBlock(raw_type=item_type or "unknown", payload=item)],
+                )
+            )
 
         return messages
 
@@ -695,20 +734,32 @@ class OpenAICliNormalizer(FormatNormalizer):
             # ToolUseBlock -> function_call
             for block in msg.content:
                 if isinstance(block, ToolUseBlock):
-                    out.append({
-                        "type": "function_call",
-                        "call_id": block.tool_id,
-                        "name": block.tool_name,
-                        "arguments": json.dumps(block.tool_input, ensure_ascii=False) if block.tool_input else "{}",
-                    })
+                    out.append(
+                        {
+                            "type": "function_call",
+                            "call_id": block.tool_id,
+                            "name": block.tool_name,
+                            "arguments": (
+                                json.dumps(block.tool_input, ensure_ascii=False)
+                                if block.tool_input
+                                else "{}"
+                            ),
+                        }
+                    )
                     continue
 
                 if isinstance(block, ToolResultBlock):
-                    out.append({
-                        "type": "function_call_output",
-                        "call_id": block.tool_use_id,
-                        "output": block.content_text if block.content_text is not None else block.output,
-                    })
+                    out.append(
+                        {
+                            "type": "function_call_output",
+                            "call_id": block.tool_use_id,
+                            "output": (
+                                block.content_text
+                                if block.content_text is not None
+                                else block.output
+                            ),
+                        }
+                    )
                     continue
 
                 # reasoning（UnknownBlock with raw_type="reasoning"）
@@ -720,10 +771,16 @@ class OpenAICliNormalizer(FormatNormalizer):
                         out.append(original)
                     else:
                         summary_text = payload.get("summary_text", "")
-                        out.append({
-                            "type": "reasoning",
-                            "summary": [{"type": "summary_text", "text": summary_text}] if summary_text else [],
-                        })
+                        out.append(
+                            {
+                                "type": "reasoning",
+                                "summary": (
+                                    [{"type": "summary_text", "text": summary_text}]
+                                    if summary_text
+                                    else []
+                                ),
+                            }
+                        )
                     continue
 
             # 普通 message（TextBlock）
@@ -763,8 +820,15 @@ class OpenAICliNormalizer(FormatNormalizer):
                     ToolDefinition(
                         name=name,
                         description=fn.get("description"),
-                        parameters=fn.get("parameters") if isinstance(fn.get("parameters"), dict) else None,
-                        extra={"openai_tool": self._extract_extra(tool, {"type", "function"}), "openai_function": self._extract_extra(fn, {"name", "description", "parameters"})},
+                        parameters=(
+                            fn.get("parameters") if isinstance(fn.get("parameters"), dict) else None
+                        ),
+                        extra={
+                            "openai_tool": self._extract_extra(tool, {"type", "function"}),
+                            "openai_function": self._extract_extra(
+                                fn, {"name", "description", "parameters"}
+                            ),
+                        },
                     )
                 )
                 continue
@@ -776,8 +840,16 @@ class OpenAICliNormalizer(FormatNormalizer):
                     ToolDefinition(
                         name=name,
                         description=tool.get("description"),
-                        parameters=tool.get("parameters") if isinstance(tool.get("parameters"), dict) else None,
-                        extra={"openai_cli": self._extract_extra(tool, {"name", "description", "parameters"})},
+                        parameters=(
+                            tool.get("parameters")
+                            if isinstance(tool.get("parameters"), dict)
+                            else None
+                        ),
+                        extra={
+                            "openai_cli": self._extract_extra(
+                                tool, {"name", "description", "parameters"}
+                            )
+                        },
                     )
                 )
         return out or None
@@ -787,16 +859,24 @@ class OpenAICliNormalizer(FormatNormalizer):
             return None
         if isinstance(tool_choice, str):
             if tool_choice == "none":
-                return ToolChoice(type=ToolChoiceType.NONE, extra={"openai_cli": {"tool_choice": tool_choice}})
+                return ToolChoice(
+                    type=ToolChoiceType.NONE, extra={"openai_cli": {"tool_choice": tool_choice}}
+                )
             if tool_choice == "auto":
-                return ToolChoice(type=ToolChoiceType.AUTO, extra={"openai_cli": {"tool_choice": tool_choice}})
+                return ToolChoice(
+                    type=ToolChoiceType.AUTO, extra={"openai_cli": {"tool_choice": tool_choice}}
+                )
             return ToolChoice(type=ToolChoiceType.AUTO, extra={"raw": tool_choice})
 
         if isinstance(tool_choice, dict):
             # OpenAI 兼容结构：{"type":"function","function":{"name":"..."}}
-            if tool_choice.get("type") == "function" and isinstance(tool_choice.get("function"), dict):
+            if tool_choice.get("type") == "function" and isinstance(
+                tool_choice.get("function"), dict
+            ):
                 name = str(tool_choice["function"].get("name") or "")
-                return ToolChoice(type=ToolChoiceType.TOOL, tool_name=name, extra={"openai_cli": tool_choice})
+                return ToolChoice(
+                    type=ToolChoiceType.TOOL, tool_name=name, extra={"openai_cli": tool_choice}
+                )
             return ToolChoice(type=ToolChoiceType.AUTO, extra={"openai_cli": tool_choice})
 
         return ToolChoice(type=ToolChoiceType.AUTO, extra={"raw": tool_choice})

src/core/api_format/conversion/registry.py

@@ -9,19 +9,17 @@ source -> internal -> target
 - 转换失败将抛出 `FormatConversionError`（不再静默回退）。
 """
 
-
 import threading
 import time
+from collections.abc import Generator
 from contextlib import contextmanager
 from typing import Any
-from collections.abc import Generator
-
-from src.core.logger import logger
-from src.core.metrics import format_conversion_duration_seconds, format_conversion_total
 
 from src.core.api_format.conversion.exceptions import FormatConversionError
 from src.core.api_format.conversion.normalizer import FormatNormalizer
 from src.core.api_format.conversion.stream_state import StreamState
+from src.core.logger import logger
+from src.core.metrics import format_conversion_duration_seconds, format_conversion_total
 
 
 @contextmanager
@@ -76,7 +74,9 @@ class FormatConversionRegistry:
         src = self._require_normalizer(source_format)
         tgt = self._require_normalizer(target_format)
 
-        with _track_conversion_metrics("request", str(source_format).upper(), str(target_format).upper()):
+        with _track_conversion_metrics(
+            "request", str(source_format).upper(), str(target_format).upper()
+        ):
             try:
                 internal = src.request_to_internal(request)
                 return tgt.request_from_internal(internal)
@@ -115,7 +115,9 @@ class FormatConversionRegistry:
         src = self._require_normalizer(source_format)
         tgt = self._require_normalizer(target_format)
 
-        with _track_conversion_metrics("response", str(source_format).upper(), str(target_format).upper()):
+        with _track_conversion_metrics(
+            "response", str(source_format).upper(), str(target_format).upper()
+        ):
             try:
                 internal = src.response_to_internal(response)
                 return tgt.response_from_internal(internal, requested_model=requested_model)
@@ -134,14 +136,19 @@ class FormatConversionRegistry:
         src = self._require_normalizer(source_format)
         tgt = self._require_normalizer(target_format)
 
-        if not (src.capabilities.supports_error_conversion and tgt.capabilities.supports_error_conversion):
+        if not (
+            src.capabilities.supports_error_conversion
+            and tgt.capabilities.supports_error_conversion
+        ):
             raise FormatConversionError(
                 source_format,
                 target_format,
                 "source/target normalizer 不支持错误转换",
             )
 
-        with _track_conversion_metrics("error", str(source_format).upper(), str(target_format).upper()):
+        with _track_conversion_metrics(
+            "error", str(source_format).upper(), str(target_format).upper()
+        ):
             try:
                 internal = src.error_to_internal(error_response)
                 return tgt.error_from_internal(internal)
@@ -179,7 +186,9 @@ class FormatConversionRegistry:
             )
             state = StreamState()
 
-        with _track_conversion_metrics("stream", str(source_format).upper(), str(target_format).upper()):
+        with _track_conversion_metrics(
+            "stream", str(source_format).upper(), str(target_format).upper()
+        ):
             try:
                 events = src.stream_chunk_to_internal(chunk, state)
                 out: list[dict[str, Any]] = []
@@ -194,7 +203,10 @@ class FormatConversionRegistry:
     def can_convert_request(self, source_format: str, target_format: str) -> bool:
         if str(source_format).upper() == str(target_format).upper():
             return True
-        return self.get_normalizer(source_format) is not None and self.get_normalizer(target_format) is not None
+        return (
+            self.get_normalizer(source_format) is not None
+            and self.get_normalizer(target_format) is not None
+        )
 
     def can_convert_response(self, source_format: str, target_format: str) -> bool:
         return self.can_convert_request(source_format, target_format)
@@ -215,15 +227,22 @@ class FormatConversionRegistry:
         tgt = self.get_normalizer(target_format)
         if src is None or tgt is None:
             return False
-        return bool(src.capabilities.supports_error_conversion and tgt.capabilities.supports_error_conversion)
+        return bool(
+            src.capabilities.supports_error_conversion
+            and tgt.capabilities.supports_error_conversion
+        )
 
-    def can_convert_full(self, format_a: str, format_b: str, *, require_stream: bool = False) -> bool:
+    def can_convert_full(
+        self, format_a: str, format_b: str, *, require_stream: bool = False
+    ) -> bool:
         if not self.can_convert_request(format_a, format_b):
             return False
         if not self.can_convert_request(format_b, format_a):
             return False
         if require_stream:
-            return self.can_convert_stream(format_a, format_b) and self.can_convert_stream(format_b, format_a)
+            return self.can_convert_stream(format_a, format_b) and self.can_convert_stream(
+                format_b, format_a
+            )
         return True
 
     def list_normalizers(self) -> list[str]:

src/core/api_format/conversion/stream_events.py

@@ -4,7 +4,6 @@
 用于把 OpenAI/Claude/Gemini 的流式协议映射为统一事件序列，再由目标格式 Normalizer 输出。
 """
 
-
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any

src/core/api_format/conversion/stream_state.py

@@ -5,7 +5,6 @@
 每个 Normalizer 通过 `substate(format_id)` 获取自己的隔离状态字典。
 """
 
-
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -46,4 +45,3 @@ class StreamState:
 __all__ = [
     "StreamState",
 ]
-

src/core/api_format/detection.py

@@ -12,63 +12,15 @@ from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from starlette.requests import Request
 
-from src.core.api_format.enums import APIFormat, AuthMethod, EndpointType
-from src.core.api_format.metadata import API_FORMAT_DEFINITIONS, ApiFormatDefinition
-
-
-def _extract_api_key_by_definition(
-    headers: dict[str, str],
-    query_params: dict[str, str] | None,
-    definition: ApiFormatDefinition,
-) -> tuple[str | None, str]:
-    """
-    根据格式定义从请求中提取 API Key
-
-    Args:
-        headers: 请求头字典（key 小写）
-        query_params: 查询参数字典（可选）
-        definition: API 格式定义
-
-    Returns:
-        (api_key, auth_method) 元组：
-        - api_key: 提取到的 API Key，或 None
-        - auth_method: 认证方式 ("header" 或 "query")
-    """
-    auth_header = definition.auth_header.lower()
-    auth_type = definition.auth_type
-
-    # Gemini 格式：query 参数优先（与 Google SDK 行为一致）
-    if definition.api_format in (APIFormat.GEMINI, APIFormat.GEMINI_CLI):
-        # 1. 优先检查 ?key= 参数
-        query_key = query_params.get("key") if query_params else None
-        if query_key:
-            return query_key, "query"
-        # 2. 再检查 x-goog-api-key 请求头
-        header_value = headers.get(auth_header)
-        if header_value:
-            return header_value, "header"
-        return None, "header"
-
-    # 其他格式：从 header 提取
-    header_value = headers.get(auth_header)
-    if not header_value:
-        return None, "header"
-
-    if auth_type == "bearer":
-        # Bearer token: "Bearer xxx"
-        if header_value.lower().startswith("bearer "):
-            return header_value[7:].strip(), "header"
-        return None, "header"
-    else:
-        # header 类型: 直接使用值
-        return header_value, "header"
+from src.core.api_format.enums import ApiFamily, AuthMethod, EndpointKind, EndpointType
+from src.core.api_format.signature import EndpointSignature, make_signature_key
 
 
 @dataclass(frozen=True)
 class RequestContext:
     """请求上下文 - 三维度信息"""
 
-    data_format: APIFormat
+    endpoint: EndpointSignature
     endpoint_type: EndpointType
     auth_method: AuthMethod
     credentials: str | None
@@ -99,18 +51,37 @@ def _detect_endpoint_type(path: str) -> EndpointType:
 
 def _detect_data_format(
     path: str, headers: dict[str, str], query_params: dict[str, str] | None
-) -> APIFormat:
+) -> EndpointSignature:
     normalized = path.lower()
+    endpoint_type = _detect_endpoint_type(path)
 
+    # Claude: /v1/messages（chat/cli 共用路径，按认证头区分）
     if normalized.startswith("/v1/messages"):
-        return APIFormat.CLAUDE
-    if normalized.startswith("/v1beta/") or normalized.startswith("/upload/v1beta/"):
-        return APIFormat.GEMINI
-    if normalized.startswith("/v1/chat/completions") or normalized.startswith("/v1/videos"):
-        return APIFormat.OPENAI
+        auth_header = headers.get("authorization", "")
+        if auth_header.lower().startswith("bearer "):
+            return EndpointSignature(api_family=ApiFamily.CLAUDE, endpoint_kind=EndpointKind.CLI)
+        return EndpointSignature(api_family=ApiFamily.CLAUDE, endpoint_kind=EndpointKind.CHAT)
 
-    api_format, _api_key, _auth_method = detect_format_from_request(headers, query_params)
-    return api_format
+    # OpenAI CLI: /responses
+    if "/responses" in normalized:
+        return EndpointSignature(api_family=ApiFamily.OPENAI, endpoint_kind=EndpointKind.CLI)
+
+    # Gemini family
+    if normalized.startswith("/v1beta/") or normalized.startswith("/upload/v1beta/"):
+        kind = EndpointKind.CHAT
+        if endpoint_type == EndpointType.VIDEO:
+            kind = EndpointKind.VIDEO
+        return EndpointSignature(api_family=ApiFamily.GEMINI, endpoint_kind=kind)
+
+    # OpenAI family
+    if normalized.startswith("/v1/videos"):
+        return EndpointSignature(api_family=ApiFamily.OPENAI, endpoint_kind=EndpointKind.VIDEO)
+    if normalized.startswith("/v1/chat/completions"):
+        return EndpointSignature(api_family=ApiFamily.OPENAI, endpoint_kind=EndpointKind.CHAT)
+
+    # Fallback: 基于认证方式猜测协议族（主要用于 /v1/models）
+    sig, _api_key, _auth_source = detect_format_from_request(headers, query_params)
+    return sig
 
 
 def _detect_auth_method(
@@ -139,7 +110,7 @@ def _detect_auth_method(
 def detect_format_from_request(
     headers: dict[str, str],
     query_params: dict[str, str] | None = None,
-) -> tuple[APIFormat, str | None, str]:
+) -> tuple[EndpointSignature, str | None, str]:
     """
     从请求头检测 API 格式和 API Key
 
@@ -153,36 +124,56 @@ def detect_format_from_request(
         query_params: 查询参数字典（可选）
 
     Returns:
-        (APIFormat, api_key, auth_method) 元组
-        - auth_method: 认证方式 ("header" 或 "query")
+        (endpoint_signature, api_key, auth_source) 元组
+        - endpoint_signature: EndpointSignature(api_family, endpoint_kind)
+        - auth_source: 认证来源 ("header" 或 "query")
     """
     # Claude: x-api-key + anthropic-version (必须同时存在)
-    claude_def = API_FORMAT_DEFINITIONS[APIFormat.CLAUDE]
-    claude_key, claude_auth_method = _extract_api_key_by_definition(
-        headers, query_params, claude_def
-    )
-    if claude_key and headers.get("anthropic-version"):
-        return APIFormat.CLAUDE, claude_key, claude_auth_method
+    if headers.get("x-api-key") and headers.get("anthropic-version"):
+        return (
+            EndpointSignature(api_family=ApiFamily.CLAUDE, endpoint_kind=EndpointKind.CHAT),
+            headers.get("x-api-key"),
+            "header",
+        )
 
-    # Gemini: x-goog-api-key (header 类型) 或 ?key=
-    gemini_def = API_FORMAT_DEFINITIONS[APIFormat.GEMINI]
-    gemini_key, gemini_auth_method = _extract_api_key_by_definition(
-        headers, query_params, gemini_def
-    )
-    if gemini_key:
-        return APIFormat.GEMINI, gemini_key, gemini_auth_method
+    # Gemini: query 参数优先（与 Google SDK 行为一致）
+    query_key = query_params.get("key") if query_params else None
+    if query_key:
+        return (
+            EndpointSignature(api_family=ApiFamily.GEMINI, endpoint_kind=EndpointKind.CHAT),
+            query_key,
+            "query",
+        )
+    x_goog_key = headers.get("x-goog-api-key")
+    if x_goog_key:
+        return (
+            EndpointSignature(api_family=ApiFamily.GEMINI, endpoint_kind=EndpointKind.CHAT),
+            x_goog_key,
+            "header",
+        )
 
     # OpenAI: Authorization: Bearer (默认)
-    # 注意: 如果只有 x-api-key 但没有 anthropic-version，也走 OpenAI 格式
-    openai_def = API_FORMAT_DEFINITIONS[APIFormat.OPENAI]
-    openai_key, openai_auth_method = _extract_api_key_by_definition(
-        headers, query_params, openai_def
+    auth_header = headers.get("authorization", "")
+    if auth_header.lower().startswith("bearer "):
+        return (
+            EndpointSignature(api_family=ApiFamily.OPENAI, endpoint_kind=EndpointKind.CHAT),
+            auth_header[7:].strip(),
+            "header",
+        )
+
+    # 兜底：兼容部分客户端用 x-api-key 携带 OpenAI token 的情况
+    if headers.get("x-api-key"):
+        return (
+            EndpointSignature(api_family=ApiFamily.OPENAI, endpoint_kind=EndpointKind.CHAT),
+            headers.get("x-api-key"),
+            "header",
+        )
+
+    return (
+        EndpointSignature(api_family=ApiFamily.OPENAI, endpoint_kind=EndpointKind.CHAT),
+        None,
+        "header",
     )
-    # 如果 OpenAI 格式没有 key，但有 x-api-key，也用它（兼容）
-    if not openai_key and claude_key:
-        openai_key = claude_key
-        openai_auth_method = claude_auth_method
-    return APIFormat.OPENAI, openai_key, openai_auth_method
 
 
 def detect_format_and_key_from_starlette(
@@ -208,8 +199,7 @@ def detect_format_and_key_from_starlette(
     api_format, api_key, auth_method = detect_format_from_request(headers, query_params)
 
     # 返回小写格式名
-    format_name = api_format.value.lower()
-    return format_name, api_key, auth_method
+    return api_format.key, api_key, auth_method
 
 
 def detect_request_context(request: Request) -> RequestContext:
@@ -227,7 +217,7 @@ def detect_request_context(request: Request) -> RequestContext:
     auth_method, credentials = _detect_auth_method(headers, query_params)
 
     return RequestContext(
-        data_format=data_format,
+        endpoint=data_format,
         endpoint_type=endpoint_type,
         auth_method=auth_method,
         credentials=credentials,
@@ -236,7 +226,7 @@ def detect_request_context(request: Request) -> RequestContext:
 
 def detect_format_from_response(
     response_data: dict,
-) -> APIFormat | None:
+) -> str | None:
     """
     从响应内容检测 API 格式
 
@@ -248,26 +238,26 @@ def detect_format_from_response(
     """
     # Claude: 有 type="message" 或特定的 content 结构
     if response_data.get("type") == "message":
-        return APIFormat.CLAUDE
+        return make_signature_key(ApiFamily.CLAUDE, EndpointKind.CHAT)
     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
+            return make_signature_key(ApiFamily.CLAUDE, EndpointKind.CHAT)
 
     # OpenAI: 有 choices 数组
     if "choices" in response_data:
-        return APIFormat.OPENAI
+        return make_signature_key(ApiFamily.OPENAI, EndpointKind.CHAT)
 
     # Gemini: 有 candidates 数组
     if "candidates" in response_data:
-        return APIFormat.GEMINI
+        return make_signature_key(ApiFamily.GEMINI, EndpointKind.CHAT)
 
     return None
 
 
 def detect_cli_format_from_path(
     path: str,
-    base_format: APIFormat,
+    base_signature: str,
 ) -> bool:
     """
     根据请求路径检测是否为 CLI 模式
@@ -285,7 +275,7 @@ def detect_cli_format_from_path(
         True 如果是 CLI 模式
     """
     # OpenAI CLI 特征: /v1/responses 路径
-    if base_format == APIFormat.OPENAI and "/responses" in path:
+    if str(base_signature).lower().startswith("openai:") and "/responses" in path.lower():
         return True
 
     # 其他 CLI 模式通常由 Adapter 层根据具体业务逻辑判断

src/core/api_format/enums.py

@@ -1,21 +1,34 @@
-"""
-API 格式枚举定义
+"""API format enums.
 
-定义所有支持的 API 格式，决定请求/响应的处理方式。
+新模式下系统使用结构化的 (ApiFamily, EndpointKind) / `family:kind` signature 作为唯一标识。
 """
 
 from enum import Enum
 
 
-class APIFormat(Enum):
-    """API 格式枚举 - 决定请求/响应的处理方式"""
+class ApiFamily(str, Enum):
+    """
+    协议族（兼容族）- 决定数据格式与认证方式的基础。
 
-    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 格式
+    注意：不叫 Provider 避免与 ORM 的 Provider 模型撞名。
+    """
+
+    OPENAI = "openai"  # openai-compatible（含 deepseek, grok, qwen 等）
+    CLAUDE = "claude"  # claude-compatible
+    GEMINI = "gemini"  # gemini-compatible
+
+
+class EndpointKind(str, Enum):
+    """
+    端点变体 - 决定 API 路径/认证变体/数据格式变体等。
+
+    注意：不复用现有 EndpointType（EndpointType 用于请求上下文检测/功能分类）。
+    """
+
+    CHAT = "chat"
+    CLI = "cli"
+    VIDEO = "video"
+    IMAGE = "image"
 
 
 class AuthMethod(str, Enum):
@@ -40,4 +53,9 @@ class EndpointType(str, Enum):
     MODELS = "models"  # Models API
 
 
-__all__ = ["APIFormat", "AuthMethod", "EndpointType"]
+__all__ = [
+    "ApiFamily",
+    "EndpointKind",
+    "AuthMethod",
+    "EndpointType",
+]

src/core/api_format/headers.py

@@ -15,9 +15,12 @@ from __future__ import annotations
 from collections.abc import Set as AbstractSet
 from typing import Any
 
-from src.core.api_format.enums import APIFormat
-from src.core.api_format.metadata import get_auth_config, get_extra_headers, get_protected_keys
-
+from src.core.api_format.metadata import (
+    get_auth_config_for_endpoint,
+    get_extra_headers_for_endpoint,
+    get_protected_keys_for_endpoint,
+)
+from src.core.api_format.signature import EndpointSignature, parse_signature_key
 
 # =============================================================================
 # 头部常量定义
@@ -118,65 +121,61 @@ def get_header_value(headers: dict[str, str], key: str, default: str = "") -> st
 # =============================================================================
 
 
-def extract_client_api_key(headers: dict[str, str], api_format: APIFormat) -> str | None:
+def extract_client_api_key_for_endpoint(
+    headers: dict[str, str],
+    endpoint: str | EndpointSignature | tuple,
+) -> str | None:
     """
-    从客户端请求头提取 API Key
-
-    自动处理大小写，根据 API 格式使用正确的认证头和类型。
+    新模式：从客户端请求头提取 API Key。
 
     Args:
         headers: 原始请求头（自动处理大小写）
-        api_format: API 格式
-
-    Returns:
-        提取的 API Key，未找到返回 None
+        endpoint: endpoint signature（`family:kind` / EndpointSignature / (ApiFamily, EndpointKind)）
     """
-
-    auth_header, auth_type = get_auth_config(api_format)
+    auth_header, auth_type = get_auth_config_for_endpoint(endpoint)
     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 value[7:]
         return None
 
-    # 直接 header 格式
     return value
 
 
-def extract_client_api_key_with_query(
+def extract_client_api_key_for_endpoint_with_query(
     headers: dict[str, str],
     query_params: dict[str, str] | None,
-    api_format: APIFormat,
+    endpoint: str | EndpointSignature | tuple,
 ) -> str | None:
     """
-    从客户端请求头或 URL 参数提取 API Key
+    新模式：从客户端请求头或 URL 参数提取 API Key。
 
-    Gemini 格式优先级（与 Google SDK 行为一致）：
+    Gemini family 优先级：
     1. URL 参数 ?key=
     2. x-goog-api-key 请求头
-
-    其他格式仅从请求头提取。
-
-    Args:
-        headers: 原始请求头（自动处理大小写）
-        query_params: URL 查询参数
-        api_format: API 格式
-
-    Returns:
-        提取的 API Key，未找到返回 None
     """
-    # Gemini 格式：query 参数优先
-    if api_format in (APIFormat.GEMINI, APIFormat.GEMINI_CLI):
+    try:
+        sig = (
+            endpoint
+            if isinstance(endpoint, EndpointSignature)
+            else (
+                parse_signature_key(endpoint)  # type: ignore[arg-type]
+                if isinstance(endpoint, str)
+                else EndpointSignature(api_family=endpoint[0], endpoint_kind=endpoint[1])
+            )  # type: ignore[index]
+        )
+    except Exception:
+        sig = None
+
+    if sig and sig.api_family.value == "gemini":
         query_key = query_params.get("key") if query_params else None
         if query_key:
             return query_key
 
-    # 其他格式或 Gemini header 方式：使用现有逻辑
-    return extract_client_api_key(headers, api_format)
+    return extract_client_api_key_for_endpoint(headers, endpoint)
 
 
 # =============================================================================
@@ -184,29 +183,33 @@ def extract_client_api_key_with_query(
 # =============================================================================
 
 
-def detect_capabilities(
+def detect_capabilities_for_endpoint(
     headers: dict[str, str],
-    api_format: APIFormat,
-    request_body: dict[str, Any] | None = None,  # noqa: ARG001 - 预留给部分格式使用
+    endpoint: str | EndpointSignature | tuple,
+    request_body: dict[str, Any] | None = None,  # noqa: ARG001 - 预留
 ) -> dict[str, bool]:
     """
-    从请求头检测能力需求
+    新模式：从请求头检测能力需求。
 
-    当前支持:
-    - Claude/Claude CLI: anthropic-beta 头中的 context-1m
-
-    Args:
-        headers: 原始请求头（自动处理大小写）
-        api_format: API 格式
-        request_body: 请求体（部分格式可能需要）
-
-    Returns:
-        能力需求字典，如 {"context_1m": True}
+    当前支持：
+    - Claude family: anthropic-beta 头中的 context-1m
     """
-
     requirements: dict[str, bool] = {}
 
-    if api_format in (APIFormat.CLAUDE, APIFormat.CLAUDE_CLI):
+    try:
+        sig = (
+            endpoint
+            if isinstance(endpoint, EndpointSignature)
+            else (
+                parse_signature_key(endpoint)  # type: ignore[arg-type]
+                if isinstance(endpoint, str)
+                else EndpointSignature(api_family=endpoint[0], endpoint_kind=endpoint[1])
+            )  # type: ignore[index]
+        )
+    except Exception:
+        sig = None
+
+    if sig and sig.api_family.value == "claude":
         beta_header = get_header_value(headers, "anthropic-beta")
         if "context-1m" in beta_header.lower():
             requirements["context_1m"] = True
@@ -242,7 +245,9 @@ class HeaderBuilder:
             self.add(k, v)
         return self
 
-    def add_protected(self, headers: dict[str, str], protected_keys: AbstractSet[str]) -> HeaderBuilder:
+    def add_protected(
+        self, headers: dict[str, str], protected_keys: AbstractSet[str]
+    ) -> HeaderBuilder:
         """
         添加头部但保护指定的 key 不被覆盖
 
@@ -310,7 +315,10 @@ class HeaderBuilder:
                 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:
+                    if (
+                        from_key.lower() not in protected_lower
+                        and to_key.lower() not in protected_lower
+                    ):
                         self.rename(from_key, to_key)
 
         return self
@@ -320,9 +328,9 @@ class HeaderBuilder:
         return {original_key: value for original_key, value in self._headers.values()}
 
 
-def build_upstream_headers(
+def build_upstream_headers_for_endpoint(
     original_headers: dict[str, str],
-    api_format: APIFormat,
+    endpoint: str | EndpointSignature | tuple,
     provider_api_key: str,
     *,
     endpoint_headers: dict[str, str] | None = None,
@@ -330,55 +338,36 @@ def build_upstream_headers(
     drop_headers: frozenset[str] | None = None,
 ) -> dict[str, str]:
     """
-    构建发送给上游 Provider 的请求头
+    新模式：构建发送给上游 Provider 的请求头（基于 endpoint signature）。
 
     优先级（后者覆盖前者）：
     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_header, auth_type = get_auth_config_for_endpoint(endpoint)
     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. 设置认证头（最高优先级，上游始终使用 header 认证）
     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"
@@ -470,39 +459,21 @@ def redact_headers_for_log(
     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,
+def build_adapter_base_headers_for_endpoint(
+    endpoint: str | EndpointSignature | tuple,
     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:
-        基础请求头字典
+    新模式：根据 endpoint signature 构建基础请求头。
     """
-    auth_header, auth_type = get_auth_config(api_format)
+    auth_header, auth_type = get_auth_config_for_endpoint(endpoint)
     auth_value = f"Bearer {api_key}" if auth_type == "bearer" else api_key
 
     headers: dict[str, str] = {
@@ -511,53 +482,33 @@ def build_adapter_base_headers(
     }
 
     if include_extra:
-        extra = get_extra_headers(api_format)
+        extra = get_extra_headers_for_endpoint(endpoint)
         if extra:
             headers.update(extra)
 
     return headers
 
 
-def build_adapter_headers(
-    api_format: APIFormat,
+def build_adapter_headers_for_endpoint(
+    endpoint: str | EndpointSignature | tuple,
     api_key: str,
     extra_headers: dict[str, str] | None = None,
 ) -> dict[str, str]:
     """
-    构建完整的 Adapter 请求头
-
-    在基础头部上合并 extra_headers，同时保护关键头部不被覆盖。
-
-    Args:
-        api_format: API 格式
-        api_key: API Key（已解密）
-        extra_headers: 调用方传入的额外头部
-
-    Returns:
-        完整的请求头字典
+    新模式：构建完整的 Adapter 请求头（包含 extra_headers）。
     """
-    base = build_adapter_base_headers(api_format, api_key)
-
+    base = build_adapter_base_headers_for_endpoint(endpoint, api_key)
     if not extra_headers:
         return base
-
-    protected = get_protected_keys(api_format)
+    protected = get_protected_keys_for_endpoint(endpoint)
     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))
+def get_adapter_protected_keys_for_endpoint(
+    endpoint: str | EndpointSignature | tuple,
+) -> tuple[str, ...]:
+    """新模式：获取 Adapter 的受保护头部 key。"""
+    return tuple(get_protected_keys_for_endpoint(endpoint))
 
 
 # =============================================================================
@@ -608,4 +559,3 @@ def get_extra_headers_from_endpoint(endpoint: Any) -> dict[str, str] | None:
     """
     header_rules = getattr(endpoint, "header_rules", None)
     return extract_set_headers_from_rules(header_rules)
-

src/core/api_format/metadata.py

@@ -1,94 +1,102 @@
 """
-API 格式元数据定义
+API endpoint metadata (new mode).
 
-集中维护 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)
+新模式下，系统以 (ApiFamily, EndpointKind) 作为结构化标识；
+在需要用 string 做 key（DB / JSON dict / metrics label / logs）时，统一使用
+`family:kind` 的 endpoint signature key（全小写）。
 """
 
+from __future__ import annotations
 
-import re
+from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import dataclass, field
-from functools import lru_cache
 from types import MappingProxyType
-from collections.abc import Iterable, Mapping, MutableMapping, Sequence
 
-from .enums import APIFormat
+from src.core.api_format.enums import ApiFamily, AuthMethod, EndpointKind
+from src.core.api_format.signature import EndpointSignature, make_signature_key, parse_signature_key
 
 
-@dataclass(frozen=True)
-class ApiFormatDefinition:
+@dataclass(frozen=True, slots=True)
+class EndpointDefinition:
     """
-    描述一个 API 格式的所有通用信息。
+    端点定义（ApiFamily + EndpointKind）。
 
-    - 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 覆盖的头部（小写）
-    - model_in_body: 是否需要在请求体中包含 model 字段（Gemini 等格式通过 URL 传递模型名）
-    - stream_in_body: 是否需要在请求体中包含 stream 字段（Gemini 等格式通过 URL 端点区分流式）
-    - data_format_id: 数据格式标识，相同 ID 的格式数据结构相同可以透传，不同则需要转换
-                      例如：CLAUDE/CLAUDE_CLI 都是 "claude"，可以透传
-                           OPENAI 是 "openai_chat"，OPENAI_CLI 是 "openai_responses"，需要转换
+    - aliases: 用于调试/展示/配置的别名（不用于“接受 legacy APIFormat”）
+    - default_path: 上游默认路径，可被 ProviderEndpoint.custom_path 覆盖
+    - auth_method/auth_header/auth_type: 认证信息（header/bearer 等）
+    - extra_headers/protected_keys: 格式固定头与保护头
+    - model_in_body/stream_in_body: 结构差异标记（用于 request/response 构造/规范化）
+    - data_format_id: 数据格式标识（相同即可透传；不同需格式转换）
     """
 
-    api_format: APIFormat
+    api_family: ApiFamily
+    endpoint_kind: EndpointKind
+
     aliases: Sequence[str] = field(default_factory=tuple)
-    default_path: str = "/"  # 上游默认请求路径
-    path_prefix: str = ""  # 本站路径前缀，为空表示无前缀
+    default_path: str = "/"
+    path_prefix: str = ""
+
+    auth_method: AuthMethod = AuthMethod.BEARER
     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（小写）
-    model_in_body: bool = True  # 是否需要在请求体中包含 model 字段
-    stream_in_body: bool = True  # 是否需要在请求体中包含 stream 字段
-    data_format_id: str = ""  # 数据格式标识，相同 ID 可透传，不同需转换
+    auth_type: str = "bearer"  # "bearer" | "header"
+
+    extra_headers: Mapping[str, str] = field(default_factory=dict)
+    protected_keys: frozenset[str] = field(default_factory=frozenset)
+
+    model_in_body: bool = True
+    stream_in_body: bool = True
+
+    data_format_id: str = ""
+
+    @property
+    def signature(self) -> EndpointSignature:
+        return EndpointSignature(api_family=self.api_family, endpoint_kind=self.endpoint_kind)
+
+    @property
+    def signature_key(self) -> str:
+        return self.signature.key
 
     def iter_aliases(self) -> Iterable[str]:
-        """返回大小写统一后的别名集合，包含枚举名本身。"""
-        yield normalize_alias_value(self.api_format.value)
+        # 统一包含 signature key（便于配置/展示）
+        yield self.signature_key
         for alias in self.aliases:
-            normalized = normalize_alias_value(alias)
-            if normalized:
-                yield normalized
+            value = str(alias or "").strip()
+            if value:
+                yield value
 
 
-_DEFINITIONS: dict[APIFormat, ApiFormatDefinition] = {
-    APIFormat.CLAUDE: ApiFormatDefinition(
-        api_format=APIFormat.CLAUDE,
+_ENDPOINT_DEFINITIONS: dict[tuple[ApiFamily, EndpointKind], EndpointDefinition] = {
+    # Claude
+    (ApiFamily.CLAUDE, EndpointKind.CHAT): EndpointDefinition(
+        api_family=ApiFamily.CLAUDE,
+        endpoint_kind=EndpointKind.CHAT,
         aliases=("claude", "anthropic", "claude_compatible"),
         default_path="/v1/messages",
-        path_prefix="",  # 通过请求头区分格式，不使用路径前缀
+        auth_method=AuthMethod.API_KEY,
         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"}),
-        data_format_id="claude",  # CLAUDE/CLAUDE_CLI 数据格式相同
+        data_format_id="claude",
     ),
-    APIFormat.CLAUDE_CLI: ApiFormatDefinition(
-        api_format=APIFormat.CLAUDE_CLI,
+    (ApiFamily.CLAUDE, EndpointKind.CLI): EndpointDefinition(
+        api_family=ApiFamily.CLAUDE,
+        endpoint_kind=EndpointKind.CLI,
         aliases=("claude_cli", "claude-cli"),
         default_path="/v1/messages",
-        path_prefix="",  # 与 CLAUDE 共享入口，通过 header 区分
+        auth_method=AuthMethod.BEARER,
         auth_header="Authorization",
         auth_type="bearer",
         protected_keys=frozenset({"authorization", "content-type"}),
-        data_format_id="claude",  # CLAUDE/CLAUDE_CLI 数据格式相同
+        data_format_id="claude",
     ),
-    APIFormat.OPENAI: ApiFormatDefinition(
-        api_format=APIFormat.OPENAI,
+    # OpenAI
+    (ApiFamily.OPENAI, EndpointKind.CHAT): EndpointDefinition(
+        api_family=ApiFamily.OPENAI,
+        endpoint_kind=EndpointKind.CHAT,
         aliases=(
             "openai",
+            "openai_compatible",
             "deepseek",
             "grok",
             "moonshot",
@@ -96,277 +104,225 @@ _DEFINITIONS: dict[APIFormat, ApiFormatDefinition] = {
             "qwen",
             "baichuan",
             "minimax",
-            "openai_compatible",
         ),
         default_path="/v1/chat/completions",
-        path_prefix="",  # 默认格式
+        auth_method=AuthMethod.BEARER,
         auth_header="Authorization",
         auth_type="bearer",
         protected_keys=frozenset({"authorization", "content-type"}),
-        data_format_id="openai_chat",  # Chat Completions API 格式
+        data_format_id="openai_chat",
     ),
-    APIFormat.OPENAI_CLI: ApiFormatDefinition(
-        api_format=APIFormat.OPENAI_CLI,
+    (ApiFamily.OPENAI, EndpointKind.CLI): EndpointDefinition(
+        api_family=ApiFamily.OPENAI,
+        endpoint_kind=EndpointKind.CLI,
         aliases=("openai_cli", "responses"),
         default_path="/responses",
-        path_prefix="",  # 与 OPENAI 共享入口
+        auth_method=AuthMethod.BEARER,
         auth_header="Authorization",
         auth_type="bearer",
         protected_keys=frozenset({"authorization", "content-type"}),
-        data_format_id="openai_responses",  # Responses API 格式，与 OPENAI 不同需转换
+        data_format_id="openai_responses",
     ),
-    APIFormat.GEMINI: ApiFormatDefinition(
-        api_format=APIFormat.GEMINI,
+    (ApiFamily.OPENAI, EndpointKind.VIDEO): EndpointDefinition(
+        api_family=ApiFamily.OPENAI,
+        endpoint_kind=EndpointKind.VIDEO,
+        aliases=("openai_video", "sora"),
+        default_path="/v1/videos",
+        auth_method=AuthMethod.BEARER,
+        auth_header="Authorization",
+        auth_type="bearer",
+        protected_keys=frozenset({"authorization", "content-type"}),
+        model_in_body=True,
+        stream_in_body=False,
+        data_format_id="openai_video",
+    ),
+    # Gemini
+    (ApiFamily.GEMINI, EndpointKind.CHAT): EndpointDefinition(
+        api_family=ApiFamily.GEMINI,
+        endpoint_kind=EndpointKind.CHAT,
         aliases=("gemini", "google", "vertex"),
         default_path="/v1beta/models/{model}:{action}",
-        path_prefix="",  # 通过请求头区分格式
+        auth_method=AuthMethod.GOOG_API_KEY,
         auth_header="x-goog-api-key",
         auth_type="header",
         protected_keys=frozenset({"x-goog-api-key", "content-type"}),
-        model_in_body=False,  # Gemini 通过 URL 路径传递模型名
-        stream_in_body=False,  # Gemini 通过 URL 端点区分流式（streamGenerateContent vs generateContent）
-        data_format_id="gemini",  # GEMINI/GEMINI_CLI 数据格式相同
+        model_in_body=False,
+        stream_in_body=False,
+        data_format_id="gemini",
     ),
-    APIFormat.GEMINI_CLI: ApiFormatDefinition(
-        api_format=APIFormat.GEMINI_CLI,
+    (ApiFamily.GEMINI, EndpointKind.CLI): EndpointDefinition(
+        api_family=ApiFamily.GEMINI,
+        endpoint_kind=EndpointKind.CLI,
         aliases=("gemini_cli", "gemini-cli"),
         default_path="/v1beta/models/{model}:{action}",
-        path_prefix="",  # 与 GEMINI 共享入口
+        auth_method=AuthMethod.GOOG_API_KEY,
         auth_header="x-goog-api-key",
         auth_type="header",
         protected_keys=frozenset({"x-goog-api-key", "content-type"}),
-        model_in_body=False,  # Gemini 通过 URL 路径传递模型名
-        stream_in_body=False,  # Gemini 通过 URL 端点区分流式
-        data_format_id="gemini",  # GEMINI/GEMINI_CLI 数据格式相同
+        model_in_body=False,
+        stream_in_body=False,
+        data_format_id="gemini",
+    ),
+    (ApiFamily.GEMINI, EndpointKind.VIDEO): EndpointDefinition(
+        api_family=ApiFamily.GEMINI,
+        endpoint_kind=EndpointKind.VIDEO,
+        aliases=("gemini_video", "veo"),
+        default_path="/v1beta/models/{model}:predictLongRunning",
+        auth_method=AuthMethod.GOOG_API_KEY,
+        auth_header="x-goog-api-key",
+        auth_type="header",
+        protected_keys=frozenset({"x-goog-api-key", "content-type"}),
+        model_in_body=False,
+        stream_in_body=False,
+        data_format_id="gemini_video",
     ),
 }
 
 # 对外只暴露只读视图，避免被随意修改
-API_FORMAT_DEFINITIONS: Mapping[APIFormat, ApiFormatDefinition] = MappingProxyType(_DEFINITIONS)
+ENDPOINT_DEFINITIONS: Mapping[tuple[ApiFamily, EndpointKind], EndpointDefinition] = (
+    MappingProxyType(_ENDPOINT_DEFINITIONS)
+)
 
 
-def get_api_format_definition(api_format: APIFormat) -> ApiFormatDefinition:
-    """获取指定格式的定义，不存在时抛出 KeyError。"""
-    return API_FORMAT_DEFINITIONS[api_format]
+def list_endpoint_definitions() -> list[EndpointDefinition]:
+    return list(ENDPOINT_DEFINITIONS.values())
 
 
-def list_api_format_definitions() -> list[ApiFormatDefinition]:
-    """返回所有定义的浅拷贝列表，供遍历使用。"""
-    return list(API_FORMAT_DEFINITIONS.values())
+def get_endpoint_definition(
+    api_family: ApiFamily, endpoint_kind: EndpointKind
+) -> EndpointDefinition:
+    return ENDPOINT_DEFINITIONS[(api_family, endpoint_kind)]
 
 
-def build_alias_lookup() -> dict[str, APIFormat]:
+def resolve_endpoint_definition(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+) -> EndpointDefinition | None:
     """
-    构建 alias -> APIFormat 的查找表。
-    每次调用都会返回新的 dict，避免可变全局引发并发问题。
+    Resolve an endpoint definition from a signature-like input.
+
+    Accepted inputs:
+    - EndpointSignature
+    - (ApiFamily, EndpointKind)
+    - "family:kind" signature string
     """
-    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)
+    try:
+        if isinstance(value, EndpointSignature):
+            return ENDPOINT_DEFINITIONS.get((value.api_family, value.endpoint_kind))
+        if isinstance(value, tuple) and len(value) == 2:
+            fam, kind = value
+            if isinstance(fam, ApiFamily) and isinstance(kind, EndpointKind):
+                return ENDPOINT_DEFINITIONS.get((fam, kind))
+        if isinstance(value, str):
+            sig = parse_signature_key(value)
+            return ENDPOINT_DEFINITIONS.get((sig.api_family, sig.endpoint_kind))
+    except Exception:
+        return None
+    return None
 
 
-def get_default_path(api_format: APIFormat) -> str:
-    """
-    获取该格式的上游默认请求路径。
-
-    可通过 Endpoint.custom_path 覆盖。
-    """
-    definition = API_FORMAT_DEFINITIONS.get(api_format)
+def get_default_path_for_endpoint(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+) -> str:
+    definition = resolve_endpoint_definition(value)
     return definition.default_path if definition else "/"
 
 
-def get_local_path(api_format: APIFormat) -> str:
+def get_local_path_for_endpoint(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+) -> str:
+    definition = resolve_endpoint_definition(value)
+    if not definition:
+        return "/"
+    prefix = definition.path_prefix or ""
+    return prefix + definition.default_path
+
+
+def get_auth_config_for_endpoint(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+) -> tuple[str, str]:
+    definition = resolve_endpoint_definition(value)
+    if not definition:
+        return "Authorization", "bearer"
+    return definition.auth_header, definition.auth_type
+
+
+def get_extra_headers_for_endpoint(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+) -> Mapping[str, str]:
+    definition = resolve_endpoint_definition(value)
+    return definition.extra_headers if definition else {}
+
+
+def get_protected_keys_for_endpoint(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+) -> frozenset[str]:
+    definition = resolve_endpoint_definition(value)
+    return (
+        definition.protected_keys
+        if definition and definition.protected_keys
+        else frozenset({"authorization", "content-type"})
+    )
+
+
+def get_data_format_id_for_endpoint(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+) -> str:
     """
-    获取该格式的本站入口路径。
+    获取端点的数据格式标识。
 
-    本站入口路径 = path_prefix + default_path
-    例如：path_prefix="/openai" + default_path="/v1/chat/completions" -> "/openai/v1/chat/completions"
+    - 相同 data_format_id 可透传（不需要数据转换）
+    - 不同 data_format_id 需要走 format conversion
     """
-    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"})
-
-
-def get_data_format_id(api_format: str | APIFormat) -> str:
-    """
-    获取格式的数据格式标识。
-
-    相同 data_format_id 的格式数据结构相同，可以透传；不同则需要转换。
-
-    Args:
-        api_format: API 格式（字符串或枚举）
-
-    Returns:
-        数据格式标识，未找到时返回格式名称本身（小写）
-    """
-    # 统一转换为 APIFormat 枚举
-    if isinstance(api_format, str):
-        resolved = resolve_api_format(api_format)
-        if resolved is None:
-            # 未知格式：返回小写，与已定义格式的 data_format_id 风格一致
-            return api_format.lower()
-        api_format = resolved
-
-    definition = API_FORMAT_DEFINITIONS.get(api_format)
+    definition = resolve_endpoint_definition(value)
     if definition and definition.data_format_id:
         return definition.data_format_id
-    # 兜底：返回格式名称本身（小写）
-    return api_format.value.lower()
+    return ""
 
 
-def can_passthrough(client_format: str | APIFormat, endpoint_format: str | APIFormat) -> bool:
+def can_passthrough_endpoint(
+    client: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+    provider: str | EndpointSignature | tuple[ApiFamily, EndpointKind],
+) -> bool:
     """
-    判断两个格式之间是否可以透传（不需要数据转换）。
+    判断两个 endpoint signature 是否可以透传（无需数据转换）。
 
     透传条件：
-    1. 格式完全相同
-    2. data_format_id 相同（如 CLAUDE 和 CLAUDE_CLI 都是 "claude"）
-
-    Args:
-        client_format: 客户端请求格式
-        endpoint_format: 端点 API 格式
-
-    Returns:
-        True 表示可以透传，False 表示需要转换
+    1) signature 完全相同
+    2) data_format_id 相同（如 claude:chat / claude:cli）
     """
-    # 统一转换为字符串比较
-    client_str = client_format.value if isinstance(client_format, APIFormat) else str(client_format).upper()
-    endpoint_str = endpoint_format.value if isinstance(endpoint_format, APIFormat) else str(endpoint_format).upper()
+    try:
+        if isinstance(client, str) and isinstance(provider, str):
+            if parse_signature_key(client).key == parse_signature_key(provider).key:
+                return True
+    except Exception:
+        pass
 
-    # 完全相同
-    if client_str == endpoint_str:
-        return True
-
-    # 检查 data_format_id
-    client_data_id = get_data_format_id(client_format)
-    endpoint_data_id = get_data_format_id(endpoint_format)
-    return client_data_id == endpoint_data_id
+    client_id = get_data_format_id_for_endpoint(client)
+    provider_id = get_data_format_id_for_endpoint(provider)
+    return bool(client_id) and client_id == provider_id
 
 
-@lru_cache(maxsize=1)
-def _alias_lookup_cache() -> dict[str, APIFormat]:
-    """缓存 alias -> APIFormat 查找表，减少重复构建。"""
-    return build_alias_lookup()
-
-
-def resolve_api_format_alias(value: str) -> APIFormat | None:
-    """根据别名查找 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: str | APIFormat | None,
-    default: APIFormat | None = None,
-) -> APIFormat | None:
+def make_endpoint_signature(api_family: str, endpoint_kind: str) -> str:
     """
-    将任意字符串/枚举值解析为 APIFormat。
+    Helper: build canonical signature key from raw strings (lowercased/trimmed).
 
-    Args:
-        value: 可以是 APIFormat 或任意字符串/别名
-        default: 未解析成功时返回的默认值
+    This is used in places that store family/kind separately in DB.
     """
-    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
+    return make_signature_key(api_family, endpoint_kind)
 
 
-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
+__all__ = [
+    "EndpointDefinition",
+    "ENDPOINT_DEFINITIONS",
+    "list_endpoint_definitions",
+    "get_endpoint_definition",
+    "resolve_endpoint_definition",
+    "get_default_path_for_endpoint",
+    "get_local_path_for_endpoint",
+    "get_auth_config_for_endpoint",
+    "get_extra_headers_for_endpoint",
+    "get_protected_keys_for_endpoint",
+    "get_data_format_id_for_endpoint",
+    "can_passthrough_endpoint",
+    "make_endpoint_signature",
+]

src/core/api_format/signature.py

@@ -0,0 +1,84 @@
+"""
+Endpoint signature utilities.
+
+新模式下，系统以 (ApiFamily, EndpointKind) 作为结构化标识；
+在需要用 string 做 key（JSON dict / metrics label / logs）时，统一使用 `family:kind` 的 signature key。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from src.core.api_format.enums import ApiFamily, EndpointKind
+
+
+@dataclass(frozen=True, slots=True)
+class EndpointSignature:
+    api_family: ApiFamily
+    endpoint_kind: EndpointKind
+
+    @property
+    def key(self) -> str:
+        return make_signature_key(self.api_family, self.endpoint_kind)
+
+
+def make_signature_key(api_family: ApiFamily | str, endpoint_kind: EndpointKind | str) -> str:
+    fam = api_family.value if isinstance(api_family, ApiFamily) else str(api_family).strip().lower()
+    kind = (
+        endpoint_kind.value
+        if isinstance(endpoint_kind, EndpointKind)
+        else str(endpoint_kind).strip().lower()
+    )
+    return f"{fam}:{kind}"
+
+
+def parse_signature_key(value: str) -> EndpointSignature:
+    """
+    Parse a signature key into structured enums.
+
+    Canonical form: `<api_family>:<endpoint_kind>`, both lowercase.
+    """
+    raw = str(value).strip()
+    if not raw or ":" not in raw:
+        raise ValueError(f"Invalid endpoint signature: {value!r}")
+    fam_raw, kind_raw = raw.split(":", 1)
+    fam = ApiFamily(fam_raw.strip().lower())
+    kind = EndpointKind(kind_raw.strip().lower())
+    return EndpointSignature(api_family=fam, endpoint_kind=kind)
+
+
+def normalize_signature_key(value: str) -> str:
+    """Normalize signature key (case/whitespace) to canonical lowercase `family:kind`."""
+    sig = parse_signature_key(value)
+    return sig.key
+
+
+def normalize_endpoint_signature(
+    value: str | EndpointSignature | tuple[ApiFamily, EndpointKind] | None,
+    *,
+    default: EndpointSignature | None = None,
+) -> EndpointSignature | None:
+    if value is None:
+        return default
+    if isinstance(value, EndpointSignature):
+        return value
+    if isinstance(value, tuple) and len(value) == 2:
+        fam, kind = value
+        if isinstance(fam, ApiFamily) and isinstance(kind, EndpointKind):
+            return EndpointSignature(api_family=fam, endpoint_kind=kind)
+        return default
+    if isinstance(value, str):
+        try:
+            return parse_signature_key(value)
+        except Exception:
+            return default
+    return default
+
+
+__all__ = [
+    "EndpointSignature",
+    "make_signature_key",
+    "parse_signature_key",
+    "normalize_signature_key",
+    "normalize_endpoint_signature",
+]

src/core/api_format/utils.py

@@ -6,43 +6,34 @@ API 格式工具函数
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
 
-if TYPE_CHECKING:
-    from src.core.api_format.enums import APIFormat
-
-
-def is_cli_format(format_id: str | APIFormat | None) -> bool:
+def is_cli_format(format_id: str | None) -> bool:
     """
     判断是否为 CLI 透传格式
 
-    CLI 格式以 _CLI 结尾，表示该入口更偏向“CLI 兼容层”（鉴权/UA/路径差异等）。
-    是否参与格式转换由转换层决定；当前项目已支持 CLI 格式参与转换。
+    新模式下使用 endpoint signature：`family:kind`，CLI 的 kind 为 `cli`。
 
     Args:
-        format_id: 格式标识符（字符串或 APIFormat 枚举）
+        format_id: endpoint signature key（如 "openai:cli"）
 
     Returns:
         True 如果是 CLI 格式
 
     Examples:
-        >>> is_cli_format("CLAUDE_CLI")
+        >>> is_cli_format("claude:cli")
         True
-        >>> is_cli_format("CLAUDE")
+        >>> is_cli_format("claude:chat")
         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")
+    text = str(format_id).strip()
+    return text.lower().endswith(":cli")
 
 
-def get_base_format(format_id: str | APIFormat | None) -> str | None:
+def get_base_format(format_id: str | None) -> str | None:
     """
-    获取基础格式（去除 _CLI 后缀）
+    获取基础格式（CLI -> CHAT）
 
     Args:
         format_id: 格式标识符
@@ -51,41 +42,55 @@ def get_base_format(format_id: str | APIFormat | None) -> str | None:
         基础格式字符串，或 None
 
     Examples:
-        >>> get_base_format("CLAUDE_CLI")
-        "CLAUDE"
-        >>> get_base_format("OPENAI")
-        "OPENAI"
+        >>> get_base_format("claude:cli")
+        "claude:chat"
+        >>> get_base_format("openai:chat")
+        "openai:chat"
     """
     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
+    text = str(format_id).strip()
+    if not text:
+        return None
+
+    from src.core.api_format.enums import EndpointKind
+    from src.core.api_format.signature import make_signature_key, parse_signature_key
+
+    try:
+        sig = parse_signature_key(text)
+    except Exception:
+        return None
+    if sig.endpoint_kind == EndpointKind.CLI:
+        return make_signature_key(sig.api_family, EndpointKind.CHAT)
+    return make_signature_key(sig.api_family, sig.endpoint_kind)
 
 
-def normalize_format(format_id: str | APIFormat | None) -> str | None:
+def normalize_format(format_id: str | None) -> str | None:
     """
-    规范化格式标识符
+    规范化 endpoint signature key（canonical: 全小写 `family:kind`）。
 
     Args:
-        format_id: 格式标识符（可能是字符串、枚举或 None）
+        format_id: endpoint signature key
 
     Returns:
-        大写的格式字符串，或 None
+        canonical signature key，或 None
     """
     if format_id is None:
         return None
-    if hasattr(format_id, "value"):
-        return str(format_id.value).upper()
-    return str(format_id).upper()
+    text = str(format_id).strip()
+    if not text:
+        return None
+    from src.core.api_format.signature import normalize_signature_key
+
+    try:
+        return normalize_signature_key(text)
+    except Exception:
+        return None
 
 
 def is_same_format(
-    format1: str | APIFormat | None,
-    format2: str | APIFormat | None,
+    format1: str | None,
+    format2: str | None,
 ) -> bool:
     """
     判断两个格式是否相同
@@ -95,14 +100,13 @@ def is_same_format(
     return normalize_format(format1) == normalize_format(format2)
 
 
-def is_convertible_format(format_id: str | APIFormat | None) -> bool:
+def is_convertible_format(format_id: str | None) -> bool:
     """
     判断是否为可转换格式
 
     .. deprecated::
         此函数语义已退化（对非 None 输入总返回 True）。
         真正的可转换性应通过 format_conversion_registry.can_convert_*() 查询。
-        保留此函数仅为向后兼容，不建议新代码使用。
     """
     if format_id is None:
         return False

tests/core/api_format/conversion/test_claude_normalizer.py

@@ -262,7 +262,9 @@ def test_claude_stream_chunk_and_event_roundtrip_basic() -> None:
     assert any(isinstance(e, MessageStartEvent) for e in events)
     assert [e.text_delta for e in events if isinstance(e, ContentDeltaEvent)] == ["Hel", "lo"]
     assert any(isinstance(e, ToolCallDeltaEvent) and e.tool_id == "toolu_1" for e in events)
-    assert any(isinstance(e, MessageStopEvent) and e.stop_reason == StopReason.END_TURN for e in events)
+    assert any(
+        isinstance(e, MessageStopEvent) and e.stop_reason == StopReason.END_TURN for e in events
+    )
 
     # internal events -> Claude events
     state2 = StreamState()
@@ -273,7 +275,11 @@ def test_claude_stream_chunk_and_event_roundtrip_basic() -> None:
     assert out_events[0]["type"] == "message_start"
     assert out_events[0]["message"]["id"] == "msg_1"
 
-    assert any(ev.get("type") == "content_block_delta" and ev.get("delta", {}).get("type") == "input_json_delta" for ev in out_events)
+    assert any(
+        ev.get("type") == "content_block_delta"
+        and ev.get("delta", {}).get("type") == "input_json_delta"
+        for ev in out_events
+    )
     assert out_events[-1]["type"] == "message_stop"
 
 
@@ -305,9 +311,7 @@ def test_claude_request_metadata_preserved() -> None:
             {"type": "text", "text": "System prompt 1"},
             {"type": "text", "text": "System prompt 2"},
         ],
-        "metadata": {
-            "user_id": "user_abc123_session_xyz456"
-        },
+        "metadata": {"user_id": "user_abc123_session_xyz456"},
         "max_tokens": 32000,
         "stream": True,
     }

tests/core/api_format/conversion/test_cli_conversion.py

@@ -33,9 +33,9 @@ def _make_registry_with_cli() -> FormatConversionRegistry:
 
 def test_registry_can_convert_full_with_cli_stream() -> None:
     reg = _make_registry_with_cli()
-    assert reg.can_convert_full("OPENAI_CLI", "OPENAI", require_stream=True) is True
-    assert reg.can_convert_full("OPENAI_CLI", "CLAUDE_CLI", require_stream=True) is True
-    assert reg.can_convert_full("GEMINI_CLI", "CLAUDE", require_stream=True) is True
+    assert reg.can_convert_full("openai:cli", "openai:chat", require_stream=True) is True
+    assert reg.can_convert_full("openai:cli", "claude:cli", require_stream=True) is True
+    assert reg.can_convert_full("gemini:cli", "claude:chat", require_stream=True) is True
 
 
 def test_openai_cli_request_to_claude() -> None:
@@ -48,7 +48,7 @@ def test_openai_cli_request_to_claude() -> None:
         "max_output_tokens": 12,
     }
 
-    claude_req = reg.convert_request(openai_cli_req, "OPENAI_CLI", "CLAUDE")
+    claude_req = reg.convert_request(openai_cli_req, "openai:cli", "claude:chat")
     assert claude_req["model"] == "gpt-4o-mini"
     assert claude_req["stream"] is True
     assert isinstance(claude_req.get("messages"), list)
@@ -69,7 +69,7 @@ def test_claude_response_to_openai_cli() -> None:
         "usage": {"input_tokens": 5, "output_tokens": 7},
     }
 
-    openai_cli_resp = reg.convert_response(claude_resp, "CLAUDE", "OPENAI_CLI")
+    openai_cli_resp = reg.convert_response(claude_resp, "claude:chat", "openai:cli")
     assert openai_cli_resp["object"] == "response"
     assert isinstance(openai_cli_resp.get("output"), list)
     msg = cast(dict[str, Any], openai_cli_resp["output"][0])
@@ -89,10 +89,12 @@ def test_stream_openai_to_openai_cli_delta() -> None:
         "object": "chat.completion.chunk",
         "created": 1,
         "model": "gpt-4o-mini",
-        "choices": [{"index": 0, "delta": {"role": "assistant", "content": "hi"}, "finish_reason": None}],
+        "choices": [
+            {"index": 0, "delta": {"role": "assistant", "content": "hi"}, "finish_reason": None}
+        ],
     }
 
-    out_events = reg.convert_stream_chunk(chunk, "OPENAI", "OPENAI_CLI", state=state)
+    out_events = reg.convert_stream_chunk(chunk, "openai:chat", "openai:cli", state=state)
     assert isinstance(out_events, list) and out_events
     assert out_events[0].get("type") == "response.created"
     assert out_events[1].get("type") == "response.output_text.delta"
@@ -109,7 +111,7 @@ def test_stream_openai_cli_to_openai_delta() -> None:
         "response": {"id": "resp_1", "model": "gpt-4o-mini"},
     }
 
-    out_events = reg.convert_stream_chunk(chunk, "OPENAI_CLI", "OPENAI", state=state)
+    out_events = reg.convert_stream_chunk(chunk, "openai:cli", "openai:chat", state=state)
     assert isinstance(out_events, list) and out_events
 
     # 第一个 chunk 先补齐 assistant role
@@ -149,7 +151,7 @@ def test_openai_cli_function_call_to_claude() -> None:
         "stream": True,
     }
 
-    claude_req = reg.convert_request(openai_cli_req, "OPENAI_CLI", "CLAUDE")
+    claude_req = reg.convert_request(openai_cli_req, "openai:cli", "claude:chat")
 
     messages = claude_req.get("messages", [])
     assert len(messages) == 3
@@ -203,14 +205,16 @@ def test_openai_cli_reasoning_preserved_in_roundtrip() -> None:
     }
 
     # 转换到 internal 再转回 OPENAI_CLI
-    converted = reg.convert_request(openai_cli_req, "OPENAI_CLI", "OPENAI_CLI")
+    converted = reg.convert_request(openai_cli_req, "openai:cli", "openai:cli")
 
     input_items = converted.get("input", [])
     # 应该有 user message, reasoning, assistant message
     assert len(input_items) >= 2
 
     # 找到 reasoning block
-    reasoning_items = [i for i in input_items if isinstance(i, dict) and i.get("type") == "reasoning"]
+    reasoning_items = [
+        i for i in input_items if isinstance(i, dict) and i.get("type") == "reasoning"
+    ]
     assert len(reasoning_items) == 1
     assert "summary" in reasoning_items[0]
 
@@ -247,7 +251,7 @@ def test_claude_tool_use_to_openai_cli() -> None:
         ],
     }
 
-    openai_cli_req = reg.convert_request(claude_req, "CLAUDE", "OPENAI_CLI")
+    openai_cli_req = reg.convert_request(claude_req, "claude:chat", "openai:cli")
 
     input_items = openai_cli_req.get("input", [])
     assert len(input_items) >= 3
@@ -259,7 +263,9 @@ def test_claude_tool_use_to_openai_cli() -> None:
     assert fc_items[0]["call_id"] == "tool_123"
 
     # 找到 function_call_output
-    fco_items = [i for i in input_items if isinstance(i, dict) and i.get("type") == "function_call_output"]
+    fco_items = [
+        i for i in input_items if isinstance(i, dict) and i.get("type") == "function_call_output"
+    ]
     assert len(fco_items) == 1
     assert fco_items[0]["call_id"] == "tool_123"
     assert fco_items[0]["output"] == "Hello World"
@@ -281,7 +287,7 @@ def test_stream_openai_cli_in_progress_event() -> None:
         },
     }
 
-    events1 = reg.convert_stream_chunk(created_chunk, "OPENAI_CLI", "CLAUDE", state=state)
+    events1 = reg.convert_stream_chunk(created_chunk, "openai:cli", "claude:chat", state=state)
     assert isinstance(events1, list) and events1
     assert events1[0].get("type") == "message_start"
 
@@ -296,7 +302,7 @@ def test_stream_openai_cli_in_progress_event() -> None:
         },
     }
 
-    events2 = reg.convert_stream_chunk(in_progress_chunk, "OPENAI_CLI", "CLAUDE", state=state)
+    events2 = reg.convert_stream_chunk(in_progress_chunk, "openai:cli", "claude:chat", state=state)
     # response.in_progress 不应产生任何事件
     assert events2 == []
 
@@ -311,7 +317,7 @@ def test_stream_openai_cli_function_call_events() -> None:
         "type": "response.created",
         "response": {"id": "resp_456", "model": "gpt-5"},
     }
-    reg.convert_stream_chunk(created_chunk, "OPENAI_CLI", "CLAUDE", state=state)
+    reg.convert_stream_chunk(created_chunk, "openai:cli", "claude:chat", state=state)
 
     # response.output_item.added (function_call)
     output_item_chunk = {
@@ -323,7 +329,7 @@ def test_stream_openai_cli_function_call_events() -> None:
         },
     }
 
-    events1 = reg.convert_stream_chunk(output_item_chunk, "OPENAI_CLI", "CLAUDE", state=state)
+    events1 = reg.convert_stream_chunk(output_item_chunk, "openai:cli", "claude:chat", state=state)
     assert isinstance(events1, list) and events1
     assert events1[0].get("type") == "content_block_start"
 
@@ -333,7 +339,7 @@ def test_stream_openai_cli_function_call_events() -> None:
         "delta": '{"city":',
     }
 
-    events2 = reg.convert_stream_chunk(args_delta_chunk, "OPENAI_CLI", "CLAUDE", state=state)
+    events2 = reg.convert_stream_chunk(args_delta_chunk, "openai:cli", "claude:chat", state=state)
     assert isinstance(events2, list) and events2
     # ToolCallDeltaEvent 转换为 Claude 的 content_block_delta
     assert events2[0].get("type") == "content_block_delta"
@@ -352,7 +358,7 @@ def test_stream_openai_cli_function_call_events() -> None:
         },
     }
 
-    events3 = reg.convert_stream_chunk(output_done_chunk, "OPENAI_CLI", "CLAUDE", state=state)
+    events3 = reg.convert_stream_chunk(output_done_chunk, "openai:cli", "claude:chat", state=state)
     assert isinstance(events3, list) and events3
     assert events3[0].get("type") == "content_block_stop"
 
@@ -486,7 +492,7 @@ def test_real_claude_cli_stream_response_conversion() -> None:
     # 收集所有转换后的 OpenAI 格式事件
     all_openai_events: list[dict[str, Any]] = []
     for chunk in chunks:
-        events = reg.convert_stream_chunk(chunk, "CLAUDE_CLI", "OPENAI", state=state)
+        events = reg.convert_stream_chunk(chunk, "claude:cli", "openai:chat", state=state)
         all_openai_events.extend(events)
 
     # 验证转换结果
@@ -542,8 +548,16 @@ def test_real_claude_cli_stream_to_openai_cli() -> None:
         },
         {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
         {"type": "ping"},
-        {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}},
-        {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": " World"}},
+        {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "text_delta", "text": "Hello"},
+        },
+        {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "text_delta", "text": " World"},
+        },
         {"type": "content_block_stop", "index": 0},
         {
             "type": "message_delta",
@@ -555,7 +569,7 @@ def test_real_claude_cli_stream_to_openai_cli() -> None:
 
     all_events: list[dict[str, Any]] = []
     for chunk in chunks:
-        events = reg.convert_stream_chunk(chunk, "CLAUDE_CLI", "OPENAI_CLI", state=state)
+        events = reg.convert_stream_chunk(chunk, "claude:cli", "openai:cli", state=state)
         all_events.extend(events)
 
     # 验证 OpenAI CLI 格式事件
@@ -573,5 +587,7 @@ def test_real_claude_cli_stream_to_openai_cli() -> None:
     assert " World" in deltas
 
     # 应该有 response.completed 或 response.done 事件
-    done_events = [e for e in all_events if e.get("type") in ("response.completed", "response.done")]
+    done_events = [
+        e for e in all_events if e.get("type") in ("response.completed", "response.done")
+    ]
     assert len(done_events) >= 1

tests/core/api_format/conversion/test_compatibility.py

@@ -16,8 +16,8 @@ from src.core.api_format.conversion.compatibility import is_format_compatible
 
 def test_same_format_is_compatible() -> None:
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "CLAUDE",
+        "claude:chat",
+        "claude:chat",
         endpoint_format_acceptance_config=None,
         is_stream=False,
         global_conversion_enabled=False,
@@ -33,8 +33,8 @@ def test_cli_format_convertible_when_converter_supports_full() -> None:
     registry.can_convert_full.return_value = True
 
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE_CLI",
-        "OPENAI",
+        "claude:cli",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=True,
@@ -48,8 +48,8 @@ def test_cli_format_convertible_when_converter_supports_full() -> None:
 def test_global_switch_disabled_blocks_conversion() -> None:
     """全局开关关闭时（环境变量 FORMAT_CONVERSION_ENABLED=false）阻止转换"""
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "OPENAI",
+        "claude:chat",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=False,
@@ -62,8 +62,8 @@ def test_global_switch_disabled_blocks_conversion() -> None:
 
 def test_endpoint_config_none_blocks_conversion() -> None:
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "OPENAI",
+        "claude:chat",
+        "openai:chat",
         endpoint_format_acceptance_config=None,
         is_stream=False,
         global_conversion_enabled=True,
@@ -76,8 +76,8 @@ def test_endpoint_config_none_blocks_conversion() -> None:
 
 def test_endpoint_disabled_blocks_conversion() -> None:
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "OPENAI",
+        "claude:chat",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": False},
         is_stream=False,
         global_conversion_enabled=True,
@@ -90,9 +90,9 @@ def test_endpoint_disabled_blocks_conversion() -> None:
 
 def test_accept_formats_allows_only_whitelist() -> None:
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "OPENAI",
-        endpoint_format_acceptance_config={"enabled": True, "accept_formats": ["OPENAI"]},
+        "claude:chat",
+        "openai:chat",
+        endpoint_format_acceptance_config={"enabled": True, "accept_formats": ["openai:chat"]},
         is_stream=False,
         global_conversion_enabled=True,
         registry=MagicMock(),
@@ -104,9 +104,9 @@ def test_accept_formats_allows_only_whitelist() -> None:
 
 def test_reject_formats_blocks_blacklist() -> None:
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "OPENAI",
-        endpoint_format_acceptance_config={"enabled": True, "reject_formats": ["CLAUDE"]},
+        "claude:chat",
+        "openai:chat",
+        endpoint_format_acceptance_config={"enabled": True, "reject_formats": ["claude:chat"]},
         is_stream=False,
         global_conversion_enabled=True,
         registry=MagicMock(),
@@ -118,8 +118,8 @@ def test_reject_formats_blocks_blacklist() -> None:
 
 def test_stream_conversion_disabled_blocks_stream() -> None:
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "OPENAI",
+        "claude:chat",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": True, "stream_conversion": False},
         is_stream=True,
         global_conversion_enabled=True,
@@ -135,8 +135,8 @@ def test_converter_support_required() -> None:
     registry.can_convert_full.return_value = False
 
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "OPENAI",
+        "claude:chat",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=True,
@@ -152,9 +152,9 @@ def test_conversion_allowed_when_converter_supports_full() -> None:
     registry.can_convert_full.return_value = True
 
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "OPENAI",
-        endpoint_format_acceptance_config={"enabled": True, "accept_formats": ["CLAUDE"]},
+        "claude:chat",
+        "openai:chat",
+        endpoint_format_acceptance_config={"enabled": True, "accept_formats": ["claude:chat"]},
         is_stream=False,
         global_conversion_enabled=True,
         registry=registry,
@@ -170,8 +170,8 @@ def test_conversion_allowed_when_converter_supports_full() -> None:
 def test_claude_cli_to_claude_no_conversion_needed() -> None:
     """CLAUDE 和 CLAUDE_CLI 格式相同，只是认证不同，可透传（需开关启用）"""
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE_CLI",
-        "CLAUDE",
+        "claude:cli",
+        "claude:chat",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=True,
@@ -185,8 +185,8 @@ def test_claude_cli_to_claude_no_conversion_needed() -> None:
 def test_claude_to_claude_cli_no_conversion_needed() -> None:
     """CLAUDE 和 CLAUDE_CLI 格式相同，只是认证不同，可透传（需开关启用）"""
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE",
-        "CLAUDE_CLI",
+        "claude:chat",
+        "claude:cli",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=True,
@@ -200,8 +200,8 @@ def test_claude_to_claude_cli_no_conversion_needed() -> None:
 def test_gemini_cli_to_gemini_no_conversion_needed() -> None:
     """GEMINI 和 GEMINI_CLI 格式相同，只是认证不同，可透传（需开关启用）"""
     ok, needs_conv, reason = is_format_compatible(
-        "GEMINI_CLI",
-        "GEMINI",
+        "gemini:cli",
+        "gemini:chat",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=True,
@@ -215,8 +215,8 @@ def test_gemini_cli_to_gemini_no_conversion_needed() -> None:
 def test_claude_cli_to_claude_blocked_when_global_switch_disabled() -> None:
     """透传格式（CLAUDE_CLI -> CLAUDE）也受全局开关限制"""
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE_CLI",
-        "CLAUDE",
+        "claude:cli",
+        "claude:chat",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=False,
@@ -229,8 +229,8 @@ def test_claude_cli_to_claude_blocked_when_global_switch_disabled() -> None:
 def test_claude_cli_to_claude_blocked_when_endpoint_not_configured() -> None:
     """透传格式（CLAUDE_CLI -> CLAUDE）也需要端点配置"""
     ok, needs_conv, reason = is_format_compatible(
-        "CLAUDE_CLI",
-        "CLAUDE",
+        "claude:cli",
+        "claude:chat",
         endpoint_format_acceptance_config=None,
         is_stream=False,
         global_conversion_enabled=True,
@@ -246,8 +246,8 @@ def test_openai_cli_to_openai_needs_conversion() -> None:
     registry.can_convert_full.return_value = True
 
     ok, needs_conv, reason = is_format_compatible(
-        "OPENAI_CLI",
-        "OPENAI",
+        "openai:cli",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": True},  # 同族转换也需要端点配置
         is_stream=False,
         global_conversion_enabled=True,  # 同族转换也需要全局开关
@@ -264,8 +264,8 @@ def test_openai_to_openai_cli_needs_conversion() -> None:
     registry.can_convert_full.return_value = True
 
     ok, needs_conv, reason = is_format_compatible(
-        "OPENAI",
-        "OPENAI_CLI",
+        "openai:chat",
+        "openai:cli",
         endpoint_format_acceptance_config={"enabled": True},  # 同族转换也需要端点配置
         is_stream=False,
         global_conversion_enabled=True,  # 同族转换也需要全局开关
@@ -282,8 +282,8 @@ def test_openai_cli_to_openai_stream_needs_conversion() -> None:
     registry.can_convert_full.return_value = True
 
     ok, needs_conv, reason = is_format_compatible(
-        "OPENAI_CLI",
-        "OPENAI",
+        "openai:cli",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": True},  # 同族转换也需要端点配置
         is_stream=True,
         global_conversion_enabled=True,  # 同族转换也需要全局开关
@@ -300,8 +300,8 @@ def test_openai_cli_to_openai_fails_without_converter() -> None:
     registry.can_convert_full.return_value = False
 
     ok, needs_conv, reason = is_format_compatible(
-        "OPENAI_CLI",
-        "OPENAI",
+        "openai:cli",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=True,
@@ -318,8 +318,8 @@ def test_openai_cli_to_openai_blocked_when_global_switch_disabled() -> None:
     registry.can_convert_full.return_value = True
 
     ok, needs_conv, reason = is_format_compatible(
-        "OPENAI_CLI",
-        "OPENAI",
+        "openai:cli",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": True},
         is_stream=False,
         global_conversion_enabled=False,  # 全局开关关闭
@@ -336,8 +336,8 @@ def test_openai_cli_to_openai_blocked_when_endpoint_disabled() -> None:
     registry.can_convert_full.return_value = True
 
     ok, needs_conv, reason = is_format_compatible(
-        "OPENAI_CLI",
-        "OPENAI",
+        "openai:cli",
+        "openai:chat",
         endpoint_format_acceptance_config={"enabled": False},  # 端点开关关闭
         is_stream=False,
         global_conversion_enabled=True,
@@ -354,8 +354,8 @@ def test_openai_cli_to_openai_blocked_when_endpoint_not_configured() -> None:
     registry.can_convert_full.return_value = True
 
     ok, needs_conv, reason = is_format_compatible(
-        "OPENAI_CLI",
-        "OPENAI",
+        "openai:cli",
+        "openai:chat",
         endpoint_format_acceptance_config=None,  # 无端点配置
         is_stream=False,
         global_conversion_enabled=True,

tests/core/api_format/conversion/test_error_conversion.py

@@ -34,7 +34,7 @@ def test_error_conversion_openai_to_claude() -> None:
         "error": {"message": "bad request", "type": "invalid_request_error", "code": "bad_request"}
     }
 
-    out = reg.convert_error_response(openai_error, "OPENAI", "CLAUDE")
+    out = reg.convert_error_response(openai_error, "openai:chat", "claude:chat")
     assert out.get("type") == "error"
     assert isinstance(out.get("error"), dict)
     assert out["error"]["message"] == "bad request"
@@ -44,7 +44,7 @@ def test_error_conversion_claude_to_openai() -> None:
     reg = _make_registry()
 
     claude_error = {"type": "error", "error": {"type": "invalid_request_error", "message": "nope"}}
-    out = reg.convert_error_response(claude_error, "CLAUDE", "OPENAI")
+    out = reg.convert_error_response(claude_error, "claude:chat", "openai:chat")
     assert isinstance(out.get("error"), dict)
     assert out["error"]["message"] == "nope"
 
@@ -63,7 +63,7 @@ def test_error_event_stream_openai_to_claude_via_registry() -> None:
 
     # OpenAI 流式错误块
     chunk = {"error": {"message": "bad", "type": "invalid_request_error"}}
-    out = reg.convert_stream_chunk(chunk, "OPENAI", "CLAUDE", state=StreamState())
+    out = reg.convert_stream_chunk(chunk, "openai:chat", "claude:chat", state=StreamState())
     assert isinstance(out, list) and out
     evt0 = cast(dict[str, Any], out[0])
     assert evt0.get("type") == "error"

tests/core/api_format/conversion/test_gemini_normalizer.py

@@ -103,14 +103,17 @@ def test_gemini_request_parts_image_tool_and_unknown_drop() -> None:
             },
             {
                 "role": "model",
-                "parts": [
-                    {"function_call": {"name": "get_weather", "args": {"city": "SF"}}}
-                ],
+                "parts": [{"function_call": {"name": "get_weather", "args": {"city": "SF"}}}],
             },
             {
                 "role": "user",
                 "parts": [
-                    {"function_response": {"name": "call_1", "response": {"result": {"temp_c": 20}}}}
+                    {
+                        "function_response": {
+                            "name": "call_1",
+                            "response": {"result": {"temp_c": 20}},
+                        }
+                    }
                 ],
             },
         ]
@@ -194,16 +197,10 @@ def test_gemini_stream_chunk_and_event_roundtrip_basic() -> None:
 
     chunks = [
         {
-            "candidates": [
-                {"content": {"parts": [{"text": "Hel"}], "role": "model"}, "index": 0}
-            ],
+            "candidates": [{"content": {"parts": [{"text": "Hel"}], "role": "model"}, "index": 0}],
             "modelVersion": "gemini-1.5",
         },
-        {
-            "candidates": [
-                {"content": {"parts": [{"text": "lo"}], "role": "model"}, "index": 0}
-            ]
-        },
+        {"candidates": [{"content": {"parts": [{"text": "lo"}], "role": "model"}, "index": 0}]},
         {
             "candidates": [
                 {
@@ -226,7 +223,11 @@ def test_gemini_stream_chunk_and_event_roundtrip_basic() -> None:
                     "index": 0,
                 }
             ],
-            "usageMetadata": {"promptTokenCount": 1, "candidatesTokenCount": 2, "totalTokenCount": 3},
+            "usageMetadata": {
+                "promptTokenCount": 1,
+                "candidatesTokenCount": 2,
+                "totalTokenCount": 3,
+            },
             "modelVersion": "gemini-1.5",
         },
     ]
@@ -237,8 +238,13 @@ def test_gemini_stream_chunk_and_event_roundtrip_basic() -> None:
 
     assert any(isinstance(e, MessageStartEvent) for e in events)
     assert [e.text_delta for e in events if isinstance(e, ContentDeltaEvent)] == ["Hel", "lo"]
-    assert any(isinstance(e, ToolCallDeltaEvent) and json.loads(e.input_delta) == {"city": "SF"} for e in events)
-    assert any(isinstance(e, MessageStopEvent) and e.stop_reason == StopReason.END_TURN for e in events)
+    assert any(
+        isinstance(e, ToolCallDeltaEvent) and json.loads(e.input_delta) == {"city": "SF"}
+        for e in events
+    )
+    assert any(
+        isinstance(e, MessageStopEvent) and e.stop_reason == StopReason.END_TURN for e in events
+    )
 
     state2 = StreamState()
     out_chunks: list[dict[str, Any]] = []
@@ -253,7 +259,9 @@ def test_gemini_stream_chunk_and_event_roundtrip_basic() -> None:
         if c["candidates"][0]["content"]["parts"]
         and "functionCall" in c["candidates"][0]["content"]["parts"][0]
     )
-    assert tool_chunk["candidates"][0]["content"]["parts"][0]["functionCall"]["name"] == "get_weather"
+    assert (
+        tool_chunk["candidates"][0]["content"]["parts"][0]["functionCall"]["name"] == "get_weather"
+    )
 
     assert out_chunks[-1]["candidates"][0]["finishReason"] == "STOP"
 

tests/core/api_format/conversion/test_golden_canonical.py

@@ -18,7 +18,6 @@ from src.core.api_format.conversion.normalizers.openai import OpenAINormalizer
 from src.core.api_format.conversion.registry import FormatConversionRegistry
 from src.core.api_format.conversion.stream_state import StreamState
 
-
 GOLDEN_DIR = Path(__file__).resolve().parent / "golden_data"
 INPUT_DIR = GOLDEN_DIR / "inputs"
 EXPECTED_DIR = GOLDEN_DIR / "expected"
@@ -53,12 +52,12 @@ def _make_registry() -> FormatConversionRegistry:
 
 def test_golden_requests() -> None:
     reg = _make_registry()
-    formats = ["OPENAI", "CLAUDE", "GEMINI"]
+    formats = ["openai:chat", "claude:chat", "gemini:chat"]
 
     inputs = {
-        "OPENAI": _load_json(INPUT_DIR / "request_openai.json"),
-        "CLAUDE": _load_json(INPUT_DIR / "request_claude.json"),
-        "GEMINI": _load_json(INPUT_DIR / "request_gemini.json"),
+        "openai:chat": _load_json(INPUT_DIR / "request_openai.json"),
+        "claude:chat": _load_json(INPUT_DIR / "request_claude.json"),
+        "gemini:chat": _load_json(INPUT_DIR / "request_gemini.json"),
     }
 
     for source in formats:
@@ -72,12 +71,12 @@ def test_golden_requests() -> None:
 
 def test_golden_responses() -> None:
     reg = _make_registry()
-    formats = ["OPENAI", "CLAUDE", "GEMINI"]
+    formats = ["openai:chat", "claude:chat", "gemini:chat"]
 
     inputs = {
-        "OPENAI": _load_json(INPUT_DIR / "response_openai.json"),
-        "CLAUDE": _load_json(INPUT_DIR / "response_claude.json"),
-        "GEMINI": _load_json(INPUT_DIR / "response_gemini.json"),
+        "openai:chat": _load_json(INPUT_DIR / "response_openai.json"),
+        "claude:chat": _load_json(INPUT_DIR / "response_claude.json"),
+        "gemini:chat": _load_json(INPUT_DIR / "response_gemini.json"),
     }
 
     for source in formats:
@@ -91,12 +90,12 @@ def test_golden_responses() -> None:
 
 def test_golden_streams() -> None:
     reg = _make_registry()
-    formats = ["OPENAI", "CLAUDE", "GEMINI"]
+    formats = ["openai:chat", "claude:chat", "gemini:chat"]
 
     inputs: dict[str, list[dict[str, Any]]] = {
-        "OPENAI": _load_json(INPUT_DIR / "stream_openai.json"),
-        "CLAUDE": _load_json(INPUT_DIR / "stream_claude.json"),
-        "GEMINI": _load_json(INPUT_DIR / "stream_gemini.json"),
+        "openai:chat": _load_json(INPUT_DIR / "stream_openai.json"),
+        "claude:chat": _load_json(INPUT_DIR / "stream_claude.json"),
+        "gemini:chat": _load_json(INPUT_DIR / "stream_gemini.json"),
     }
 
     for source in formats:
@@ -106,7 +105,7 @@ def test_golden_streams() -> None:
             expected = _load_json(EXPECTED_DIR / f"stream_{source}_to_{target}.json")
 
             state = StreamState()
-            if source == "GEMINI":
+            if source == "gemini:chat":
                 state.message_id = "gemini_1"
 
             out: list[dict[str, Any]] = []

tests/core/api_format/conversion/test_internal.py

@@ -39,7 +39,6 @@ from src.core.api_format.conversion.internal import (
 )
 from src.core.api_format.conversion.stream_state import StreamState
 
-
 # ============================================================================
 # Enum 类型测试
 # ============================================================================
@@ -302,7 +301,9 @@ class TestInternalMessage:
             role=Role.ASSISTANT,
             content=[
                 TextBlock(text="Let me check the weather"),
-                ToolUseBlock(tool_id="t1", tool_name="get_weather", tool_input={"city": "Shanghai"}),
+                ToolUseBlock(
+                    tool_id="t1", tool_name="get_weather", tool_input={"city": "Shanghai"}
+                ),
             ],
         )
         assert msg.role == Role.ASSISTANT

tests/core/api_format/conversion/test_openai_normalizer.py

@@ -297,7 +297,9 @@ def test_openai_stream_chunk_and_event_roundtrip_basic() -> None:
     assert any(isinstance(e, ContentBlockStartEvent) for e in events)
     assert [e.text_delta for e in events if isinstance(e, ContentDeltaEvent)] == ["Hel", "lo"]
     assert any(isinstance(e, ToolCallDeltaEvent) for e in events)
-    assert any(isinstance(e, MessageStopEvent) and e.stop_reason == StopReason.TOOL_USE for e in events)
+    assert any(
+        isinstance(e, MessageStopEvent) and e.stop_reason == StopReason.TOOL_USE for e in events
+    )
 
     # internal events -> OpenAI chunks（验证关键字段与 tool_calls index 稳定）
     state2 = StreamState()
@@ -313,14 +315,20 @@ def test_openai_stream_chunk_and_event_roundtrip_basic() -> None:
 
     # tool_calls start chunk
     tool_start = next(
-        c for c in out_chunks if c["choices"][0]["delta"].get("tool_calls") and c["choices"][0]["delta"]["tool_calls"][0]["function"].get("name")
+        c
+        for c in out_chunks
+        if c["choices"][0]["delta"].get("tool_calls")
+        and c["choices"][0]["delta"]["tool_calls"][0]["function"].get("name")
     )
     assert tool_start["choices"][0]["delta"]["tool_calls"][0]["id"] == "call_1"
     assert tool_start["choices"][0]["delta"]["tool_calls"][0]["index"] == 0
 
     # tool_calls delta chunk（arguments 片段）
     tool_delta = next(
-        c for c in out_chunks if c["choices"][0]["delta"].get("tool_calls") and "arguments" in c["choices"][0]["delta"]["tool_calls"][0]["function"]
+        c
+        for c in out_chunks
+        if c["choices"][0]["delta"].get("tool_calls")
+        and "arguments" in c["choices"][0]["delta"]["tool_calls"][0]["function"]
     )
     assert tool_delta["choices"][0]["delta"]["tool_calls"][0]["id"] == "call_1"
     assert tool_delta["choices"][0]["delta"]["tool_calls"][0]["index"] == 0

tests/core/api_format/conversion/test_registry_canonical.py

@@ -37,9 +37,9 @@ def _first_openai_choice_message(resp: dict[str, Any]) -> dict[str, Any]:
 
 def test_registry_canonical_can_convert_full_stream() -> None:
     reg = _make_registry()
-    assert reg.can_convert_full("OPENAI", "CLAUDE", require_stream=True) is True
-    assert reg.can_convert_full("OPENAI", "GEMINI", require_stream=True) is True
-    assert reg.can_convert_full("CLAUDE", "GEMINI", require_stream=True) is True
+    assert reg.can_convert_full("openai:chat", "claude:chat", require_stream=True) is True
+    assert reg.can_convert_full("openai:chat", "gemini:chat", require_stream=True) is True
+    assert reg.can_convert_full("claude:chat", "gemini:chat", require_stream=True) is True
 
 
 def test_registry_canonical_request_openai_to_claude() -> None:
@@ -57,7 +57,7 @@ def test_registry_canonical_request_openai_to_claude() -> None:
         "stream": True,
     }
 
-    claude_req = reg.convert_request(openai_req, "OPENAI", "CLAUDE")
+    claude_req = reg.convert_request(openai_req, "openai:chat", "claude:chat")
     assert claude_req["model"] == "gpt-4o-mini"
     assert claude_req["system"] == "sys\n\ndev"
     assert claude_req["stream"] is True
@@ -79,7 +79,7 @@ def test_registry_canonical_response_claude_to_openai() -> None:
         "usage": {"input_tokens": 5, "output_tokens": 7},
     }
 
-    openai_resp = reg.convert_response(claude_resp, "CLAUDE", "OPENAI")
+    openai_resp = reg.convert_response(claude_resp, "claude:chat", "openai:chat")
     assert openai_resp["object"] == "chat.completion"
     msg = _first_openai_choice_message(openai_resp)
     assert msg["role"] == "assistant"
@@ -94,11 +94,13 @@ def test_registry_canonical_stream_openai_to_claude() -> None:
         "object": "chat.completion.chunk",
         "created": 1,
         "model": "gpt-4o-mini",
-        "choices": [{"index": 0, "delta": {"role": "assistant", "content": "hi"}, "finish_reason": None}],
+        "choices": [
+            {"index": 0, "delta": {"role": "assistant", "content": "hi"}, "finish_reason": None}
+        ],
     }
 
     state = StreamState()
-    out_events = reg.convert_stream_chunk(chunk, "OPENAI", "CLAUDE", state=state)
+    out_events = reg.convert_stream_chunk(chunk, "openai:chat", "claude:chat", state=state)
     assert isinstance(out_events, list) and out_events
 
     types = [cast(dict[str, Any], e).get("type") for e in cast(list[dict[str, Any]], out_events)]

tests/unit/test_api_format_signature.py

@@ -0,0 +1,41 @@
+import pytest
+
+from src.core.api_format.enums import ApiFamily, EndpointKind
+from src.core.api_format.signature import (
+    EndpointSignature,
+    make_signature_key,
+    normalize_endpoint_signature,
+    normalize_signature_key,
+    parse_signature_key,
+)
+
+
+def test_make_signature_key_accepts_enums_and_strings() -> None:
+    assert make_signature_key(ApiFamily.OPENAI, EndpointKind.CHAT) == "openai:chat"
+    assert make_signature_key("OpenAI", "CHAT") == "openai:chat"
+
+
+def test_parse_signature_key_roundtrip() -> None:
+    sig = parse_signature_key("OpenAI:CHAT")
+    assert sig.api_family == ApiFamily.OPENAI
+    assert sig.endpoint_kind == EndpointKind.CHAT
+    assert sig.key == "openai:chat"
+
+
+def test_parse_signature_key_invalid_raises() -> None:
+    with pytest.raises(ValueError):
+        parse_signature_key("OPENAI")  # missing ':'
+
+
+def test_normalize_signature_key() -> None:
+    assert normalize_signature_key("  OpenAI:CHAT  ") == "openai:chat"
+
+
+def test_normalize_endpoint_signature_tuple_input() -> None:
+    sig = normalize_endpoint_signature((ApiFamily.GEMINI, EndpointKind.VIDEO))
+    assert sig == EndpointSignature(api_family=ApiFamily.GEMINI, endpoint_kind=EndpointKind.VIDEO)
+
+
+def test_normalize_endpoint_signature_invalid_string_returns_default() -> None:
+    default = EndpointSignature(api_family=ApiFamily.CLAUDE, endpoint_kind=EndpointKind.CHAT)
+    assert normalize_endpoint_signature("not-a-signature", default=default) == default