refactor: 统一请求头处理逻辑到 headers.py

- 新增 src/core/headers.py 集中管理头部处理函数
- 扩展 api_format_metadata.py 添加 extra_headers 和 protected_keys 配置
- 将各 adapter 中重复的头部方法提取到基类，统一调用 headers.py
- 移除 transport.py 中未使用的 build_provider_headers 函数
- 添加 headers 模块的单元测试

src/api/handlers/base/chat_adapter_base.py

@@ -22,12 +22,13 @@ from abc import abstractmethod
 from typing import Any, Dict, Optional, Tuple, Type
 
 import httpx
-from fastapi import HTTPException
+from fastapi import HTTPException, Request
 from fastapi.responses import JSONResponse
 
 from src.api.base.adapter import ApiAdapter, ApiMode
 from src.api.base.context import ApiRequestContext
 from src.api.handlers.base.chat_handler_base import ChatHandlerBase
+from src.core.enums import APIFormat
 from src.core.exceptions import (
     InvalidRequestException,
     ModelNotSupportedException,
@@ -39,6 +40,12 @@ from src.core.exceptions import (
     QuotaExceededException,
     UpstreamClientException,
 )
+from src.core.headers import (
+    build_adapter_base_headers,
+    build_adapter_headers,
+    extract_client_api_key,
+    get_adapter_protected_keys,
+)
 from src.core.logger import logger
 from src.services.billing import calculate_request_cost as _calculate_request_cost
 from src.services.request.result import RequestResult
@@ -67,6 +74,14 @@ class ChatAdapterBase(ApiAdapter):
     # 计费模板配置（子类可覆盖，如 "claude", "openai", "gemini"）
     BILLING_TEMPLATE: str = "claude"
 
+    @classmethod
+    def _get_api_format(cls) -> APIFormat:
+        """获取 API 格式枚举，用于调用 headers.py 的统一函数"""
+        try:
+            return APIFormat[cls.FORMAT_ID]
+        except KeyError:
+            return APIFormat.OPENAI  # 默认回退
+
     # 子类可以配置的特殊方法（用于check_endpoint）
     @classmethod
     def build_endpoint_url(cls, base_url: str) -> str:
@@ -76,18 +91,20 @@ class ChatAdapterBase(ApiAdapter):
 
     @classmethod
     def build_base_headers(cls, api_key: str) -> Dict[str, str]:
-        """构建基础请求头，子类可以覆盖以自定义认证头"""
-        # 默认实现：Bearer token认证
-        return {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json",
-        }
+        """构建基础请求头，使用统一的 headers.py 实现"""
+        return build_adapter_base_headers(cls._get_api_format(), api_key)
 
     @classmethod
     def get_protected_header_keys(cls) -> tuple:
-        """返回不应被extra_headers覆盖的头部key，子类可以覆盖"""
-        # 默认保护认证相关头部
-        return ("authorization", "content-type")
+        """返回不应被extra_headers覆盖的头部key，使用统一的 headers.py 实现"""
+        return get_adapter_protected_keys(cls._get_api_format())
+
+    @classmethod
+    def build_headers_with_extra(
+        cls, api_key: str, extra_headers: Optional[Dict[str, str]] = None
+    ) -> Dict[str, str]:
+        """构建完整请求头（包含 extra_headers），使用统一的 headers.py 实现"""
+        return build_adapter_headers(cls._get_api_format(), api_key, extra_headers)
 
     @classmethod
     def build_request_body(cls, request_data: Dict[str, Any]) -> Dict[str, Any]:
@@ -95,6 +112,10 @@ class ChatAdapterBase(ApiAdapter):
         # 默认实现：直接使用请求数据
         return request_data.copy()
 
+    def extract_api_key(self, request: Request) -> Optional[str]:
+        """从请求中提取 API 密钥，使用统一的 headers.py 实现"""
+        return extract_client_api_key(dict(request.headers), self._get_api_format())
+
     def __init__(self, allowed_api_formats: Optional[list[str]] = None):
         self.allowed_api_formats = allowed_api_formats or [self.FORMAT_ID]
 
@@ -626,13 +647,11 @@ class ChatAdapterBase(ApiAdapter):
         Returns:
             测试响应数据
         """
-        from src.api.handlers.base.endpoint_checker import build_safe_headers, run_endpoint_check
+        from src.api.handlers.base.endpoint_checker import run_endpoint_check
 
         # 使用子类配置方法构建请求组件
         url = cls.build_endpoint_url(base_url)
-        base_headers = cls.build_base_headers(api_key)
-        protected_keys = cls.get_protected_header_keys()
-        headers = build_safe_headers(base_headers, extra_headers, protected_keys)
+        headers = cls.build_headers_with_extra(api_key, extra_headers)
         body = cls.build_request_body(request_data)
 
         # 使用通用的endpoint checker执行请求

src/api/handlers/base/cli_adapter_base.py

@@ -20,12 +20,13 @@ import traceback
 from typing import Any, Dict, Optional, Tuple, Type
 
 import httpx
-from fastapi import HTTPException
+from fastapi import HTTPException, Request
 from fastapi.responses import JSONResponse
 
 from src.api.base.adapter import ApiAdapter, ApiMode
 from src.api.base.context import ApiRequestContext
 from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
+from src.core.enums import APIFormat
 from src.core.exceptions import (
     InvalidRequestException,
     ModelNotSupportedException,
@@ -37,6 +38,12 @@ from src.core.exceptions import (
     QuotaExceededException,
     UpstreamClientException,
 )
+from src.core.headers import (
+    build_adapter_base_headers,
+    build_adapter_headers,
+    extract_client_api_key,
+    get_adapter_protected_keys,
+)
 from src.core.logger import logger
 from src.services.billing import calculate_request_cost as _calculate_request_cost
 from src.services.request.result import RequestResult
@@ -68,6 +75,55 @@ class CliAdapterBase(ApiAdapter):
     def __init__(self, allowed_api_formats: Optional[list[str]] = None):
         self.allowed_api_formats = allowed_api_formats or [self.FORMAT_ID]
 
+    # =========================================================================
+    # API 格式与头部处理 - 使用统一的 headers.py 函数
+    # =========================================================================
+
+    @classmethod
+    def _get_api_format(cls) -> APIFormat:
+        """将 FORMAT_ID 转换为 APIFormat 枚举"""
+        try:
+            return APIFormat[cls.FORMAT_ID]
+        except KeyError:
+            return APIFormat.OPENAI
+
+    def extract_api_key(self, request: Request) -> Optional[str]:
+        """
+        从请求中提取 API 密钥
+
+        使用统一的头部处理函数，根据 API 格式自动识别认证头。
+        """
+        return extract_client_api_key(dict(request.headers), self._get_api_format())
+
+    @classmethod
+    def build_base_headers(cls, api_key: str) -> Dict[str, str]:
+        """
+        构建 CLI API 认证头
+
+        使用统一的头部处理函数。
+        """
+        return build_adapter_base_headers(cls._get_api_format(), api_key)
+
+    @classmethod
+    def build_headers_with_extra(
+        cls, api_key: str, extra_headers: Optional[Dict[str, str]] = None
+    ) -> Dict[str, str]:
+        """
+        构建带额外头部的完整请求头
+
+        使用统一的头部处理函数，自动保护关键头部不被覆盖。
+        """
+        return build_adapter_headers(cls._get_api_format(), api_key, extra_headers)
+
+    @classmethod
+    def get_protected_header_keys(cls) -> tuple[str, ...]:
+        """
+        返回 CLI API 的保护头部 key
+
+        使用统一的头部处理函数。
+        """
+        return get_adapter_protected_keys(cls._get_api_format())
+
     async def handle(self, context: ApiRequestContext):
         """处理 CLI API 请求"""
         http_request = context.request
@@ -575,20 +631,19 @@ class CliAdapterBase(ApiAdapter):
         Returns:
             测试响应数据
         """
-        from src.api.handlers.base.endpoint_checker import build_safe_headers, run_endpoint_check
+        from src.api.handlers.base.endpoint_checker import run_endpoint_check
 
         # 构建请求组件
         url = cls.build_endpoint_url(base_url, request_data, model_name)
-        base_headers = cls.build_base_headers(api_key)
-        protected_keys = cls.get_protected_header_keys()
 
-        # 添加CLI User-Agent
+        # 添加 CLI User-Agent 到 extra_headers
         cli_user_agent = cls.get_cli_user_agent()
+        merged_extra = dict(extra_headers) if extra_headers else {}
         if cli_user_agent:
-            base_headers["User-Agent"] = cli_user_agent
-            protected_keys = tuple(list(protected_keys) + ["user-agent"])
+            merged_extra["User-Agent"] = cli_user_agent
 
-        headers = build_safe_headers(base_headers, extra_headers, protected_keys)
+        # 使用统一的头部构建函数
+        headers = cls.build_headers_with_extra(api_key, merged_extra if merged_extra else None)
         body = cls.build_request_body(request_data)
 
         # 获取有效的模型名称
@@ -610,7 +665,7 @@ class CliAdapterBase(ApiAdapter):
         )
 
     # =========================================================================
