feat: 支持固定类型 Provider OAuth 授权

- 新增 provider_type 字段区分自定义/预置 Provider 类型（claude_code/codex/gemini_cli/antigravity） - 实现完整 OAuth 2.0 授权流程：start（生成授权 URL + PKCE）、complete（换取 token）、refresh - 前端 KeyFormDialog 添加 OAuth 授权 UI，支持开始授权、粘贴回调 URL、完成授权、强制刷新 - 请求时自动检测 token 过期并刷新（120s 预留窗口 + Redis 分布式锁防并发） - 固定类型 Provider 自动创建预置端点并锁定 base_url/custom_path - 数据库迁移：添加 providers.provider_type，扩展 api_key 列为 TEXT - 可选依赖 tls-client 用于 Claude token 请求的 TLS 指纹伪装
2026-09-02 09:20:22 +08:00 · 2026-02-04 10:24:25 +08:00
parent f6dac1c38a
commit e4fdc65e52
25 changed files with 1818 additions and 36 deletions
--- a/src/api/admin/init.py
+++ b/src/api/admin/init.py
@@ -13,6 +13,7 @@ from .monitoring import router as monitoring_router
 from .provider_ops import router as provider_ops_router
 from .provider_query import router as provider_query_router
 from .provider_strategy import router as provider_strategy_router
+from .provider_oauth import router as provider_oauth_router
 from .providers import router as providers_router
 from .security import router as security_router
 from .stats import router as stats_router
@@ -31,6 +32,7 @@ router.include_router(usage_router)
 router.include_router(monitoring_router)
 router.include_router(endpoints_router)
 router.include_router(provider_strategy_router)
+router.include_router(provider_oauth_router)
 router.include_router(adaptive_router)
 router.include_router(models_router)
 router.include_router(security_router)
--- a/src/api/admin/endpoints/keys.py
+++ b/src/api/admin/endpoints/keys.py
@@ -241,11 +241,11 @@ class AdminUpdateEndpointKeyAdapter(AdminApiAdapter):
        # auth_type 切换校验 + 字段归一化
        if "auth_type" in update_data:
            if target_auth_type == "api_key":
-                if current_auth_type == "vertex_ai" and not update_data.get("api_key"):
+                if current_auth_type in {"vertex_ai", "oauth"} and not update_data.get("api_key"):
                    raise InvalidRequestException(
-                        "从 Vertex AI 切换到 API Key 认证模式时，必须提供新的 API Key"
+                        "切换到 API Key 认证模式时，必须提供新的 API Key"
                    )
-                # 切换回 API Key：清理 Service Account 配置
+                # 切换回 API Key：清理非本模式配置
                update_data["auth_config"] = None
            elif target_auth_type == "vertex_ai":
                if current_auth_type != "vertex_ai" and not update_data.get("auth_config"):
@@ -255,6 +255,12 @@ class AdminUpdateEndpointKeyAdapter(AdminApiAdapter):
                # Vertex AI 不使用 api_key：写入占位符（若未提供 api_key）
                if "api_key" not in update_data:
                    update_data["api_key"] = "__placeholder__"
+            elif target_auth_type == "oauth":
+                # OAuth 的 token 不允许在 key 更新接口里手工写入
+                if update_data.get("api_key"):
+                    raise InvalidRequestException("OAuth 认证模式下不允许直接填写 api_key")
+                if "api_key" not in update_data:
+                    update_data["api_key"] = "__placeholder__"

        # 加密 api_key（非 None 时）
        if "api_key" in update_data and update_data["api_key"] is not None:
