自定义认证
现在您可以覆盖默认的 API 密钥身份验证。
用法
1. 创建一个自定义身份验证文件。
请确保响应类型遵循 UserAPIKeyAuth pydantic 对象。这用于记录该用户密钥特定的使用情况。
from fastapi import Request
from litellm.proxy._types import UserAPIKeyAuth
async def user_api_key_auth(request: Request, api_key: str) -> UserAPIKeyAuth:
try:
modified_master_key = "sk-my-master-key"
if api_key == modified_master_key:
return UserAPIKeyAuth(api_key=api_key)
raise Exception
except:
raise Exception
UserAPIKeyAuth 字段参考
UserAPIKeyAuth 对象支持以下字段,用于全面的身份验证配置
核心身份验证字段
UserAPIKeyAuth(
# Basic auth fields
api_key: Optional[str] = None, # The API key (will be hashed automatically)
token: Optional[str] = None, # Hashed token for internal use
key_name: Optional[str] = None, # Human-readable key name
key_alias: Optional[str] = None, # Key alias for identification
# User identification
user_id: Optional[str] = None, # Unique user identifier
user_email: Optional[str] = None, # User email address
user_role: Optional[LitellmUserRoles] = None, # User role (PROXY_ADMIN, INTERNAL_USER, etc.)
# Team/Organization
team_id: Optional[str] = None, # Team identifier
team_alias: Optional[str] = None, # Team display name
org_id: Optional[str] = None, # Organization identifier
)
预算和支出跟踪
UserAPIKeyAuth(
# User budgets
max_budget: Optional[float] = None, # Maximum budget for the key
spend: float = 0.0, # Current spend amount
soft_budget: Optional[float] = None, # Soft budget limit (warnings)
model_max_budget: Dict = {}, # Per-model budget limits
model_spend: Dict = {}, # Per-model spend tracking
# Team budgets
team_max_budget: Optional[float] = None, # Team's maximum budget
team_spend: Optional[float] = None, # Team's current spend
team_member_spend: Optional[float] = None, # This user's spend within the team
# Budget timing
budget_duration: Optional[str] = None, # Budget reset period
budget_reset_at: Optional[datetime] = None, # When budget resets
)
速率限制
UserAPIKeyAuth(
# User limits
tpm_limit: Optional[int] = None, # Tokens per minute limit
rpm_limit: Optional[int] = None, # Requests per minute limit
user_tpm_limit: Optional[int] = None, # User-specific TPM limit
user_rpm_limit: Optional[int] = None, # User-specific RPM limit
# Team limits
team_tpm_limit: Optional[int] = None, # Team TPM limit
team_rpm_limit: Optional[int] = None, # Team RPM limit
team_member_tpm_limit: Optional[int] = None, # Per-member TPM limit
team_member_rpm_limit: Optional[int] = None, # Per-member RPM limit
# Per-model limits
rpm_limit_per_model: Optional[Dict[str, int]] = None, # RPM limits by model
tpm_limit_per_model: Optional[Dict[str, int]] = None, # TPM limits by model
)
最终用户跟踪
UserAPIKeyAuth(
# End user identification and limits
end_user_id: Optional[str] = None, # End user identifier
end_user_tpm_limit: Optional[int] = None, # End user TPM limit
end_user_rpm_limit: Optional[int] = None, # End user RPM limit
end_user_max_budget: Optional[float] = None, # End user budget limit
)
模型和路由访问
UserAPIKeyAuth(
# Model access control
models: List = [], # Allowed models list (enforced when custom_auth_run_common_checks: true)
team_models: List = [], # Team's allowed models
aliases: Dict = {}, # Model aliases
# Route permissions
allowed_routes: Optional[list] = [], # Allowed API routes
allowed_cache_controls: Optional[list] = [], # Cache control permissions
permissions: Dict = {}, # General permissions
)
对象权限示例(MCP、代理等)
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
def _server_id(name: str) -> str:
server = global_mcp_server_manager.get_mcp_server_by_name(name)
if not server:
raise ValueError(f"Unknown MCP server '{name}'")
return server.server_id
object_permission = LiteLLM_ObjectPermissionTable(
mcp_servers=[_server_id("deepwiki"), _server_id("everything")], # MCP servers this key is allowed to use
mcp_tool_permissions={"deepwiki": ["search", "read_doc"]}, # optional per-server tool allow-list
)
UserAPIKeyAuth(
object_permission=object_permission,
)
高级配置
UserAPIKeyAuth(
# Request handling
max_parallel_requests: Optional[int] = None, # Concurrent request limit
allowed_model_region: Optional[AllowedModelRegion] = None, # Geographic restrictions
# Expiration and status
expires: Optional[Union[str, datetime]] = None, # Key expiration
blocked: Optional[bool] = None, # Whether key is blocked
# Metadata and configuration
metadata: Dict = {}, # Custom metadata
config: Dict = {}, # Configuration settings
team_metadata: Optional[Dict] = None, # Team metadata
# Internal tracking
request_route: Optional[str] = None, # Current request route
last_refreshed_at: Optional[float] = None, # Cache refresh timestamp
)
完整示例
from fastapi import Request
from datetime import datetime, timedelta
from litellm.proxy._types import UserAPIKeyAuth, LitellmUserRoles
async def user_api_key_auth(request: Request, api_key: str) -> UserAPIKeyAuth:
try:
# Example: Comprehensive auth configuration
if api_key.startswith("sk-admin-"):
return UserAPIKeyAuth(
api_key=api_key,
user_id="admin_user_123",
user_email="admin@company.com",
user_role=LitellmUserRoles.PROXY_ADMIN,
team_id="admin_team",
team_alias="Administrative Team",
max_budget=1000.0,
soft_budget=800.0,
tpm_limit=10000,
rpm_limit=100,
models=["gpt-4", "claude-3-sonnet", "gpt-3.5-turbo"],
allowed_routes=["/chat/completions", "/embeddings"],
expires=datetime.now() + timedelta(days=30),
metadata={"department": "engineering", "cost_center": "ai_ops"}
)
elif api_key.startswith("sk-team-"):
return UserAPIKeyAuth(
api_key=api_key,
user_id="team_user_456",
user_email="user@company.com",
user_role=LitellmUserRoles.INTERNAL_USER,
team_id="dev_team",
team_alias="Development Team",
max_budget=100.0,
tpm_limit=1000,
rpm_limit=20,
models=["gpt-3.5-turbo", "claude-3-haiku"],
team_member_tpm_limit=500, # Limit within team
end_user_tpm_limit=100, # Per end-user limit
metadata={"project": "chatbot_v2"}
)
else:
raise Exception("Invalid API key")
except Exception:
raise Exception("Authentication failed")
2. 传递文件路径(相对于 config.yaml)
将文件路径传递给 config.yaml
例如,如果它们都在同一个目录下 - ./config.yaml 和 ./custom_auth.py,它看起来像这样
model_list:
- model_name: "openai-model"
litellm_params:
model: "gpt-3.5-turbo"
litellm_settings:
drop_params: True
set_verbose: True
general_settings:
custom_auth: custom_auth.user_api_key_auth
3. 启动代理
$ litellm --config /path/to/config.yaml
强制执行模型访问、预算和团队/项目检查
默认情况下,当您的自定义身份验证处理程序返回 UserAPIKeyAuth 对象后,LiteLLM 不会运行标准的代理身份验证检查(模型允许列表、预算、团队/项目限制)。在返回的对象上设置 models=[...] 只会记录用于日志的允许列表——除非您选择启用,否则它不会阻止请求。
要强制执行 LiteLLM 的内置检查以及自定义身份验证,请设置
general_settings:
custom_auth: custom_auth.user_api_key_auth
custom_auth_run_common_checks: true
当 custom_auth_run_common_checks: true 时,LiteLLM 会运行与虚拟密钥相同的验证,包括
- 密钥级模型访问 — 您返回的
UserAPIKeyAuth上的models列表 - 团队/用户/项目模型访问 — 使用令牌上的
team_id、user_id和project_id从 LiteLLM 的数据库加载 - 预算和速率限制 — 密钥、团队、用户、项目和最终用户在配置时的预算
示例:在自定义身份验证中限制模型
async def user_api_key_auth(request: Request, api_key: str) -> UserAPIKeyAuth:
# ... validate api_key and load project_context from your system ...
return UserAPIKeyAuth(
api_key=api_key,
user_id=user_id,
team_id=project_context.team_id,
project_id=project_context.project_id,
models=project_context.models, # e.g. ["gpt-4o-mini", "claude-3-haiku"]
)
general_settings:
custom_auth: my_auth.user_api_key_auth
custom_auth_run_common_checks: true
如果没有 custom_auth_run_common_checks: true,客户端可以调用代理配置的任何模型(例如 gpt-4o),即使它不在您的 models 列表中。
密钥 models vs 项目 models
这些是独立的控制措施
| 字段 | 执行位置 | 单一事实来源 |
|---|---|---|
UserAPIKeyAuth 上的 models | 密钥级允许列表 | 您从自定义身份验证返回的值 |
UserAPIKeyAuth 上的 project_id | 项目级允许列表 | LiteLLM 数据库中项目记录上的 models |
如果您设置了 project_id,请同时使用正确的 models 列表在 LiteLLM 中创建/更新项目(通过 /project/new 或 UI)。请参阅 项目管理。
说明
- 空的
models列表 ([]) 表示在该范围内没有限制(允许所有模型)。 - 模型名称必须与代理配置中的模型组名称匹配,或者在支持的情况下使用通配符模式。
- 当启用通用检查时,请求体中的回退(fallback)模型也会根据密钥允许列表进行验证。
另请参阅:配置设置中的 custom_auth_run_common_checks。
✨ 支持 LiteLLM 虚拟密钥 + 自定义身份验证
v1.72.2+ 版本起支持
用法
- 设置自定义身份验证文件
"""
Example custom auth function.
This will allow all keys starting with "my-custom-key" to pass through.
"""
from typing import Union
from fastapi import Request
from litellm.proxy._types import UserAPIKeyAuth
async def user_api_key_auth(
request: Request, api_key: str
) -> Union[UserAPIKeyAuth, str]:
try:
if api_key.startswith("my-custom-key"):
return "sk-P1zJMdsqCPNN54alZd_ETw"
else:
raise Exception("Invalid API key")
except Exception:
raise Exception("Invalid API key")
- 设置 config.yaml
关键变更:设置 mode: auto。这将同时检查 LiteLLM API 密钥身份验证和自定义身份验证。
model_list:
- model_name: "openai-model"
litellm_params:
model: "gpt-3.5-turbo"
api_key: os.environ/OPENAI_API_KEY
general_settings:
custom_auth: custom_auth_auto.user_api_key_auth
custom_auth_settings:
mode: "auto" # can be 'on', 'off', 'auto' - 'auto' checks both litellm api key auth + custom auth
流程
-
优先检查自定义身份验证
-
如果自定义身份验证失败,则检查 LiteLLM API 密钥身份验证
-
如果两者都失败,则返回 401
-
测试它!
curl -L -X POST 'http://0.0.0.0:4000/v1/chat/completions' \
-H 'Content-Type: application/json' \
-H 'Authorization: Bearer sk-P1zJMdsqCPNN54alZd_ETw' \
-d '{
"model": "openai-model",
"messages": [
{
"role": "user",
"content": "Hey! My name is John"
}
]
}'
冒泡自定义异常
如果您想向上抛出自定义异常,可以通过引发 ProxyException 来实现。
"""
Example custom auth function.
This will allow all keys starting with "my-custom-key" to pass through.
"""
from typing import Union
from fastapi import Request
from litellm.proxy._types import UserAPIKeyAuth, ProxyException
async def user_api_key_auth(
request: Request, api_key: str
) -> Union[UserAPIKeyAuth, str]:
try:
if api_key.startswith("my-custom-key"):
return "sk-P1zJMdsqCPNN54alZd_ETw"
if api_key == "invalid-api-key":
# raise a custom exception back to the client
raise ProxyException(
message="Invalid API key",
type="invalid_request_error",
param="api_key",
code=401,
)
else:
raise Exception("Invalid API key")
except Exception:
raise Exception("Invalid API key")