-    # CLI Adapter 配置方法 - 子类应覆盖这些方法而不是整个 check_endpoint
+    # CLI Adapter 配置方法 - 子类应覆盖这些方法
     # =========================================================================
 
     @classmethod
@@ -628,29 +683,6 @@ class CliAdapterBase(ApiAdapter):
         """
         raise NotImplementedError(f"{cls.FORMAT_ID} adapter must implement build_endpoint_url")
 
-    @classmethod
-    def build_base_headers(cls, api_key: str) -> Dict[str, str]:
-        """
-        构建CLI API认证头 - 子类应覆盖
-
-        Args:
-            api_key: API密钥
-
-        Returns:
-            基础认证头部字典
-        """
-        raise NotImplementedError(f"{cls.FORMAT_ID} adapter must implement build_base_headers")
-
-    @classmethod
-    def get_protected_header_keys(cls) -> tuple:
-        """
-        返回CLI API的保护头部key - 子类应覆盖
-
-        Returns:
-            保护头部key的元组
-        """
-        raise NotImplementedError(f"{cls.FORMAT_ID} adapter must implement get_protected_header_keys")
-
     @classmethod
     def build_request_body(cls, request_data: Dict[str, Any]) -> Dict[str, Any]:
         """

src/api/handlers/base/endpoint_checker.py