@@ -604,6 +610,8 @@ def _build_key_response(
    if auth_type == "vertex_ai":
        # Vertex AI 使用 Service Account，不显示占位符
        masked_key = "[Service Account]"
+    elif auth_type == "oauth":
+        masked_key = "[OAuth Token]"
    else:
        try:
            decrypted_key = crypto_service.decrypt(key.api_key)
@@ -726,6 +734,10 @@ class AdminCreateProviderKeyAdapter(AdminApiAdapter):
        elif auth_type == "vertex_ai":
            if not self.key_data.auth_config:
                raise InvalidRequestException("Service Account 认证模式下 auth_config 为必填字段")
+        elif auth_type == "oauth":
+            # OAuth key 的 token 通过 provider-oauth 授权流程写入（此处不允许手填）
+            if self.key_data.api_key:
+                raise InvalidRequestException("OAuth 认证模式下不允许直接填写 api_key")

        # 允许同一个 API Key 在同一 Provider 下添加多次
        # 用户可以为不同的 API 格式创建独立的配置记录，便于分开管理
@@ -736,6 +748,9 @@ class AdminCreateProviderKeyAdapter(AdminApiAdapter):
            if self.key_data.api_key
            else crypto_service.encrypt("__placeholder__")  # 占位符，保持 NOT NULL 约束
        )
+        # OAuth 类型 key 初始写入占位符（token 由 provider-oauth 流程写入）
+        if auth_type == "oauth":
+            encrypted_key = crypto_service.encrypt("__placeholder__")
        now = datetime.now(timezone.utc)

        # 加密 auth_config（包含敏感的 Service Account 凭证）
@@ -787,9 +802,10 @@ class AdminCreateProviderKeyAdapter(AdminApiAdapter):
        db.commit()
        db.refresh(new_key)

+        key_tail = (self.key_data.api_key or "")[-4:]
        logger.info(
            f"[OK] 添加 Key: Provider={self.provider_id}, "
-            f"Formats={self.key_data.api_formats}, Key=***{self.key_data.api_key[-4:]}, ID={new_key.id}"
+            f"Formats={self.key_data.api_formats}, Key=***{key_tail}, ID={new_key.id}"
        )

        # 如果开启了 auto_fetch_models，同步执行模型获取
--- a/src/api/admin/endpoints/routes.py
+++ b/src/api/admin/endpoints/routes.py
@@ -281,6 +281,11 @@ class AdminCreateProviderEndpointAdapter(AdminApiAdapter):
        if not provider:
            raise NotFoundException(f"Provider {self.provider_id} 不存在")

+        # 固定类型 Provider：禁止通过该接口新增 Endpoints（端点由模板自动创建并锁定）
+        provider_type = (getattr(provider, "provider_type", "custom") or "custom").strip()
+        if provider_type != "custom":
+            raise InvalidRequestException("固定类型 Provider 不允许手动新增 Endpoint")
+
        if self.endpoint_data.provider_id != self.provider_id:
            raise InvalidRequestException("provider_id 不匹配")

@@ -414,6 +419,16 @@ class AdminUpdateProviderEndpointAdapter(AdminApiAdapter):

        update_data = self.endpoint_data.model_dump(exclude_unset=True)

+        # 固定类型 Provider 的 endpoint：锁定 base_url/custom_path（前端禁用仅是 UX，后端必须强校验）
+        provider = db.query(Provider).filter(Provider.id == endpoint.provider_id).first()
+        if provider:
+            provider_type = (getattr(provider, "provider_type", "custom") or "custom").strip()
+            if provider_type != "custom":
+                if "base_url" in update_data or "custom_path" in update_data:
+                    raise InvalidRequestException(
+                        "固定类型 Provider 的 Endpoint 不允许修改 base_url/custom_path"
+                    )
+
        # 把 proxy 转换为 dict 存储，支持显式设置为 None 清除代理
        if "proxy" in update_data:
            if update_data["proxy"] is not None:
--- a/src/api/admin/provider_oauth.py
+++ b/src/api/admin/provider_oauth.py
@@ -0,0 +1,512 @@
+"""管理员 Provider OAuth 管理 API。
+
+用于固定类型 Provider 的 OAuth2 授权：
+- start: 生成授权 URL（PKCE/state）
+- complete: 粘贴 callback_url 完成换 token
+- refresh: 手动强制刷新 token
+
+注意：
+- 该模块是“上游 Provider OAuth（用于反代调用）”，不是用户登录/绑定 OAuth。
+- 不得在日志或响应中返回 access_token/refresh_token/client_secret。
+"""
+
+from __future__ import annotations
+
+import json
+import secrets
+import time
+from dataclasses import dataclass
+from typing import Any
+import base64
+import hashlib
+
+from urllib.parse import parse_qsl, urlencode, urlparse
+
+import httpx
+from fastapi import APIRouter, Depends, Request
+from pydantic import BaseModel, Field
+from redis.asyncio import Redis
+from sqlalchemy.orm import Session
+
+from src.clients.redis_client import get_redis_client
+from src.core.crypto import crypto_service
+from src.core.exceptions import InvalidRequestException, NotFoundException
+from src.core.logger import logger
+from src.database.database import get_db
+from src.models.database import Provider, ProviderAPIKey
+
+from src.core.provider_oauth_utils import enrich_auth_config, post_oauth_token
+from src.core.provider_templates.fixed_providers import FIXED_PROVIDERS
+from src.core.provider_templates.types import ProviderType
+
+router = APIRouter(prefix="/api/admin/provider-oauth", tags=["Provider OAuth"])
+
+
+# ==============================================================================
+# Redis state storage
+# ==============================================================================
+
+_PROVIDER_OAUTH_STATE_TTL_SECONDS = 600
+_PROVIDER_OAUTH_STATE_PREFIX = "provider_oauth_state:"
+
+
+_CONSUME_STATE_SCRIPT = r"""
+local value = redis.call("GET", KEYS[1])
+if value then
+    redis.call("DEL", KEYS[1])
+end
+return value
+"""
+
+
+def _state_key(nonce: str) -> str:
+    return f"{_PROVIDER_OAUTH_STATE_PREFIX}{nonce}"
+
+
+@dataclass(frozen=True)
+class ProviderOAuthStateData:
+    nonce: str
+    key_id: str
+    provider_type: str
+    pkce_verifier: str | None
+    created_at: int
+
+
+async def _create_state(
+    redis: Redis,
+    *,
+    key_id: str,
+    provider_type: str,
+    pkce_verifier: str | None,
+) -> str:
+    nonce = secrets.token_urlsafe(24)
+    data = {
+        "nonce": nonce,
+        "key_id": key_id,
+        "provider_type": provider_type,
+        "pkce_verifier": pkce_verifier,
+        "created_at": int(time.time()),
+    }
+    await redis.setex(_state_key(nonce), _PROVIDER_OAUTH_STATE_TTL_SECONDS, json.dumps(data))
+    return nonce
+
+
+async def _consume_state(redis: Redis, nonce: str) -> ProviderOAuthStateData | None:
+    if not nonce:
+        return None
+    key = _state_key(nonce)
+    raw = await redis.eval(_CONSUME_STATE_SCRIPT, 1, key)
+    if not raw:
+        return None
+    try:
+        parsed = json.loads(raw)
+    except Exception:
+        return None
+    return ProviderOAuthStateData(
+        nonce=str(parsed.get("nonce") or ""),
+        key_id=str(parsed.get("key_id") or ""),
+        provider_type=str(parsed.get("provider_type") or ""),
+        pkce_verifier=parsed.get("pkce_verifier"),
+        created_at=int(parsed.get("created_at") or 0),
+    )
+
+
+# ==============================================================================
+# Requests / responses
+# ==============================================================================
+
+
+class StartOAuthResponse(BaseModel):
+    authorization_url: str
+    redirect_uri: str
+    provider_type: str
+    instructions: str
+
+
+class CompleteOAuthRequest(BaseModel):
+    callback_url: str = Field(..., min_length=5, description="浏览器地址栏中的完整回调 URL")
+
+
+class CompleteOAuthResponse(BaseModel):
+    provider_type: str
+    expires_at: int | None = None
+    has_refresh_token: bool = False
+
+
+# ==============================================================================
+# Helpers
+# ==============================================================================
+
+
+def _require_fixed_provider(provider: Provider) -> str:
+    provider_type = (getattr(provider, "provider_type", "custom") or "custom").strip()
+    if provider_type == "custom":
+        raise InvalidRequestException("该 Provider 不是固定类型，无法使用 provider-oauth")
+    return provider_type
+
+
+def _pkce_s256(verifier: str) -> str:
+    digest = hashlib.sha256(verifier.encode("utf-8")).digest()
+    return base64.urlsafe_b64encode(digest).decode("utf-8").rstrip("=")
+
+
+def _parse_callback_params(callback_url: str) -> dict[str, str]:
+    parsed = urlparse(callback_url.strip())
+    query = dict(parse_qsl(parsed.query, keep_blank_values=True))
+    fragment = dict(parse_qsl((parsed.fragment or "").lstrip("#"), keep_blank_values=True))
+    merged = {**query, **fragment}
+
+    # Claude 参考实现里：code 参数可能包含 "<code>#<state>" 的拼接形式
+    code = merged.get("code")
+    if code and "#" in code:
+        code_part, state_part = code.split("#", 1)
+        merged["code"] = code_part
+        if "state" not in merged and state_part:
+            merged["state"] = state_part
+
+    return {str(k): str(v) for k, v in merged.items()}
+
+
+# ==============================================================================
+# Routes
+# ==============================================================================
+
+
+@router.get("/supported-types")
+async def supported_types() -> list[dict[str, Any]]:
+    # 不返回 client_secret
+    result: list[dict[str, Any]] = []
+    for provider_type, template in FIXED_PROVIDERS.items():
+        result.append(
+            {
+                "provider_type": str(provider_type.value) if hasattr(provider_type, "value") else str(provider_type),
+                "display_name": template.display_name,
+                "scopes": list(template.oauth.scopes),
+                "redirect_uri": template.oauth.redirect_uri,
+                "authorize_url": template.oauth.authorize_url,
+                "token_url": template.oauth.token_url,
+                "use_pkce": bool(template.oauth.use_pkce),
+            }
+        )
+    return result
+
+
+@router.post("/keys/{key_id}/start", response_model=StartOAuthResponse)
+async def start_oauth(
+    key_id: str,
+    request: Request,
+    db: Session = Depends(get_db),
+) -> StartOAuthResponse:
+    key = db.query(ProviderAPIKey).filter(ProviderAPIKey.id == key_id).first()
+    if not key:
+        raise NotFoundException("Key 不存在", "key")
+    if (getattr(key, "auth_type", "api_key") or "api_key") != "oauth":
+        raise InvalidRequestException("该 Key 不是 oauth 认证类型")
+
+    provider = db.query(Provider).filter(Provider.id == key.provider_id).first()
+    if not provider:
+        raise NotFoundException("Provider 不存在", "provider")
+    provider_type = _require_fixed_provider(provider)
+
+    try:
+        template = FIXED_PROVIDERS.get(ProviderType(provider_type))
+    except Exception:
+        template = None
+    if not template:
+        raise InvalidRequestException("不支持的 provider_type")
+
+    redis = await get_redis_client(require_redis=True)
+    assert redis is not None
+
+    pkce_verifier: str | None = None
+    code_challenge: str | None = None
+
+    if template.oauth.use_pkce:
+        pkce_verifier = secrets.token_urlsafe(32)
+        code_challenge = _pkce_s256(pkce_verifier)
+
+    state = await _create_state(
+        redis,
+        key_id=key_id,
+        provider_type=provider_type,
+        pkce_verifier=pkce_verifier,
+    )
+
+    params: dict[str, Any] = {
+        "client_id": template.oauth.client_id,
+        "response_type": "code",
+        "redirect_uri": template.oauth.redirect_uri,
+        "scope": " ".join(template.oauth.scopes),
+        "state": state,
+    }
+
+    # Codex 参考实现额外参数
+    if provider_type == ProviderType.CODEX.value:
+        params.update(
+            {
+                "prompt": "login",
+                "id_token_add_organizations": "true",
+                "codex_cli_simplified_flow": "true",
+            }
+        )
+
+    if template.oauth.use_pkce and code_challenge:
+        params["code_challenge"] = code_challenge
+        params["code_challenge_method"] = "S256"
+
+    authorization_url = f"{template.oauth.authorize_url}?{urlencode(params)}"
+
+    return StartOAuthResponse(
+        authorization_url=authorization_url,
+        redirect_uri=template.oauth.redirect_uri,
+        provider_type=provider_type,
+        instructions=(
+            "1) 打开 authorization_url 完成授权\n"
+            "2) 授权后会跳转到 redirect_uri（localhost）\n"
+            "3) 复制浏览器地址栏完整 URL，调用 complete 接口粘贴 callback_url"
+        ),
+    )
+
+
+@router.post("/keys/{key_id}/complete", response_model=CompleteOAuthResponse)
+async def complete_oauth(
+    key_id: str,
+    payload: CompleteOAuthRequest,
+    request: Request,
+    db: Session = Depends(get_db),
+) -> CompleteOAuthResponse:
+    redis = await get_redis_client(require_redis=True)
+    assert redis is not None
+
+    params = _parse_callback_params(payload.callback_url)
+    code = params.get("code")
+    state = params.get("state")
+    if not code or not state:
+        raise InvalidRequestException("callback_url 缺少 code/state")
+
+    state_data = await _consume_state(redis, state)
+    if not state_data or state_data.key_id != key_id:
+        raise InvalidRequestException("state 无效或已过期")
+
+    key = db.query(ProviderAPIKey).filter(ProviderAPIKey.id == key_id).first()
+    if not key:
+        raise NotFoundException("Key 不存在", "key")
+    if (getattr(key, "auth_type", "api_key") or "api_key") != "oauth":
+        raise InvalidRequestException("该 Key 不是 oauth 认证类型")
+
+    provider = db.query(Provider).filter(Provider.id == key.provider_id).first()
+    if not provider:
+        raise NotFoundException("Provider 不存在", "provider")
+    provider_type = _require_fixed_provider(provider)
+
+    try:
+        template = FIXED_PROVIDERS.get(ProviderType(provider_type))
+    except Exception:
+        template = None
+    if not template:
+        raise InvalidRequestException("不支持的 provider_type")
+
+    # exchange token
+    token_url = template.oauth.token_url
+
+    # Claude token endpoint 是 JSON；Codex/Google 是 form。这里先做最小实现：按 URL 判断。
+    is_json = "anthropic.com" in token_url
+
+    if is_json:
+        body: dict[str, Any] = {
+            "grant_type": "authorization_code",
+            "client_id": template.oauth.client_id,
+            "redirect_uri": template.oauth.redirect_uri,
+            "code": code,
+            "state": state,
+        }
+        if state_data.pkce_verifier:
+            body["code_verifier"] = state_data.pkce_verifier
+        headers = {"Content-Type": "application/json", "Accept": "application/json"}
+        data = None
+        json_body = body
+    else:
+        form: dict[str, str] = {
+            "grant_type": "authorization_code",
+            "client_id": template.oauth.client_id,
+            "redirect_uri": template.oauth.redirect_uri,
+            "code": code,
+        }
+        if template.oauth.client_secret:
+            form["client_secret"] = template.oauth.client_secret
+        if state_data.pkce_verifier:
+            form["code_verifier"] = state_data.pkce_verifier
+        headers = {"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}
+        data = form
+        json_body = None
+
+    proxy_config = getattr(provider, "proxy", None)
+
+    resp = await post_oauth_token(
+        provider_type=provider_type,
+        token_url=token_url,
+        headers=headers,
+        data=data,
+        json_body=json_body,
+        proxy_config=proxy_config,
+        timeout_seconds=30.0,
+    )
+
+    if resp.status_code < 200 or resp.status_code >= 300:
+        raise InvalidRequestException("token exchange 失败")
+
+    token = resp.json()
+    access_token = str(token.get("access_token") or "")
+    refresh_token = str(token.get("refresh_token") or "")
+    expires_in = token.get("expires_in")
+    expires_at: int | None = None
+    try:
+        if expires_in is not None:
+            expires_at = int(time.time()) + int(expires_in)
+    except Exception:
+        expires_at = None
+
+    if not access_token:
+        raise InvalidRequestException("token exchange 返回缺少 access_token")
+
+    # store
+    key.api_key = crypto_service.encrypt(access_token)
+    auth_config: dict[str, Any] = {
+        "provider_type": provider_type,
+        "token_type": token.get("token_type"),
+        "refresh_token": refresh_token or None,
+        "expires_at": expires_at,
+        "scope": token.get("scope"),
+        "updated_at": int(time.time()),
+    }
+
+    auth_config = await enrich_auth_config(
+        provider_type=provider_type,
+        auth_config=auth_config,
+        token_response=token,
+        access_token=access_token,
+        proxy_config=proxy_config,
+    )
+
+    key.auth_config = crypto_service.encrypt(json.dumps(auth_config))
+    db.commit()
+
+    return CompleteOAuthResponse(
+        provider_type=provider_type,
+        expires_at=expires_at,
+        has_refresh_token=bool(refresh_token),
+    )
+
+
+@router.post("/keys/{key_id}/refresh", response_model=CompleteOAuthResponse)
+async def refresh_oauth(
+    key_id: str,
+    request: Request,
+    db: Session = Depends(get_db),
+) -> CompleteOAuthResponse:
+    key = db.query(ProviderAPIKey).filter(ProviderAPIKey.id == key_id).first()
+    if not key:
+        raise NotFoundException("Key 不存在", "key")
+    if (getattr(key, "auth_type", "api_key") or "api_key") != "oauth":
+        raise InvalidRequestException("该 Key 不是 oauth 认证类型")
+
+    provider = db.query(Provider).filter(Provider.id == key.provider_id).first()
+    if not provider:
+        raise NotFoundException("Provider 不存在", "provider")
+    provider_type = _require_fixed_provider(provider)
+
+    try:
+        template = FIXED_PROVIDERS.get(ProviderType(provider_type))
+    except Exception:
+        template = None
+    if not template:
+        raise InvalidRequestException("不支持的 provider_type")
+
+    encrypted_auth_config = getattr(key, "auth_config", None)
+    if not encrypted_auth_config:
+        raise InvalidRequestException("缺少 auth_config，无法 refresh")
+
+    decrypted = crypto_service.decrypt(encrypted_auth_config)
+    parsed = json.loads(decrypted)
+    refresh_token = str(parsed.get("refresh_token") or "")
+    if not refresh_token:
+        raise InvalidRequestException("缺少 refresh_token，需要重新授权")
+
+    token_url = template.oauth.token_url
+    is_json = "anthropic.com" in token_url
+
+    if is_json:
+        body: dict[str, Any] = {
+            "grant_type": "refresh_token",
+            "client_id": template.oauth.client_id,
+            "refresh_token": refresh_token,
+        }
+        headers = {"Content-Type": "application/json", "Accept": "application/json"}
+        data = None
+        json_body = body
+    else:
+        form: dict[str, str] = {
+            "grant_type": "refresh_token",
+            "client_id": template.oauth.client_id,
+            "refresh_token": refresh_token,
+        }
+        if template.oauth.client_secret:
+            form["client_secret"] = template.oauth.client_secret
+        headers = {"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}
+        data = form
+        json_body = None
+
+    proxy_config = getattr(provider, "proxy", None)
+
+    resp = await post_oauth_token(
+        provider_type=provider_type,
+        token_url=token_url,
+        headers=headers,
+        data=data,
+        json_body=json_body,
+        proxy_config=proxy_config,
+        timeout_seconds=30.0,
+    )
+
+    if resp.status_code < 200 or resp.status_code >= 300:
+        raise InvalidRequestException("token refresh 失败")
+
+    token = resp.json()
+    access_token = str(token.get("access_token") or "")
+    new_refresh_token = str(token.get("refresh_token") or "")
+    expires_in = token.get("expires_in")
+    expires_at: int | None = None
+    try:
+        if expires_in is not None:
+            expires_at = int(time.time()) + int(expires_in)
+    except Exception:
+        expires_at = None
+
+    if not access_token:
+        raise InvalidRequestException("token refresh 返回缺少 access_token")
+
+    # store
+    key.api_key = crypto_service.encrypt(access_token)
+    parsed["token_type"] = token.get("token_type")
+    if new_refresh_token:
+        parsed["refresh_token"] = new_refresh_token
+    parsed["expires_at"] = expires_at
+    parsed["scope"] = token.get("scope")
+    parsed["updated_at"] = int(time.time())
+
+    parsed = await enrich_auth_config(
+        provider_type=provider_type,
+        auth_config=parsed,
+        token_response=token,
+        access_token=access_token,
+        proxy_config=proxy_config,
+    )
+
+    key.auth_config = crypto_service.encrypt(json.dumps(parsed))
+    db.commit()
+
+    return CompleteOAuthResponse(
+        provider_type=provider_type,
+        expires_at=expires_at,
+        has_refresh_token=bool(parsed.get("refresh_token")),
+    )
--- a/src/api/admin/providers/routes.py
+++ b/src/api/admin/providers/routes.py
@@ -3,6 +3,7 @@
 from __future__ import annotations

 import asyncio
+import uuid
 from datetime import datetime, timezone
 from typing import Any

@@ -20,10 +21,13 @@ from src.core.logger import logger
 from src.core.model_permissions import match_model_with_pattern, parse_allowed_models_to_list
 from src.database import get_db
 from src.models.admin_requests import CreateProviderRequest, UpdateProviderRequest
-from src.models.database import GlobalModel, Provider, ProviderAPIKey
+from src.models.database import GlobalModel, Provider, ProviderAPIKey, ProviderEndpoint
 from src.services.cache.model_cache import ModelCacheService
 from src.services.cache.provider_cache import ProviderCacheService

+from src.core.provider_templates.fixed_providers import FIXED_PROVIDERS
+from src.core.provider_templates.types import ProviderType
+
 router = APIRouter(tags=["Provider CRUD"])
 pipeline = ApiRequestPipeline()

@@ -290,6 +294,7 @@ class AdminCreateProviderAdapter(AdminApiAdapter):
            # 创建 Provider 对象
            provider = Provider(
                name=validated_data.name,
+                provider_type=validated_data.provider_type or "custom",
                description=validated_data.description,
                website=validated_data.website,
                billing_type=billing_type,
@@ -309,6 +314,37 @@ class AdminCreateProviderAdapter(AdminApiAdapter):
            )

            db.add(provider)
+            db.flush()  # flush 获取 ID，但不提交，保持在同一事务中
+
+            # 固定类型 Provider：自动创建并锁定预置 Endpoints（同一事务）
+            provider_type = (provider.provider_type or "custom").strip()
+            if provider_type != "custom":
+                try:
+                    template = FIXED_PROVIDERS.get(ProviderType(provider_type))
+                except Exception:
+                    template = None
+                if template:
+                    now = datetime.now(timezone.utc)
+                    for sig in template.endpoint_signatures:
+                        endpoint = ProviderEndpoint(
+                            id=str(uuid.uuid4()),
+                            provider_id=provider.id,
+                            api_format=sig,
+                            api_family=sig.split(":", 1)[0],
+                            endpoint_kind=sig.split(":", 1)[1],
+                            base_url=template.api_base_url,
+                            custom_path=None,
+                            header_rules=None,
+                            max_retries=provider.max_retries or 2,
+                            is_active=True,
+                            config=None,
+                            proxy=None,
+                            format_acceptance_config=None,
+                            created_at=now,
+                            updated_at=now,
+                        )
+                        db.add(endpoint)
+
            db.commit()
            db.refresh(provider)

@@ -369,6 +405,8 @@ class AdminUpdateProviderAdapter(AdminApiAdapter):
                if field == "billing_type" and value is not None:
                    # billing_type 需要转换为枚举
                    setattr(provider, field, ProviderBillingType(value))
+                elif field == "provider_type" and value is not None:
+                    setattr(provider, field, value)
                elif field == "proxy" and value is not None:
                    # proxy 需要转换为 dict（如果是 Pydantic 模型）
                    setattr(
--- a/src/api/admin/providers/summary.py
+++ b/src/api/admin/providers/summary.py
@@ -305,6 +305,7 @@ def _build_provider_summary(db: Session, provider: Provider) -> ProviderWithEndp
    return ProviderWithEndpointsSummary(
        id=provider.id,
        name=provider.name,
+        provider_type=getattr(provider, "provider_type", None),
        description=provider.description,
        website=provider.website,
        provider_priority=provider.provider_priority,