@@ -27,23 +27,11 @@ from collections import defaultdict
 import httpx
 
 from src.core.logger import logger
-
-
-_SENSITIVE_HEADER_KEYS = {
-    "authorization",
-    "x-api-key",
-    "x-goog-api-key",
-}
+from src.core.headers import CORE_REDACT_HEADERS, merge_headers_with_protection, redact_headers_for_log
 
 
 def _redact_headers(headers: Dict[str, str]) -> Dict[str, str]:
-    redacted: Dict[str, str] = {}
-    for key, value in headers.items():
-        if key.lower() in _SENSITIVE_HEADER_KEYS:
-            redacted[key] = "***"
-        else:
-            redacted[key] = value
-    return redacted
+    return redact_headers_for_log(headers, CORE_REDACT_HEADERS)
 
 
 def _truncate_repr(value: Any, limit: int = 1200) -> str:
@@ -64,14 +52,7 @@ def build_safe_headers(
     """
     合并 extra_headers，但防止覆盖 protected_keys（大小写不敏感）。
     """
-    headers = dict(base_headers)
-    if not extra_headers:
-        return headers
-
-    protected = {k.lower() for k in protected_keys}
-    safe_headers = {k: v for k, v in extra_headers.items() if k.lower() not in protected}
-    headers.update(safe_headers)
-    return headers
+    return merge_headers_with_protection(base_headers, extra_headers, set(protected_keys))
 
 
 async def run_endpoint_check(

src/api/handlers/base/request_builder.py

@@ -17,27 +17,14 @@ from abc import ABC, abstractmethod
 from typing import Any, Dict, FrozenSet, Optional, Tuple
 
 from src.core.crypto import crypto_service
+from src.core.headers import HeaderBuilder, UPSTREAM_DROP_HEADERS
 
 # ==============================================================================
 # 统一的头部配置常量
 # ==============================================================================
 
-# 敏感头部 - 透传时需要清理（黑名单）
-# 这些头部要么包含认证信息，要么由代理层重新生成
-SENSITIVE_HEADERS: FrozenSet[str] = frozenset(
-    {
-        "authorization",
-        "x-api-key",
-        "x-goog-api-key",  # Gemini API 认证头
-        "host",
-        "content-length",
-        "transfer-encoding",
-        "connection",
-        # 不透传 accept-encoding，让 httpx 自己协商压缩格式
-        # 避免客户端请求 brotli/zstd 但 httpx 不支持解压的问题
-        "accept-encoding",
-    }
-)
+# 兼容别名：历史代码使用 SENSITIVE_HEADERS 命名
+SENSITIVE_HEADERS: FrozenSet[str] = UPSTREAM_DROP_HEADERS
 
 
 # ==============================================================================
@@ -140,8 +127,6 @@ class PassthroughRequestBuilder(RequestBuilder):
         """
         from src.core.api_format_metadata import get_auth_config, resolve_api_format
 
-        headers: Dict[str, str] = {}
-
         # 1. 根据 API 格式自动设置认证头
         decrypted_key = crypto_service.decrypt(key.api_key)
         api_format = getattr(endpoint, "api_format", None)
@@ -150,32 +135,32 @@ class PassthroughRequestBuilder(RequestBuilder):
             get_auth_config(resolved_format) if resolved_format else ("Authorization", "bearer")
         )
 
-        if auth_type == "bearer":
-            headers[auth_header] = f"Bearer {decrypted_key}"
-        else:
-            headers[auth_header] = decrypted_key
+        auth_value = f"Bearer {decrypted_key}" if auth_type == "bearer" else decrypted_key
+        protected_keys = {auth_header.lower(), "content-type"}
 
-        # 2. 添加 endpoint 配置的额外头部
-        if endpoint.headers:
-            headers.update(endpoint.headers)
+        builder = HeaderBuilder()
 
-        # 3. 透传原始头部（排除敏感头部 - 黑名单模式）
+        # 2. 透传原始头部（排除敏感头部 - 黑名单模式）
         if original_headers:
             for name, value in original_headers.items():
-                lower_name = name.lower()
-
-                # 跳过敏感头部
-                if lower_name in SENSITIVE_HEADERS:
+                if name.lower() in SENSITIVE_HEADERS:
                     continue
+                builder.add(name, value)
 
-                headers[name] = value
+        # 3. 添加 endpoint 配置的额外头部（不能覆盖认证头/Content-Type）
+        if endpoint.headers:
+            builder.add_protected(endpoint.headers, protected_keys)
 
         # 4. 添加额外头部
         if extra_headers:
-            headers.update(extra_headers)
+            builder.add_many(extra_headers)
 
-        # 5. 确保有 Content-Type
-        if "Content-Type" not in headers and "content-type" not in headers:
+        # 5. 设置认证头（最高优先级）
+        builder.add(auth_header, auth_value)
+
+        # 6. 确保有 Content-Type
+        headers = builder.build()
+        if not any(k.lower() == "content-type" for k in headers):
             headers["Content-Type"] = "application/json"
 
         return headers

src/api/handlers/base/utils.py

@@ -6,6 +6,7 @@ import json
 from typing import Any, Dict, Optional
 
 from src.core.exceptions import EmbeddedErrorException, ProviderNotAvailableException
+from src.core.headers import filter_response_headers
 from src.core.logger import logger
 
 
@@ -94,25 +95,6 @@ def build_sse_headers(extra_headers: Optional[Dict[str, str]] = None) -> Dict[st
     return headers
 
 
-_PROXY_RESPONSE_HEADER_BLOCKLIST = frozenset(
-    {
-        # Body-dependent headers: 我们会重编码响应体（JSONResponse / SSE），不能透传上游值
-        "content-length",
-        "content-encoding",
-        "transfer-encoding",
-        "content-type",
-        # Hop-by-hop headers (RFC 7230)
-        "connection",
-        "keep-alive",
-        "proxy-authenticate",
-        "proxy-authorization",
-        "te",
-        "trailer",
-        "upgrade",
-    }
-)
-
-
 def filter_proxy_response_headers(headers: Optional[Dict[str, str]]) -> Dict[str, str]:
     """
     过滤上游响应头中不应透传给客户端的字段。
@@ -123,9 +105,7 @@ def filter_proxy_response_headers(headers: Optional[Dict[str, str]]) -> Dict[str
 
     如果透传上游的 `content-length/content-encoding/...`，会导致客户端解码失败或等待更多字节。
     """
-    if not headers:
-        return {}
-    return {k: v for k, v in headers.items() if k.lower() not in _PROXY_RESPONSE_HEADER_BLOCKLIST}
+    return filter_response_headers(headers)
 
 
 def check_html_response(line: str) -> bool:

src/api/handlers/claude/adapter.py

@@ -14,6 +14,7 @@ from src.api.base.adapter import ApiAdapter, ApiMode
 from src.api.base.context import ApiRequestContext
 from src.api.handlers.base.chat_adapter_base import ChatAdapterBase, register_adapter
 from src.api.handlers.base.chat_handler_base import ChatHandlerBase
+from src.core.headers import get_header_value
 from src.core.logger import logger
 from src.core.optimization_utils import TokenCounter
 from src.models.claude import ClaudeMessagesRequest, ClaudeTokenCountRequest
@@ -39,17 +40,10 @@ class ClaudeCapabilityDetector:
         """
         requirements: Dict[str, bool] = {}
 
-        # 检查 anthropic-beta 请求头（大小写不敏感）
-        beta_header = None
-        for key, value in headers.items():
-            if key.lower() == "anthropic-beta":
-                beta_header = value
-                break
-
-        if beta_header:
-            # 检查是否包含 context-1m 标识
-            if "context-1m" in beta_header.lower():
-                requirements["context_1m"] = True
+        # 使用统一的大小写不敏感获取
+        beta_header = get_header_value(headers, "anthropic-beta")
+        if beta_header and "context-1m" in beta_header.lower():
+            requirements["context_1m"] = True
 
         return requirements
 
@@ -77,10 +71,6 @@ class ClaudeChatAdapter(ChatAdapterBase):
         super().__init__(allowed_api_formats or ["CLAUDE"])
         logger.info(f"[{self.name}] 初始化Chat模式适配器 | API格式: {self.allowed_api_formats}")
 
-    def extract_api_key(self, request: Request) -> Optional[str]:
-        """从请求中提取 API 密钥 (x-api-key)"""
-        return request.headers.get("x-api-key")
-
     def detect_capability_requirements(
         self,
         headers: Dict[str, str],
@@ -166,18 +156,7 @@ class ClaudeChatAdapter(ChatAdapterBase):
         extra_headers: Optional[Dict[str, str]] = None,
     ) -> Tuple[list, Optional[str]]:
         """查询 Claude API 支持的模型列表"""
-        headers = {
-            "x-api-key": api_key,
-            "Authorization": f"Bearer {api_key}",
-            "anthropic-version": "2023-06-01",
-        }
-        if extra_headers:
-            # 防止 extra_headers 覆盖认证头
-            safe_headers = {
-                k: v for k, v in extra_headers.items()
-                if k.lower() not in ("x-api-key", "authorization", "anthropic-version")
-            }
-            headers.update(safe_headers)
+        headers = cls.build_headers_with_extra(api_key, extra_headers)
 
         # 构建 /v1/models URL
         base_url = base_url.rstrip("/")
@@ -219,20 +198,6 @@ class ClaudeChatAdapter(ChatAdapterBase):
         else:
             return f"{base_url}/v1/messages"
 
-    @classmethod
-    def build_base_headers(cls, api_key: str) -> Dict[str, str]:
-        """构建Claude API认证头"""
-        return {
-            "x-api-key": api_key,
-            "Content-Type": "application/json",
-            "anthropic-version": "2023-06-01",
-        }
-
-    @classmethod
-    def get_protected_header_keys(cls) -> tuple:
-        """返回Claude API的保护头部key"""
-        return ("x-api-key", "content-type", "anthropic-version")
-
     @classmethod
     def build_request_body(cls, request_data: Dict[str, Any]) -> Dict[str, Any]:
         """构建Claude API请求体"""

src/api/handlers/claude_cli/adapter.py

@@ -4,10 +4,9 @@ Claude CLI Adapter - 基于通用 CLI Adapter 基类的简化实现
 继承 CliAdapterBase，只需配置 FORMAT_ID 和 HANDLER_CLASS。
 """
 
-from typing import Any, AsyncIterator, Dict, Optional, Tuple, Type, Union
+from typing import Any, Dict, Optional, Tuple, Type
 
 import httpx
-from fastapi import Request
 
 from src.api.handlers.base.cli_adapter_base import CliAdapterBase, register_cli_adapter
 from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
@@ -37,13 +36,6 @@ class ClaudeCliAdapter(CliAdapterBase):
     def __init__(self, allowed_api_formats: Optional[list[str]] = None):
         super().__init__(allowed_api_formats or ["CLAUDE_CLI"])
 
-    def extract_api_key(self, request: Request) -> Optional[str]:
-        """从请求中提取 API 密钥 (Authorization: Bearer)"""
-        authorization = request.headers.get("authorization")
-        if authorization and authorization.startswith("Bearer "):
-            return authorization.replace("Bearer ", "")
-        return None
-
     def detect_capability_requirements(
         self,
         headers: Dict[str, str],
@@ -136,19 +128,6 @@ class ClaudeCliAdapter(CliAdapterBase):
         else:
             return f"{base_url}/v1/messages"
 
-    @classmethod
-    def build_base_headers(cls, api_key: str) -> Dict[str, str]:
-        """构建Claude CLI API认证头"""
-        return {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json",
-        }
-
-    @classmethod
-    def get_protected_header_keys(cls) -> tuple:
-        """返回Claude CLI API的保护头部key"""
-        return ("authorization", "content-type")
-
     @classmethod
     def build_request_body(cls, request_data: Dict[str, Any]) -> Dict[str, Any]:
         """构建Claude CLI API请求体"""

src/api/handlers/gemini/adapter.py

@@ -4,15 +4,14 @@ Gemini Chat Adapter
 处理 Gemini API 格式的请求适配
 """
 
-from typing import Any, AsyncIterator, Dict, Optional, Tuple, Type, Union
+from typing import Any, Dict, Optional, Tuple, Type
 
 import httpx
-from fastapi import HTTPException, Request
+from fastapi import HTTPException
 from fastapi.responses import JSONResponse
 
 from src.api.handlers.base.chat_adapter_base import ChatAdapterBase, register_adapter
 from src.api.handlers.base.chat_handler_base import ChatHandlerBase
-from src.api.handlers.base.endpoint_checker import build_safe_headers, run_endpoint_check
 from src.core.logger import logger
 from src.models.gemini import GeminiRequest
 
@@ -41,10 +40,6 @@ class GeminiChatAdapter(ChatAdapterBase):
         super().__init__(allowed_api_formats or ["GEMINI"])
         logger.info(f"[{self.name}] 初始化 Gemini Chat 适配器 | API格式: {self.allowed_api_formats}")
 
-    def extract_api_key(self, request: Request) -> Optional[str]:
-        """从请求中提取 API 密钥 (x-goog-api-key)"""
-        return request.headers.get("x-goog-api-key")
-
     def _merge_path_params(
         self, original_request_body: Dict[str, Any], path_params: Dict[str, Any]  # noqa: ARG002
     ) -> Dict[str, Any]:
@@ -163,7 +158,7 @@ class GeminiChatAdapter(ChatAdapterBase):
         extra_headers: Optional[Dict[str, str]] = None,
     ) -> Tuple[list, Optional[str]]:
         """查询 Gemini API 支持的模型列表"""
-        # 兼容 base_url 已包含 /v1beta 的情况
+        # Gemini 使用 URL 参数传递 key，不需要 headers 中的认证
         base_url_clean = base_url.rstrip("/")
         if base_url_clean.endswith("/v1beta"):
             models_url = f"{base_url_clean}/models?key={api_key}"
@@ -210,19 +205,6 @@ class GeminiChatAdapter(ChatAdapterBase):
         else:
             return f"{base_url}/v1beta"
 
-    @classmethod
-    def build_base_headers(cls, api_key: str) -> Dict[str, str]:
-        """构建Gemini API认证头"""
-        return {
-            "x-goog-api-key": api_key,
-            "Content-Type": "application/json",
-        }
-
-    @classmethod
-    def get_protected_header_keys(cls) -> tuple:
-        """返回Gemini API的保护头部key"""
-        return ("x-goog-api-key", "content-type")
-
     @classmethod
     def build_request_body(cls, request_data: Dict[str, Any]) -> Dict[str, Any]:
         """构建Gemini API请求体"""
@@ -263,13 +245,11 @@ class GeminiChatAdapter(ChatAdapterBase):
             }
 
         # 使用基类配置方法，但重写URL构建逻辑
-        base_url = cls.build_endpoint_url(base_url)
-        url = f"{base_url}/models/{effective_model_name}:generateContent"
+        base_url_resolved = cls.build_endpoint_url(base_url)
+        url = f"{base_url_resolved}/models/{effective_model_name}:generateContent"
 
         # 构建请求组件
-        base_headers = cls.build_base_headers(api_key)
-        protected_keys = cls.get_protected_header_keys()
-        headers = build_safe_headers(base_headers, extra_headers, protected_keys)
+        headers = cls.build_headers_with_extra(api_key, extra_headers)
         body = cls.build_request_body(request_data)
 
         # 使用基类的通用endpoint checker
@@ -290,7 +270,7 @@ class GeminiChatAdapter(ChatAdapterBase):
         )
 
 
-def build_gemini_adapter(x_app_header: str = "") -> GeminiChatAdapter:
+def build_gemini_adapter(x_app_header: str = "") -> GeminiChatAdapter:  # noqa: ARG001
     """
     根据请求头构建适当的 Gemini 适配器
 

src/api/handlers/gemini_cli/adapter.py

@@ -4,10 +4,9 @@ Gemini CLI Adapter - 基于通用 CLI Adapter 基类的实现
 继承 CliAdapterBase，处理 Gemini CLI 格式的请求。
 """
 
-from typing import Any, AsyncIterator, Dict, Optional, Tuple, Type, Union
+from typing import Any, Dict, Optional, Tuple, Type
 
 import httpx
-from fastapi import Request
 
 from src.api.handlers.base.cli_adapter_base import CliAdapterBase, register_cli_adapter
 from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
@@ -37,10 +36,6 @@ class GeminiCliAdapter(CliAdapterBase):
     def __init__(self, allowed_api_formats: Optional[list[str]] = None):
         super().__init__(allowed_api_formats or ["GEMINI_CLI"])
 
-    def extract_api_key(self, request: Request) -> Optional[str]:
-        """从请求中提取 API 密钥 (x-goog-api-key)"""
-        return request.headers.get("x-goog-api-key")
-
     def _merge_path_params(
         self, original_request_body: Dict[str, Any], path_params: Dict[str, Any]  # noqa: ARG002
     ) -> Dict[str, Any]:
@@ -138,19 +133,6 @@ class GeminiCliAdapter(CliAdapterBase):
             prefix = f"{base_url}/v1beta"
         return f"{prefix}/models/{effective_model_name}:generateContent"
 
-    @classmethod
-    def build_base_headers(cls, api_key: str) -> Dict[str, str]:
-        """构建Gemini CLI API认证头"""
-        return {
-            "x-goog-api-key": api_key,
-            "Content-Type": "application/json",
-        }
-
-    @classmethod
-    def get_protected_header_keys(cls) -> tuple:
-        """返回Gemini CLI API的保护头部key"""
-        return ("x-goog-api-key", "content-type")
-
     @classmethod
     def build_request_body(cls, request_data: Dict[str, Any]) -> Dict[str, Any]:
         """构建Gemini CLI API请求体"""

src/api/handlers/openai/adapter.py

@@ -4,14 +4,12 @@ OpenAI Chat Adapter - 基于 ChatAdapterBase 的 OpenAI Chat API 适配器
 处理 /v1/chat/completions 端点的 OpenAI Chat 格式请求。
 """
 
-from typing import Any, AsyncIterator, Dict, Optional, Tuple, Type, Union
+from typing import Any, Dict, Optional, Tuple, Type
 
 import httpx
-from fastapi import Request
 from fastapi.responses import JSONResponse
 
 from src.api.handlers.base.chat_adapter_base import ChatAdapterBase, register_adapter
-from src.api.handlers.base.endpoint_checker import build_safe_headers, run_endpoint_check
 from src.api.handlers.base.chat_handler_base import ChatHandlerBase
 from src.core.logger import logger
 from src.models.openai import OpenAIRequest
@@ -39,13 +37,6 @@ class OpenAIChatAdapter(ChatAdapterBase):
     def __init__(self, allowed_api_formats: Optional[list[str]] = None):
         super().__init__(allowed_api_formats or ["OPENAI"])
 
-    def extract_api_key(self, request: Request) -> Optional[str]:
-        """从请求中提取 API 密钥 (Authorization: Bearer)"""
-        authorization = request.headers.get("authorization")
-        if authorization and authorization.startswith("Bearer "):
-            return authorization.replace("Bearer ", "")
-        return None
-
     def _validate_request_body(self, original_request_body: dict, path_params: dict = None):
         """验证请求体"""
         if not isinstance(original_request_body, dict):
@@ -117,13 +108,7 @@ class OpenAIChatAdapter(ChatAdapterBase):
         extra_headers: Optional[Dict[str, str]] = None,
     ) -> Tuple[list, Optional[str]]:
         """查询 OpenAI 兼容 API 支持的模型列表"""
-        headers = {
-            "Authorization": f"Bearer {api_key}",
-        }
-        if extra_headers:
-            # 防止 extra_headers 覆盖 Authorization
-            safe_headers = {k: v for k, v in extra_headers.items() if k.lower() != "authorization"}
-            headers.update(safe_headers)
+        headers = cls.build_headers_with_extra(api_key, extra_headers)
 
         # 构建 /v1/models URL
         base_url = base_url.rstrip("/")
@@ -165,23 +150,5 @@ class OpenAIChatAdapter(ChatAdapterBase):
         else:
             return f"{base_url}/v1/chat/completions"
 
-    @classmethod
-    def build_base_headers(cls, api_key: str) -> Dict[str, str]:
-        """构建OpenAI API认证头"""
-        return {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json",
-        }
-
-    @classmethod
-    def get_protected_header_keys(cls) -> tuple:
-        """返回OpenAI API的保护头部key"""
-        return ("authorization", "content-type")
-
-    @classmethod
-    def build_request_body(cls, request_data: Dict[str, Any]) -> Dict[str, Any]:
-        """构建OpenAI API请求体"""
-        return request_data.copy()
-
 
 __all__ = ["OpenAIChatAdapter"]

src/api/handlers/openai_cli/adapter.py

@@ -4,10 +4,9 @@ OpenAI CLI Adapter - 基于通用 CLI Adapter 基类的简化实现
 继承 CliAdapterBase，只需配置 FORMAT_ID 和 HANDLER_CLASS。
 """
 
-from typing import Any, AsyncIterator, Dict, Optional, Tuple, Type, Union
+from typing import Any, Dict, Optional, Tuple, Type
 
 import httpx
-from fastapi import Request
 
 from src.api.handlers.base.cli_adapter_base import CliAdapterBase, register_cli_adapter
 from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
@@ -37,13 +36,6 @@ class OpenAICliAdapter(CliAdapterBase):
     def __init__(self, allowed_api_formats: Optional[list[str]] = None):
         super().__init__(allowed_api_formats or ["OPENAI_CLI"])
 
-    def extract_api_key(self, request: Request) -> Optional[str]:
-        """从请求中提取 API 密钥 (Authorization: Bearer)"""
-        authorization = request.headers.get("authorization")
-        if authorization and authorization.startswith("Bearer "):
-            return authorization.replace("Bearer ", "")
-        return None
-
     # =========================================================================
     # 模型列表查询
     # =========================================================================
@@ -78,19 +70,6 @@ class OpenAICliAdapter(CliAdapterBase):
         else:
             return f"{base_url}/v1/chat/completions"
 
-    @classmethod
-    def build_base_headers(cls, api_key: str) -> Dict[str, str]:
-        """构建OpenAI CLI API认证头"""
-        return {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json",
-        }
-
-    @classmethod
-    def get_protected_header_keys(cls) -> tuple:
-        """返回OpenAI CLI API的保护头部key"""
-        return ("authorization", "content-type")
-
     @classmethod
     def build_request_body(cls, request_data: Dict[str, Any]) -> Dict[str, Any]:
         """构建OpenAI CLI API请求体"""