Files
taiji-AI-PAD/services/mcp-server/app/routes/channel.py
T
chenchenandClaude Opus 4.7 eb17ed84f8 fix(mcp-server): plug LiteLLM API key leak via register & tenant model assignment errors
A failed POST /api/auth/register returned the SQLAlchemy IntegrityError verbatim
to the caller, which included the full INSERT INTO tenant_model_keys statement
along with every bound parameter — ~50 plaintext LiteLLM API keys per failed
attempt. Same pattern was reproduced in 3 channel.py endpoints that wrap
LiteLLM key INSERTs.

Changes:
- channel.py: assign_resources_to_tenant / assign_model_to_tenant /
  update_tenant_model_quota — log full exc_info, return a typed
  {code, message} error instead of f"...{str(e)}". 6 leakage points sealed.
- email_verification.py: add peek_verification_code() — checks a code
  without burning it. Lets the register handler verify *before* the
  multi-step transaction so a downstream failure doesn't waste the user's
  one-shot code.
- scripts/cleanup_orphan_litellm_keys.py: one-shot orphan key reaper.
  Scans LiteLLM /key/list by metadata.tenant_id (plus a manual list of
  the 8 publicly-leaked sk- prefixes from the original incident).
  Used to nuke 16 orphan keys for tenant fab9dc27-… on 2026-05-12.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 15:42:38 +08:00

3936 lines
139 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
渠道合作伙伴API路由
"""
from datetime import datetime, timedelta
from typing import List, Optional, Dict, Any
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import select, func, and_, desc, or_, update
from sqlalchemy.ext.asyncio import AsyncSession
import uuid
import structlog
from database import get_db
from models import (
User, Channel, Agent, ResourceAllocation,
RechargeRecord, Application, ModelProvider,
ChannelProviderAccess, ProviderApplication, TenantCustomAgentQuota,
ChannelCustomAgentQuota, ResourceApplication, PlatformAgentQuota,
AgentBillingRecord, PlatformAgentTemplateConfig, TenantModelKey, Balance
)
# 注意:BillingRecord 已废弃,使用 AgentBillingRecord 和 ModelBillingRecord 替代
from app.auth import require_auth, get_password_hash
from app.permissions import has_permission
from app.schemas import (
SuccessResponse,
TenantCreateRequest,
AllocateResourcesRequest,
UpdateBillingRequest,
RechargeTenantRequest,
RechargeTenantResponse,
SetCreditLimitRequest,
SetCreditLimitResponse,
ResourceApplicationRequest,
ChannelBillingResponse,
ApplyProviderRequest,
UpdateTenantStatusRequest,
UpdateTenantPermissionsRequest,
CreateAdminRequest,
ChangeTenantPasswordRequest,
ApplyPlatformAgentRequest,
AllocatePlatformAgentRequest,
)
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
from app.litellm_client import get_litellm_client, LiteLLMClientError
logger = structlog.get_logger(__name__)
# ============= 默认值常量 =============
# 配额和余额默认值,除非前端明确传递
DEFAULT_MAX_BUDGET = 500 # 默认最大预算 $500
DEFAULT_BUDGET_DURATION = "monthly" # 默认月度预算
router = APIRouter(prefix="/api/channel", tags=["渠道合作伙伴"])
def _get_role(principal: dict) -> str:
"""从principal获取角色"""
return principal.get("claims", {}).get("role", "")
def _verify_permission(principal: dict, permission: str):
"""验证是否拥有指定权限(使用统一权限系统)"""
role = _get_role(principal)
# super_admin 拥有所有权限
if role == "super_admin":
return
if not has_permission(role, permission):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"需要权限: {permission}"
)
def _get_channel_id(principal: dict) -> Optional[uuid.UUID]:
"""获取当前用户的渠道ID(如果是渠道下的管理员)"""
role = _get_role(principal)
# 超级管理员没有固定的channel_id,可以访问所有渠道
if role == "super_admin":
return None
if role in ["billing_admin", "operations_admin", "channel_admin"]:
channel_id_str = principal.get("claims", {}).get("channelId")
if channel_id_str:
try:
return uuid.UUID(channel_id_str)
except ValueError:
return None
return None
# ============= 租户管理 =============
@router.get("/tenants", response_model=SuccessResponse)
async def list_tenants(
channel_id: Optional[str] = Query(None, description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取渠道下的租户列表
权限:view:tenants (channel_admin, billing_admin, operations_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数
- 其他管理员自动使用自己所属的渠道
"""
_verify_permission(principal, "view:tenants")
user_channel_id = _get_channel_id(principal)
role = _get_role(principal)
# 确定目标渠道ID
if role == "super_admin":
# 超级管理员必须提供 channel_id
if not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
target_channel_id = uuid.UUID(channel_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
target_channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证渠道存在
channel_result = await db.execute(
select(Channel).where(Channel.id == target_channel_id)
)
channel = channel_result.scalar_one_or_none()
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="渠道不存在"
)
# 查询渠道下的租户
result = await db.execute(
select(User).where(User.channel_id == target_channel_id)
)
tenants = result.scalars().all()
# 获取所有租户的余额(从 Balance 表)
tenant_ids = [tenant.id for tenant in tenants]
balance_result = await db.execute(
select(Balance).where(Balance.user_id.in_(tenant_ids))
)
balances = {b.user_id: float(b.eu_balance) for b in balance_result.scalars().all()}
data = [
{
"id": str(tenant.id),
"name": tenant.name,
"email": tenant.email,
"subscriptionTier": tenant.subscription_tier,
"balance": balances.get(tenant.id, 0.0), # 从 Balance 表获取余额
"creditLimit": float(tenant.credit_limit),
"status": tenant.status,
"createdAt": tenant.created_at.isoformat(),
}
for tenant in tenants
]
return SuccessResponse(data={"tenants": data, "channelId": str(target_channel_id), "channelName": channel.name})
@router.post("/tenants/create", response_model=SuccessResponse)
async def create_tenant(
req: TenantCreateRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
创建租户
权限:manage:tenants (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channelId 参数
- 其他管理员自动使用自己所属的渠道
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
# 超级管理员必须提供 channelId
if not hasattr(req, 'channelId') or not req.channelId:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员创建租户时必须提供 channelId 参数"
)
try:
channel_id = uuid.UUID(req.channelId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 检查邮箱是否已存在
result = await db.execute(
select(User).where(User.email == req.email)
)
if result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="邮箱已被使用"
)
# 创建租户
password_hash = get_password_hash(req.password)
tenant = User(
name=req.name,
email=req.email,
password_hash=password_hash,
hashed_password=password_hash, # 兼容
username=req.email.split("@")[0],
full_name=req.name,
role="user",
channel_id=channel_id,
subscription_tier=req.subscriptionTier,
status="active",
credit_limit=0,
)
db.add(tenant)
await db.commit()
await db.refresh(tenant)
return SuccessResponse(
data={
"id": str(tenant.id),
"name": tenant.name,
"email": tenant.email,
},
message="租户创建成功"
)
@router.put("/tenants/{tenant_id}/resources", response_model=SuccessResponse)
async def allocate_tenant_resources(
tenant_id: str,
req: AllocateResourcesRequest,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
为租户分配资源
支持分配:
- 平台端 Agent(agents 字段)
- 模型资源(models 字段)
- 自定义 Agent 配额(customAgentQuota 字段)
权限:manage:resources (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:resources")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证 tenant_id 格式并确认租户属于指定渠道
try:
tenant_uuid = uuid.UUID(tenant_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的租户ID格式"
)
result = await db.execute(
select(User).where(
and_(
User.id == tenant_uuid,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 获取租户所属渠道(用于验证配额)
tenant_channel_id = tenant.channel_id
# 追加模式:不再删除现有资源分配,而是在原有基础上追加
from sqlalchemy import delete
# 分配平台 Agent 资源(更新 PlatformAgentQuota)
for agent_alloc in req.agents:
# 1. 追加模式:查找或创建 ResourceAllocation 记录
existing_alloc_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == tenant_id,
ResourceAllocation.target_type == "tenant",
ResourceAllocation.resource_type == "agent",
ResourceAllocation.resource_id == agent_alloc.agentId
)
)
)
existing_alloc = existing_alloc_result.scalar_one_or_none()
if existing_alloc:
# 追加模式:在现有数量基础上增加
existing_alloc.quantity = (existing_alloc.quantity or 0) + agent_alloc.quantity
else:
# 创建新的分配记录
allocation = ResourceAllocation(
target_id=tenant_id,
target_type="tenant",
resource_type="agent",
resource_id=agent_alloc.agentId,
quantity=agent_alloc.quantity,
)
db.add(allocation)
# 2. 更新 PlatformAgentQuota(平台 Agent 配额管理)
# agentId 在这里是模板名称(如 echo_agent)
template_name = agent_alloc.agentId
# 获取渠道的平台 Agent 配额(使用行锁防止并发更新)
channel_quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_id == channel_id,
PlatformAgentQuota.target_type == "channel",
PlatformAgentQuota.template_name == template_name
)
)
.with_for_update() # 行锁
)
channel_quota = channel_quota_result.scalar_one_or_none()
if channel_quota:
# 计算已分配给该渠道其他租户的配额
other_tenants_quota_result = await db.execute(
select(func.sum(PlatformAgentQuota.pod_quota).label("total"))
.select_from(PlatformAgentQuota)
.join(User, PlatformAgentQuota.target_id == User.id)
.where(
and_(
User.channel_id == channel_id,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.template_name == template_name,
PlatformAgentQuota.target_id != tenant_id
)
)
)
other_quota = other_tenants_quota_result.scalar() or 0
# 获取当前租户已有配额(使用行锁防止并发更新)
tenant_quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_id == tenant_id,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.template_name == template_name
)
)
.with_for_update() # 行锁
)
tenant_quota = tenant_quota_result.scalar_one_or_none()
current_tenant_quota = tenant_quota.pod_quota if tenant_quota else 0
# 追加模式:quantity 表示要追加的配额数量
quota_delta = agent_alloc.quantity # 追加的配额量
new_quota = current_tenant_quota + quota_delta # 新的总配额
# 检查追加量是否超过剩余配额
remaining = channel_quota.pod_quota - other_quota - current_tenant_quota
if quota_delta > remaining:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"平台 Agent '{template_name}' 配额超出渠道剩余配额。渠道剩余: {remaining},请求追加: {quota_delta}"
)
# 更新或创建租户配额记录
if tenant_quota:
# 追加模式:在现有配额基础上增加
tenant_quota.pod_quota = new_quota
tenant_quota.allocated_at = datetime.utcnow()
else:
# 创建新配额记录
tenant_quota = PlatformAgentQuota(
target_id=tenant_id,
target_type="tenant",
template_name=template_name,
pod_quota=new_quota,
pod_used=0,
allocated_by=None, # 渠道管理员分配
allocated_at=datetime.utcnow(),
)
db.add(tenant_quota)
# 更新渠道的 pod_used(分配给租户的配额视为渠道已使用的配额)
channel_quota.pod_used = (channel_quota.pod_used or 0) + quota_delta
logger.info(
f"追加平台 Agent 配额: template={template_name}, tenant={tenant_id}, "
f"追加量={quota_delta}, 新配额={new_quota}, channel_pod_used={channel_quota.pod_used}"
)
else:
# 渠道没有该平台 Agent 的配额,记录警告但不阻止操作
logger.warning(
f"渠道没有平台 Agent '{template_name}' 的配额,仅创建 ResourceAllocation 记录"
)
# 分配模型资源(集成 LiteLLM)
# 获取渠道信息(用于 LiteLLM 集成)
channel_result = await db.execute(
select(Channel).where(Channel.id == channel_id)
)
channel = channel_result.scalar_one_or_none()
for model_alloc in req.models:
# 1. 验证渠道是否有该模型的权限(通过 ResourceAllocation 检查)
channel_model_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == channel_id,
ResourceAllocation.target_type == "channel",
ResourceAllocation.resource_type == "model",
ResourceAllocation.resource_id == model_alloc.modelName
)
)
)
channel_model_allocation = channel_model_result.scalar_one_or_none()
if not channel_model_allocation:
# 渠道没有该模型的权限,拒绝分配
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"渠道没有模型 '{model_alloc.modelName}' 的权限,请先向管理员申请"
)
# 2. 检查租户是否已有该模型的 Key
existing_key_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == tenant_id,
TenantModelKey.model_name == model_alloc.modelName
)
)
)
existing_key = existing_key_result.scalar_one_or_none()
# 3. 如果渠道有 LiteLLM team_id,则集成 LiteLLM
# 注意:LiteLLM 操作失败时,整个分配操作也会失败
if channel and channel.litellm_team_id:
try:
litellm_client = get_litellm_client()
if existing_key:
# 更新现有 Key 的配额(使用默认值 500)
await litellm_client.update_key(
key=existing_key.litellm_key_id,
rpm_limit=model_alloc.rpm,
tpm_limit=model_alloc.tpm,
max_budget=DEFAULT_MAX_BUDGET, # 默认 500
budget_duration=DEFAULT_BUDGET_DURATION, # 默认 monthly
)
# 更新数据库记录
existing_key.rpm_limit = model_alloc.rpm
existing_key.tpm_limit = model_alloc.tpm
existing_key.max_budget = DEFAULT_MAX_BUDGET
existing_key.budget_duration = DEFAULT_BUDGET_DURATION
logger.info(f"更新租户 {tenant.name} 的模型 {model_alloc.modelName} 配额(默认预算 ${DEFAULT_MAX_BUDGET}/月)")
else:
# 创建新的 LiteLLM Key(使用默认值 500)
key = await litellm_client.generate_key(
team_id=channel.litellm_team_id,
models=[model_alloc.modelName],
rpm_limit=model_alloc.rpm,
tpm_limit=model_alloc.tpm,
max_budget=DEFAULT_MAX_BUDGET, # 默认 500
budget_duration=DEFAULT_BUDGET_DURATION, # 默认 monthly
key_name=f"tenant-{tenant_id}-{model_alloc.modelName}",
metadata={
"tenant_id": str(tenant_id),
"tenant_name": tenant.name,
"channel_id": str(channel_id),
"channel_name": channel.name,
"model": model_alloc.modelName,
}
)
# 加密存储 Key
encrypted_key = litellm_client.encrypt_key(key.key)
# 保存到数据库
tenant_key = TenantModelKey(
tenant_id=tenant_id,
channel_id=channel_id,
model_name=model_alloc.modelName,
litellm_key_id=key.key,
litellm_key_hash=encrypted_key,
rpm_limit=model_alloc.rpm,
tpm_limit=model_alloc.tpm,
max_budget=DEFAULT_MAX_BUDGET,
budget_duration=DEFAULT_BUDGET_DURATION,
status="active",
)
db.add(tenant_key)
logger.info(f"为租户 {tenant.name} 创建模型 {model_alloc.modelName} 的 LiteLLM Key(默认预算 ${DEFAULT_MAX_BUDGET}/月)")
except LiteLLMClientError as e:
# LiteLLM 操作失败,整个分配操作也失败
# 注意:不能把 str(e) 回给客户端 —— 上下文里有 INSERT 进
# tenant_model_keys 的 SQL params,含明文 litellm_key_id。
logger.error(f"LiteLLM 操作失败,资源分配已取消", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"code": "LITELLM_KEY_OP_FAILED",
"message": f"模型 '{model_alloc.modelName}' 的 LiteLLM Key 操作失败,资源分配已取消"},
)
except Exception:
# LiteLLM 连接失败,整个分配操作也失败
logger.error("LiteLLM 连接失败,资源分配已取消", exc_info=True)
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail={"code": "LITELLM_UNAVAILABLE",
"message": "无法连接 LiteLLM Gateway,资源分配已取消"},
)
else:
# 渠道未配置 LiteLLM team,无法分配模型资源
logger.error(f"渠道 {channel_id} 未配置 LiteLLM team,无法分配模型资源")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="渠道尚未关联 LiteLLM team,无法分配模型资源,请联系管理员"
)
# 同时记录到 ResourceAllocation(兼容旧逻辑)
# 查找模型供应商(可选,用于记录 resource_id)
result = await db.execute(
select(ModelProvider).where(ModelProvider.name == model_alloc.modelName)
)
model_provider = result.scalar_one_or_none()
allocation = ResourceAllocation(
target_id=tenant_id,
target_type="tenant",
resource_type="model",
resource_id=str(model_provider.id) if model_provider else model_alloc.modelName,
rpm=model_alloc.rpm,
tpm=model_alloc.tpm,
)
db.add(allocation)
# 分配自定义 Agent 配额
if req.customAgentQuota:
# 获取渠道的配额上限
channel_quota_result = await db.execute(
select(ChannelCustomAgentQuota).where(
ChannelCustomAgentQuota.channel_id == tenant_channel_id
)
)
channel_quota = channel_quota_result.scalar_one_or_none()
# Bug 修复:如果渠道没有自定义 Agent 配额,不允许分配给租户
if not channel_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="渠道没有自定义 Agent 配额,请先向管理员申请"
)
# 查找租户现有配额记录
tenant_quota_result = await db.execute(
select(TenantCustomAgentQuota).where(
TenantCustomAgentQuota.tenant_id == tenant_id
)
)
tenant_quota = tenant_quota_result.scalar_one_or_none()
# 获取租户当前配额
current_tenant_cpu = float(tenant_quota.cpu_quota) if tenant_quota else 0
current_tenant_memory = float(tenant_quota.memory_quota) if tenant_quota else 0
# 追加模式:请求值表示要追加的量
cpu_delta = req.customAgentQuota.cpuQuota
memory_delta = req.customAgentQuota.memoryQuota
# 计算新配额
new_cpu_quota = current_tenant_cpu + cpu_delta
new_memory_quota = current_tenant_memory + memory_delta
# 计算渠道已分配给其他租户的配额
other_tenants_quota_result = await db.execute(
select(
func.sum(TenantCustomAgentQuota.cpu_quota).label("total_cpu"),
func.sum(TenantCustomAgentQuota.memory_quota).label("total_memory")
)
.select_from(TenantCustomAgentQuota)
.join(User, TenantCustomAgentQuota.tenant_id == User.id)
.where(
and_(
User.channel_id == tenant_channel_id,
TenantCustomAgentQuota.tenant_id != tenant_id
)
)
)
other_quota = other_tenants_quota_result.first()
other_cpu = float(other_quota.total_cpu or 0) if other_quota else 0
other_memory = float(other_quota.total_memory or 0) if other_quota else 0
# 检查追加量是否超过渠道剩余配额
# 渠道剩余 = 渠道配额 - 其他租户已分配 - 当前租户已分配
remaining_cpu = float(channel_quota.cpu_quota) - other_cpu - current_tenant_cpu
remaining_memory = float(channel_quota.memory_quota) - other_memory - current_tenant_memory
if cpu_delta > remaining_cpu:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"CPU 配额超出渠道剩余配额。渠道剩余: {remaining_cpu:.2f} 核,请求追加: {cpu_delta:.2f} 核"
)
if memory_delta > remaining_memory:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"内存配额超出渠道剩余配额。渠道剩余: {remaining_memory:.2f} GB,请求追加: {memory_delta:.2f} GB"
)
if tenant_quota:
# 追加模式:在现有配额基础上增加
tenant_quota.cpu_quota = new_cpu_quota
tenant_quota.memory_quota = new_memory_quota
logger.info(
f"追加自定义 Agent 配额: tenant={tenant_id}, "
f"CPU 追加量={cpu_delta}, 新配额={new_cpu_quota}, "
f"内存 追加量={memory_delta}, 新配额={new_memory_quota}"
)
else:
# 创建新配额记录
tenant_quota = TenantCustomAgentQuota(
tenant_id=tenant_id,
cpu_quota=new_cpu_quota,
memory_quota=new_memory_quota,
cpu_used=0,
memory_used=0,
agent_count=0,
)
db.add(tenant_quota)
logger.info(
f"创建自定义 Agent 配额: tenant={tenant_id}, "
f"CPU={new_cpu_quota}, 内存={new_memory_quota}"
)
# Bug 修复:在查询之前 flush,确保数据库能看到刚才的更改
await db.flush()
# 更新渠道已分配量(channel_quota 已在前面验证存在)
# 重新计算渠道已分配总量
all_tenants_quota_result = await db.execute(
select(
func.sum(TenantCustomAgentQuota.cpu_quota).label("total_cpu"),
func.sum(TenantCustomAgentQuota.memory_quota).label("total_memory")
)
.select_from(TenantCustomAgentQuota)
.join(User, TenantCustomAgentQuota.tenant_id == User.id)
.where(User.channel_id == tenant_channel_id)
)
all_quota = all_tenants_quota_result.first()
# 直接使用查询结果,已经包含所有租户的配额(包括当前租户)
channel_quota.cpu_allocated = float(all_quota.total_cpu or 0) if all_quota else 0
channel_quota.memory_allocated = float(all_quota.total_memory or 0) if all_quota else 0
await db.commit()
return SuccessResponse(message="资源分配成功")
@router.put("/tenants/{tenant_id}/billing", response_model=SuccessResponse)
async def update_tenant_billing(
tenant_id: str,
req: UpdateBillingRequest,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
更新租户计费设置
权限:manage:billing (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:billing")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 更新计费设置
tenant.subscription_tier = req.subscriptionTier
tenant.discount = req.discount
await db.commit()
return SuccessResponse(message="计费设置更新成功")
@router.post("/tenants/{tenant_id}/recharge", response_model=SuccessResponse)
async def recharge_tenant(
tenant_id: str,
req: RechargeTenantRequest,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
为租户充值
权限:manage:billing (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:billing")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 使用行锁保护余额更新,防止并发充值问题
balance_result = await db.execute(
select(Balance)
.where(Balance.user_id == tenant_id)
.with_for_update() # 行锁
)
balance_obj = balance_result.scalar_one_or_none()
if balance_obj is None:
# 如果余额记录不存在,创建一个新的
balance_obj = Balance(user_id=tenant_id, eu_balance=0.0)
db.add(balance_obj)
await db.flush() # 确保记录创建
# 更新余额(使用 Balance 表)
old_balance = float(balance_obj.eu_balance)
new_balance = old_balance + req.amount
balance_obj.eu_balance = new_balance
# 创建充值记录
recharge = RechargeRecord(
user_id=tenant_id,
channel_id=channel_id,
amount=req.amount,
payment_method="channel_recharge",
status="success",
order_id=f"CH{datetime.utcnow().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:8]}",
completed_at=datetime.utcnow(),
)
db.add(recharge)
await db.commit()
await db.refresh(balance_obj)
return SuccessResponse(
data={
"tenantId": str(tenant.id),
"newBalance": float(balance_obj.eu_balance),
"rechargeAmount": req.amount,
}
)
@router.put("/tenants/{tenant_id}/credit", response_model=SuccessResponse)
async def set_tenant_credit_limit(
tenant_id: str,
req: SetCreditLimitRequest,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
设置租户授信额度
权限:manage:billing (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:billing")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 更新授信额度
tenant.credit_limit = req.creditLimit
await db.commit()
return SuccessResponse(
data={
"tenantId": str(tenant.id),
"creditLimit": float(req.creditLimit),
}
)
@router.delete("/tenants/{tenant_id}", response_model=SuccessResponse)
async def delete_tenant(
tenant_id: str,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
删除租户(软删除)
将租户状态标记为inactive,保留数据但禁止使用
权限:manage:tenants (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 检查租户是否还有余额(从 Balance 表获取)
balance_result = await db.execute(
select(Balance).where(Balance.user_id == tenant.id)
)
balance_obj = balance_result.scalar_one_or_none()
tenant_balance = float(balance_obj.eu_balance) if balance_obj else 0.0
if tenant_balance > 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"租户还有余额 {tenant_balance:.2f},请先处理余额后再删除"
)
# 软删除:标记为不活跃
tenant.status = "inactive"
await db.commit()
return SuccessResponse(
data={
"id": str(tenant.id),
"name": tenant.name,
},
message="租户已删除"
)
@router.put("/tenants/{tenant_id}/status", response_model=SuccessResponse)
async def update_tenant_status(
tenant_id: str,
req: UpdateTenantStatusRequest,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
更新租户状态
可设置状态:
- active: 正常使用
- inactive: 已停用(软删除)
- suspended: 暂停使用(临时停用,可恢复)
权限:manage:tenants (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
old_status = tenant.status
tenant.status = req.status
await db.commit()
return SuccessResponse(
data={
"tenantId": str(tenant.id),
"name": tenant.name,
"oldStatus": old_status,
"newStatus": req.status,
},
message=f"租户状态已更新为 {req.status}"
)
@router.put("/tenants/{tenant_id}/permissions", response_model=SuccessResponse)
async def update_tenant_permissions(
tenant_id: str,
req: UpdateTenantPermissionsRequest,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
更新租户权限
可配置的权限列表包括:
- use:platform_agents - 使用平台Agent
- use:custom_agents - 使用自定义Agent
- create:agents - 创建Agent
- read:billing - 查看计费信息
- export:data - 导出数据
权限:manage:tenants (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 有效权限列表
valid_permissions = {
"use:platform_agents",
"use:custom_agents",
"create:agents",
"read:billing",
"export:data",
}
# 验证权限
invalid_permissions = set(req.permissions) - valid_permissions
if invalid_permissions:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"无效的权限: {', '.join(invalid_permissions)}"
)
# 更新用户的permissions字段(假设User模型有permissions JSON字段)
# 如果没有该字段,可以存储在metadata或创建新表
tenant.permissions = req.permissions
await db.commit()
return SuccessResponse(
data={
"tenantId": str(tenant.id),
"name": tenant.name,
"permissions": req.permissions,
},
message="租户权限已更新"
)
@router.put("/tenants/{tenant_id}/password", response_model=SuccessResponse)
async def change_tenant_password(
tenant_id: str,
req: ChangeTenantPasswordRequest,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
管理员为租户重置密码
权限:manage:tenants (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:tenants")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 更新密码
new_hash = get_password_hash(req.newPassword)
tenant.password_hash = new_hash
tenant.hashed_password = new_hash # 兼容旧字段
await db.commit()
return SuccessResponse(
data={
"tenantId": str(tenant.id),
"name": tenant.name,
},
message="租户密码已重置"
)
# ============= 租户配额查询 =============
@router.get("/tenants/{tenant_id}/custom-agent-quota", response_model=SuccessResponse)
async def get_tenant_custom_agent_quota(
tenant_id: str,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取租户的自定义 Agent 配额使用情况
返回数据:
- cpuQuota: CPU 配额上限(核心数)
- memoryQuota: 内存配额上限(GB)
- cpuUsed: 已使用 CPU(核心数)
- memoryUsed: 已使用内存(GB)
- cpuRemaining: 剩余 CPU(核心数)
- memoryRemaining: 剩余内存(GB)
- agentCount: 已创建的自定义 Agent 数量
权限:view:resources (channel_admin, billing_admin, operations_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能查看该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 获取租户配额
quota_result = await db.execute(
select(TenantCustomAgentQuota).where(
TenantCustomAgentQuota.tenant_id == tenant_id
)
)
quota = quota_result.scalar_one_or_none()
if not quota:
# 如果没有配额记录,返回默认值
return SuccessResponse(
data={
"tenantId": tenant_id,
"tenantName": tenant.name,
"cpuQuota": 0,
"memoryQuota": 0,
"cpuUsed": 0,
"memoryUsed": 0,
"cpuRemaining": 0,
"memoryRemaining": 0,
"agentCount": 0,
}
)
cpu_quota = float(quota.cpu_quota or 0)
memory_quota = float(quota.memory_quota or 0)
cpu_used = float(quota.cpu_used or 0)
memory_used = float(quota.memory_used or 0)
return SuccessResponse(
data={
"tenantId": tenant_id,
"tenantName": tenant.name,
"cpuQuota": cpu_quota,
"memoryQuota": memory_quota,
"cpuUsed": cpu_used,
"memoryUsed": memory_used,
"cpuRemaining": max(0, cpu_quota - cpu_used),
"memoryRemaining": max(0, memory_quota - memory_used),
"agentCount": quota.agent_count or 0,
}
)
# ============= 租户模型分配(LiteLLM 集成)=============
@router.put("/tenants/{tenant_id}/models", response_model=SuccessResponse)
async def allocate_model_to_tenant(
tenant_id: str,
model_name: str = Query(..., description="模型名称,如 azure/gpt-4"),
rpm_limit: int = Query(60, ge=0, description="每分钟请求数限制"),
tpm_limit: int = Query(10000, ge=0, description="每分钟 Token 数限制"),
max_budget: float = Query(100.0, ge=0, description="最大预算"),
budget_duration: str = Query("monthly", pattern="^(monthly|total)$", description="预算周期"),
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
分配模型给租户(创建 LiteLLM Key)
在 LiteLLM 中为租户创建 API Key,绑定指定的模型和配额。
租户的 Agent 启动时会使用此 Key 访问模型。
权限:manage:resources (channel_admin, billing_admin, super_admin)
注意:
- 渠道必须先拥有该模型的权限(通过 ResourceAllocation 分配)
- 每个租户每个模型只能有一个 Key
- 超级管理员必须提供 channel_id 参数
"""
_verify_permission(principal, "manage:resources")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
tenant_result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = tenant_result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 获取渠道信息
channel_result = await db.execute(
select(Channel).where(Channel.id == channel_id)
)
channel = channel_result.scalar_one_or_none()
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="渠道不存在"
)
# 验证渠道是否有该模型的权限
model_allocation_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == channel_id,
ResourceAllocation.target_type == "channel",
ResourceAllocation.resource_type == "model",
ResourceAllocation.resource_id == model_name
)
)
)
model_allocation = model_allocation_result.scalar_one_or_none()
if not model_allocation:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"渠道没有模型 '{model_name}' 的权限"
)
# 检查租户是否已有该模型的 Key
existing_key_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == tenant_id,
TenantModelKey.model_name == model_name
)
)
)
existing_key = existing_key_result.scalar_one_or_none()
if existing_key:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"租户已有模型 '{model_name}' 的 Key,请使用更新配额接口"
)
# 检查渠道是否有 LiteLLM team_id
if not channel.litellm_team_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="渠道尚未关联 LiteLLM team,请联系管理员"
)
# 在 LiteLLM 中创建 Key
try:
from app.litellm_client import get_litellm_client, LiteLLMClientError
litellm_client = get_litellm_client()
key = await litellm_client.generate_key(
team_id=channel.litellm_team_id,
models=[model_name],
rpm_limit=rpm_limit,
tpm_limit=tpm_limit,
max_budget=max_budget,
budget_duration=budget_duration,
key_name=f"tenant-{tenant_id}-{model_name}",
metadata={
"tenant_id": str(tenant_id),
"tenant_name": tenant.name,
"channel_id": str(channel_id),
"channel_name": channel.name,
"model": model_name,
}
)
# 加密存储 Key
encrypted_key = litellm_client.encrypt_key(key.key)
# 保存到数据库
tenant_key = TenantModelKey(
tenant_id=tenant_id,
channel_id=channel_id,
model_name=model_name,
litellm_key_id=key.key,
litellm_key_hash=encrypted_key,
rpm_limit=rpm_limit,
tpm_limit=tpm_limit,
max_budget=max_budget,
budget_duration=budget_duration,
status="active",
)
db.add(tenant_key)
# 同时记录到 ResourceAllocation
tenant_allocation = ResourceAllocation(
target_id=tenant_id,
target_type="tenant",
resource_type="model",
resource_id=model_name,
rpm=rpm_limit,
tpm=tpm_limit,
)
db.add(tenant_allocation)
await db.commit()
logger.info(
f"为租户 {tenant.name} 分配模型 {model_name} 成功",
extra={"tenant_id": tenant_id, "model": model_name}
)
return SuccessResponse(
data={
"tenantId": str(tenant_id),
"tenantName": tenant.name,
"modelName": model_name,
"rpmLimit": rpm_limit,
"tpmLimit": tpm_limit,
"maxBudget": max_budget,
"budgetDuration": budget_duration,
"status": "active",
},
message=f"模型 '{model_name}' 分配成功"
)
except LiteLLMClientError:
logger.error("LiteLLM Key 创建失败", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"code": "LITELLM_KEY_CREATE_FAILED",
"message": "LiteLLM Key 创建失败"},
)
except Exception:
# 不回 str(e) —— SQL params 含明文 litellm_key_id
logger.error("模型分配失败", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"code": "MODEL_ASSIGN_FAILED",
"message": "模型分配失败"},
)
@router.delete("/tenants/{tenant_id}/models/{model_name}", response_model=SuccessResponse)
async def revoke_model_from_tenant(
tenant_id: str,
model_name: str,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
取消租户的模型分配(删除 LiteLLM Key)
删除租户在 LiteLLM 中的 API Key,租户将无法再使用该模型。
权限:manage:resources (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数
- 删除后租户正在运行的 Agent 将无法继续使用该模型
"""
_verify_permission(principal, "manage:resources")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
tenant_result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = tenant_result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 查找租户的模型 Key
key_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == tenant_id,
TenantModelKey.model_name == model_name
)
)
)
tenant_key = key_result.scalar_one_or_none()
if not tenant_key:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"租户没有模型 '{model_name}' 的分配记录"
)
# 在 LiteLLM 中删除 Key
litellm_error = None
try:
from app.litellm_client import get_litellm_client, LiteLLMClientError
litellm_client = get_litellm_client()
await litellm_client.delete_key(tenant_key.litellm_key_id)
logger.info(f"LiteLLM Key 删除成功: {tenant_key.litellm_key_id[:20]}...")
except LiteLLMClientError as e:
litellm_error = str(e)
logger.warning(f"LiteLLM Key 删除失败: {e}")
except Exception as e:
litellm_error = str(e)
logger.warning(f"LiteLLM 连接失败: {e}")
# 删除数据库记录
await db.delete(tenant_key)
# 删除 ResourceAllocation 记录
allocation_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == tenant_id,
ResourceAllocation.target_type == "tenant",
ResourceAllocation.resource_type == "model",
ResourceAllocation.resource_id == model_name
)
)
)
allocation = allocation_result.scalar_one_or_none()
if allocation:
await db.delete(allocation)
await db.commit()
response_data = {
"tenantId": str(tenant_id),
"tenantName": tenant.name,
"modelName": model_name,
}
if litellm_error:
response_data["litellmWarning"] = f"LiteLLM Key 删除失败: {litellm_error}"
return SuccessResponse(
data=response_data,
message=f"模型 '{model_name}' 分配已取消"
)
@router.put("/tenants/{tenant_id}/models/{model_name}/quota", response_model=SuccessResponse)
async def update_tenant_model_quota(
tenant_id: str,
model_name: str,
rpm_limit: Optional[int] = Query(None, ge=0, description="每分钟请求数限制"),
tpm_limit: Optional[int] = Query(None, ge=0, description="每分钟 Token 数限制"),
max_budget: Optional[float] = Query(None, ge=0, description="最大预算"),
budget_duration: Optional[str] = Query(None, pattern="^(monthly|total)$", description="预算周期"),
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
更新租户的模型配额(更新 LiteLLM Key)
更新租户在 LiteLLM 中的 API Key 配额,包括 RPM、TPM 和预算限制。
更新后立即生效,无需重启任何服务。
权限:manage:resources (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数
- 只更新提供的参数,未提供的参数保持不变
"""
_verify_permission(principal, "manage:resources")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
tenant_result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = tenant_result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 查找租户的模型 Key
key_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == tenant_id,
TenantModelKey.model_name == model_name
)
)
)
tenant_key = key_result.scalar_one_or_none()
if not tenant_key:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"租户没有模型 '{model_name}' 的分配记录"
)
# 在 LiteLLM 中更新 Key
try:
from app.litellm_client import get_litellm_client, LiteLLMClientError
litellm_client = get_litellm_client()
await litellm_client.update_key(
key=tenant_key.litellm_key_id,
rpm_limit=rpm_limit,
tpm_limit=tpm_limit,
max_budget=max_budget,
budget_duration=budget_duration,
)
# 更新数据库记录
if rpm_limit is not None:
tenant_key.rpm_limit = rpm_limit
if tpm_limit is not None:
tenant_key.tpm_limit = tpm_limit
if max_budget is not None:
tenant_key.max_budget = max_budget
if budget_duration is not None:
tenant_key.budget_duration = budget_duration
# 同时更新 ResourceAllocation
allocation_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == tenant_id,
ResourceAllocation.target_type == "tenant",
ResourceAllocation.resource_type == "model",
ResourceAllocation.resource_id == model_name
)
)
)
allocation = allocation_result.scalar_one_or_none()
if allocation:
if rpm_limit is not None:
allocation.rpm = rpm_limit
if tpm_limit is not None:
allocation.tpm = tpm_limit
await db.commit()
logger.info(
f"租户 {tenant.name} 的模型 {model_name} 配额更新成功",
extra={"tenant_id": tenant_id, "model": model_name}
)
return SuccessResponse(
data={
"tenantId": str(tenant_id),
"tenantName": tenant.name,
"modelName": model_name,
"rpmLimit": tenant_key.rpm_limit,
"tpmLimit": tenant_key.tpm_limit,
"maxBudget": float(tenant_key.max_budget) if tenant_key.max_budget else None,
"budgetDuration": tenant_key.budget_duration,
"status": tenant_key.status,
},
message="模型配额更新成功,立即生效"
)
except LiteLLMClientError:
logger.error("LiteLLM Key 更新失败", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"code": "LITELLM_KEY_UPDATE_FAILED",
"message": "LiteLLM Key 更新失败"},
)
except Exception:
# 不回 str(e) —— 上下文涉及 tenant_model_keys 更新,SQL params 有明文 key
logger.error("配额更新失败", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"code": "QUOTA_UPDATE_FAILED",
"message": "配额更新失败"},
)
@router.get("/tenants/{tenant_id}/models", response_model=SuccessResponse)
async def get_tenant_models(
tenant_id: str,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取租户的模型分配列表
返回租户已分配的所有模型及其配额信息。
权限:view:resources (channel_admin, billing_admin, operations_admin, super_admin)
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
tenant_result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = tenant_result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 获取租户的所有模型 Key
keys_result = await db.execute(
select(TenantModelKey).where(
TenantModelKey.tenant_id == tenant_id
)
)
keys = keys_result.scalars().all()
data = []
for key in keys:
data.append({
"modelName": key.model_name,
"rpmLimit": key.rpm_limit,
"tpmLimit": key.tpm_limit,
"maxBudget": float(key.max_budget) if key.max_budget else None,
"budgetDuration": key.budget_duration,
"status": key.status,
"createdAt": key.created_at.isoformat() if key.created_at else None,
"updatedAt": key.updated_at.isoformat() if key.updated_at else None,
})
return SuccessResponse(
data={
"tenantId": str(tenant_id),
"tenantName": tenant.name,
"models": data,
}
)
# ============= 管理员管理 =============
@router.post("/admins/create", response_model=SuccessResponse)
async def create_channel_admin(
req: CreateAdminRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
创建渠道下的管理员(渠道管理员可用)
可创建的角色:
- billing_admin: 计费管理员(渠道下的计费管理员,完整写入权限,可管理该渠道下的租户、计费操作)
- operations_admin: 运维管理员(渠道下的运维管理员,只读权限,仅查看和监控该渠道的数据)
注意:
- 渠道管理员创建管理员时,channel_id会自动设置为当前渠道
- 请求中的channelId字段会被忽略,使用当前渠道的ID
权限:manage:admins (channel_admin)
"""
_verify_permission(principal, "manage:admins")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 超级管理员可以通过请求参数指定channelId
if role == "super_admin" and hasattr(req, 'channelId') and req.channelId:
try:
channel_id = uuid.UUID(req.channelId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID,超级管理员创建管理员时需要提供channelId参数"
)
# 验证渠道存在
result = await db.execute(
select(Channel).where(Channel.id == channel_id)
)
channel = result.scalar_one_or_none()
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="渠道不存在"
)
# 检查邮箱是否已存在
result = await db.execute(
select(User).where(User.email == req.email)
)
if result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="邮箱已被使用"
)
# 创建管理员用户(自动关联到当前渠道)
password_hash = get_password_hash(req.password)
admin = User(
name=req.name,
email=req.email,
password_hash=password_hash,
hashed_password=password_hash, # 兼容
username=req.email.split("@")[0],
full_name=req.name,
role=req.role,
channel_id=channel_id, # 自动设置为当前渠道
status="active",
credit_limit=0,
)
db.add(admin)
await db.commit()
await db.refresh(admin)
return SuccessResponse(
data={
"id": str(admin.id),
"name": admin.name,
"email": admin.email,
"role": admin.role,
"channelId": str(channel_id),
},
message=f"管理员创建成功"
)
@router.get("/admins", response_model=SuccessResponse)
async def list_channel_admins(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取渠道下的管理员列表(渠道管理员可用)
权限:view:admins (channel_admin)
"""
_verify_permission(principal, "view:admins")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 超级管理员可以查看所有管理员,其他角色只能查看自己渠道的管理员
if role == "super_admin":
result = await db.execute(
select(User).where(
and_(
User.role.in_(["channel_admin", "billing_admin", "operations_admin"]),
User.status == "active"
)
)
)
elif channel_id:
result = await db.execute(
select(User).where(
and_(
User.channel_id == channel_id,
User.role.in_(["billing_admin", "operations_admin", "channel_admin"]),
User.status == "active"
)
)
)
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
admins = result.scalars().all()
data = [
{
"id": str(admin.id),
"name": admin.name or admin.full_name,
"email": admin.email,
"role": admin.role,
"status": admin.status,
"createdAt": admin.created_at.isoformat() if admin.created_at else None,
}
for admin in admins
]
return SuccessResponse(data={"admins": data})
# ============= 资源申请 =============
@router.post("/resources/apply", response_model=SuccessResponse)
async def apply_for_resources(
req: ResourceApplicationRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
申请资源(模型或Agent)
权限:view:applications (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:applications")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 超级管理员申请资源时需要提供channelId参数(通过请求体)
if role == "super_admin" and hasattr(req, 'channelId') and req.channelId:
try:
channel_id = uuid.UUID(req.channelId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID,超级管理员申请资源时需要提供channelId参数"
)
# 创建申请
application = Application(
channel_id=channel_id,
type=req.type,
model_name=req.modelName,
rpm=req.rpm,
tpm=req.tpm,
agent_type=req.agentType,
quantity=req.quantity,
reason=req.reason,
status="pending",
)
db.add(application)
await db.commit()
await db.refresh(application)
return SuccessResponse(
data={"id": str(application.id), "status": "pending"},
message="申请已提交,等待审批"
)
# ============= 计费统计 =============
@router.get("/billing/stats", response_model=SuccessResponse)
async def get_channel_billing_stats(
startTime: str = Query(...),
endTime: str = Query(...),
tenantName: Optional[str] = Query(None),
minCalls: Optional[int] = Query(None),
maxCalls: Optional[int] = Query(None),
export: Optional[str] = Query(None, pattern="^(excel|csv|pdf)$"),
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取渠道计费统计
权限:view:billing (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:billing")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 解析时间(移除时区信息,使用 naive datetime)
try:
start_dt = datetime.fromisoformat(startTime.replace("Z", "+00:00")).replace(tzinfo=None)
end_dt = datetime.fromisoformat(endTime.replace("Z", "+00:00")).replace(tzinfo=None)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的时间格式,请使用 ISO 8601 格式"
)
# 查询渠道下的租户
tenants_result = await db.execute(
select(User).where(User.channel_id == channel_id)
)
tenants = {str(t.id): t for t in tenants_result.scalars().all()}
tenant_ids = list(tenants.keys())
# 租户统计(从 AgentBillingRecord 统计)
tenant_stats_result = await db.execute(
select(
AgentBillingRecord.user_id.label("tenant_id"),
func.count(AgentBillingRecord.id).label("calls"),
func.sum(AgentBillingRecord.eu_consumed).label("total_eu"),
func.sum(AgentBillingRecord.cost).label("total_cost"),
)
.where(
and_(
AgentBillingRecord.user_id.in_(tenant_ids),
AgentBillingRecord.start_time >= start_dt,
AgentBillingRecord.start_time <= end_dt,
)
)
.group_by(AgentBillingRecord.user_id)
)
tenant_stats = []
for row in tenant_stats_result.all():
tenant = tenants.get(str(row.tenant_id))
if tenant:
tenant_stats.append({
"tenantId": str(row.tenant_id),
"tenantName": tenant.name,
"calls": row.calls,
"totalEU": float(row.total_eu or 0),
"totalCost": float(row.total_cost or 0),
})
# 调用记录(从 AgentBillingRecord 查询)
records_result = await db.execute(
select(AgentBillingRecord)
.where(
and_(
AgentBillingRecord.user_id.in_(tenant_ids),
AgentBillingRecord.start_time >= start_dt,
AgentBillingRecord.start_time <= end_dt,
)
)
.order_by(desc(AgentBillingRecord.start_time))
.limit(100)
)
call_records = []
for record in records_result.scalars().all():
tenant = tenants.get(str(record.user_id))
if tenant:
call_records.append({
"id": str(record.id),
"timestamp": record.start_time.isoformat() if record.start_time else record.created_at.isoformat(),
"tenantName": tenant.name,
"agentName": record.agent_name,
"duration": record.duration_seconds,
"eu": record.eu_consumed,
"cost": float(record.cost),
})
# 如果是导出请求
if export:
file_url = f"https://exports.taiji-ai.com/{channel_id}/{export}/billing_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}.{export}"
expires_at = (datetime.utcnow() + timedelta(hours=24)).isoformat()
return SuccessResponse(
data={
"fileUrl": file_url,
"format": export,
"expiresAt": expires_at,
}
)
return SuccessResponse(
data={
"tenantStats": tenant_stats,
"callRecords": call_records,
}
)
# ============= 供应商管理 =============
@router.get("/providers", response_model=SuccessResponse)
async def list_available_providers(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取所有可用的模型供应商列表(渠道管理员视图)
返回所有活跃的供应商,并标注该渠道是否已获得授权使用
权限:view:resources (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 超级管理员可以查看所有供应商,其他角色需要channel_id
if role == "super_admin":
channel_id = None # 超级管理员不需要channel_id限制
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 获取所有活跃的供应商
providers_result = await db.execute(
select(ModelProvider).where(ModelProvider.is_active == True)
)
providers = providers_result.scalars().all()
# 获取该渠道的供应商授权
access_result = await db.execute(
select(ChannelProviderAccess).where(
ChannelProviderAccess.channel_id == channel_id
)
)
access_map = {str(a.provider_id): a for a in access_result.scalars().all()}
# 获取该渠道的待审批申请
pending_result = await db.execute(
select(ProviderApplication).where(
and_(
ProviderApplication.channel_id == channel_id,
ProviderApplication.status == "pending"
)
)
)
pending_apps = {str(a.provider_id) for a in pending_result.scalars().all()}
data = []
for provider in providers:
provider_id = str(provider.id)
access = access_map.get(provider_id)
item = {
"id": provider_id,
"name": provider.name,
"provider": provider.provider,
"supportedModels": provider.supported_models,
"rpm": provider.rpm,
"tpm": provider.tpm,
"status": provider.status,
"hasAccess": access is not None and access.status == "active",
"accessStatus": access.status if access else None,
"rpmLimit": access.rpm_limit if access else None,
"tpmLimit": access.tpm_limit if access else None,
"pendingApplication": provider_id in pending_apps,
}
data.append(item)
return SuccessResponse(data={"providers": data})
@router.post("/providers/apply", response_model=SuccessResponse)
async def apply_for_provider(
req: ApplyProviderRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
申请使用模型供应商
渠道管理员可以申请使用某个模型供应商,需要管理员审批后才能使用
权限:view:applications (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:applications")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 超级管理员申请供应商时需要提供channelId参数(通过请求体)
if role == "super_admin" and hasattr(req, 'channelId') and req.channelId:
try:
channel_id = uuid.UUID(req.channelId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID"
)
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID,超级管理员申请供应商时需要提供channelId参数"
)
# 验证 providerId 是否是有效的 UUID
try:
provider_uuid = uuid.UUID(req.providerId)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的供应商ID格式,必须是有效的UUID"
)
# 检查供应商是否存在
provider_result = await db.execute(
select(ModelProvider).where(
and_(
ModelProvider.id == provider_uuid,
ModelProvider.is_active == True
)
)
)
provider = provider_result.scalar_one_or_none()
if not provider:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="供应商不存在或已停用"
)
# 检查是否已有授权
access_result = await db.execute(
select(ChannelProviderAccess).where(
and_(
ChannelProviderAccess.channel_id == channel_id,
ChannelProviderAccess.provider_id == req.providerId,
ChannelProviderAccess.status == "active"
)
)
)
if access_result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="您已获得该供应商的使用授权"
)
# 检查是否有待审批的申请
pending_result = await db.execute(
select(ProviderApplication).where(
and_(
ProviderApplication.channel_id == channel_id,
ProviderApplication.provider_id == req.providerId,
ProviderApplication.status == "pending"
)
)
)
if pending_result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="您已有一个待审批的申请"
)
# 创建申请
application = ProviderApplication(
channel_id=channel_id,
provider_id=req.providerId,
requested_rpm=req.requestedRpm,
requested_tpm=req.requestedTpm,
reason=req.reason,
status="pending",
)
db.add(application)
await db.commit()
await db.refresh(application)
return SuccessResponse(
data={
"id": str(application.id),
"providerId": req.providerId,
"providerName": provider.name,
"status": "pending",
},
message="申请已提交,等待管理员审批"
)
@router.get("/providers/applications", response_model=SuccessResponse)
async def list_provider_applications(
status_filter: Optional[str] = Query(None, alias="status", pattern="^(pending|approved|rejected)$"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取渠道的供应商申请列表
权限:view:applications (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:applications")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 超级管理员可以查看所有申请,其他角色只能查看自己渠道的申请
if role == "super_admin":
query = select(ProviderApplication)
elif channel_id:
query = select(ProviderApplication).where(
ProviderApplication.channel_id == channel_id
)
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 构建查询(如果还没有构建)
if role != "super_admin" and not channel_id:
query = select(ProviderApplication).where(
ProviderApplication.channel_id == channel_id
)
if status_filter:
query = query.where(ProviderApplication.status == status_filter)
query = query.order_by(desc(ProviderApplication.created_at))
result = await db.execute(query)
applications = result.scalars().all()
# 获取供应商信息
provider_ids = [str(a.provider_id) for a in applications]
providers_result = await db.execute(
select(ModelProvider).where(ModelProvider.id.in_(provider_ids))
)
providers_map = {str(p.id): p for p in providers_result.scalars().all()}
data = []
for app in applications:
provider = providers_map.get(str(app.provider_id))
data.append({
"id": str(app.id),
"providerId": str(app.provider_id),
"providerName": provider.name if provider else "未知",
"requestedRpm": app.requested_rpm,
"requestedTpm": app.requested_tpm,
"reason": app.reason,
"status": app.status,
"createdAt": app.created_at.isoformat(),
"reviewedAt": app.reviewed_at.isoformat() if app.reviewed_at else None,
"reviewReason": app.review_reason,
})
return SuccessResponse(data={"applications": data})
@router.get("/providers/access", response_model=SuccessResponse)
async def list_provider_access(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取渠道已授权的供应商列表
权限:view:resources (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
# 超级管理员可以查看所有供应商,其他角色需要channel_id
if role == "super_admin":
channel_id = None # 超级管理员不需要channel_id限制
elif not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 获取授权列表
result = await db.execute(
select(ChannelProviderAccess).where(
ChannelProviderAccess.channel_id == channel_id
)
)
access_list = result.scalars().all()
# 获取供应商信息
provider_ids = [str(a.provider_id) for a in access_list]
providers_result = await db.execute(
select(ModelProvider).where(ModelProvider.id.in_(provider_ids))
)
providers_map = {str(p.id): p for p in providers_result.scalars().all()}
data = []
for access in access_list:
provider = providers_map.get(str(access.provider_id))
data.append({
"id": str(access.id),
"providerId": str(access.provider_id),
"providerName": provider.name if provider else "未知",
"provider": provider.provider if provider else "unknown",
"supportedModels": provider.supported_models if provider else [],
"status": access.status,
"rpmLimit": access.rpm_limit,
"tpmLimit": access.tpm_limit,
"approvedAt": access.approved_at.isoformat() if access.approved_at else None,
"expiresAt": access.expires_at.isoformat() if access.expires_at else None,
})
return SuccessResponse(data={"accessList": data})
# ============= 平台 Agent 资源申请 =============
# 模板资源配置建议
TEMPLATE_RESOURCE_CONFIG = {
"echo_agent": {"cpuRequest": "100m", "cpuLimit": "500m", "memoryRequest": "128Mi", "memoryLimit": "512Mi"},
"chat_agent": {"cpuRequest": "200m", "cpuLimit": "1000m", "memoryRequest": "256Mi", "memoryLimit": "1Gi"},
"code_agent": {"cpuRequest": "500m", "cpuLimit": "2000m", "memoryRequest": "512Mi", "memoryLimit": "2Gi"},
"search_agent": {"cpuRequest": "200m", "cpuLimit": "1000m", "memoryRequest": "256Mi", "memoryLimit": "1Gi"},
"jina_search_agent": {"cpuRequest": "500m", "cpuLimit": "2000m", "memoryRequest": "1Gi", "memoryLimit": "4Gi"},
"mysql_agent": {"cpuRequest": "100m", "cpuLimit": "500m", "memoryRequest": "128Mi", "memoryLimit": "512Mi"},
"postgresql_agent": {"cpuRequest": "100m", "cpuLimit": "500m", "memoryRequest": "128Mi", "memoryLimit": "512Mi"},
}
async def _get_platform_templates_from_agent_manager(db: AsyncSession = None) -> Dict[str, Dict[str, Any]]:
"""
从 Agent Manager 获取平台模板列表,并合并管理员配置
Args:
db: 数据库会话,用于获取管理员配置。如果为 None,则不获取管理员配置
Returns:
模板字典,key 为模板名称,value 为模板信息
"""
try:
client = get_agent_manager_client()
templates = await client.list_platform_templates()
# 获取管理员配置(如果提供了数据库会话)
admin_configs = {}
if db:
config_result = await db.execute(
select(PlatformAgentTemplateConfig).where(
PlatformAgentTemplateConfig.is_enabled == True
)
)
for config in config_result.scalars().all():
admin_configs[config.template_name] = config
result = {}
for t in templates:
template_name = t.template
admin_config = admin_configs.get(template_name)
# 优先使用 Agent Manager 返回的 displayName 和 description
agent_manager_display_name = t.display_name
agent_manager_description = t.description
agent_manager_category = t.category
# 如果有管理员配置,使用管理员配置的值;否则返回 null/0
if admin_config:
resource_config = {
"cpuRequest": admin_config.cpu_request,
"cpuLimit": admin_config.cpu_limit,
"memoryRequest": admin_config.memory_request,
"memoryLimit": admin_config.memory_limit,
"maxPods": admin_config.max_pods or 0,
"isConfigured": True,
}
# 优先级:管理员配置 > Agent Manager 返回 > 模板名称
display_name = admin_config.display_name or agent_manager_display_name or template_name
description = admin_config.description or agent_manager_description or f"{template_name} Agent"
else:
# 未配置时返回 null/0,表示管理员尚未配置
resource_config = {
"cpuRequest": None,
"cpuLimit": None,
"memoryRequest": None,
"memoryLimit": None,
"maxPods": 0,
"isConfigured": False,
}
# 优先级:Agent Manager 返回 > 模板名称
display_name = agent_manager_display_name or template_name
description = agent_manager_description or f"{template_name} Agent"
result[template_name] = {
"name": template_name,
"displayName": display_name,
"description": description,
"category": agent_manager_category or "general",
"version": "1.0.0",
"port": t.port,
"envInfo": t.env_info,
"status": "available" if (admin_config and admin_config.is_enabled) else "not_configured",
**resource_config,
}
return result
except AgentManagerError as e:
logger.error("failed_to_get_platform_templates", error=str(e))
# 返回空字典,让调用方处理
return {}
except Exception as e:
logger.error("unexpected_error_getting_templates", error=str(e))
return {}
async def _get_custom_templates_from_agent_manager() -> Dict[str, Dict[str, Any]]:
"""
从 Agent Manager 获取自定义模板列表
Returns:
模板字典,key 为模板名称,value 为模板信息
"""
try:
client = get_agent_manager_client()
templates = await client.list_custom_templates()
result = {}
for t in templates:
template_name = t.template
resource_config = TEMPLATE_RESOURCE_CONFIG.get(template_name, {
"cpuRequest": "100m",
"cpuLimit": "500m",
"memoryRequest": "128Mi",
"memoryLimit": "512Mi"
})
# 优先使用 Agent Manager 返回的 displayName 和 description
agent_manager_display_name = t.display_name
agent_manager_description = t.description
agent_manager_category = t.category
result[template_name] = {
"name": template_name,
"displayName": agent_manager_display_name or template_name,
"description": agent_manager_description or f"{template_name} Agent",
"category": agent_manager_category or "general",
"version": "1.0.0",
"port": t.port,
"envInfo": t.env_info,
"requiredEnvVars": t.env_info.get("required", {}),
"optionalEnvVars": t.env_info.get("optional", {}),
"status": "available",
**resource_config,
}
return result
except AgentManagerError as e:
logger.error("failed_to_get_custom_templates", error=str(e))
return {}
except Exception as e:
logger.error("unexpected_error_getting_custom_templates", error=str(e))
return {}
@router.get("/available-platform-agents", response_model=SuccessResponse)
async def list_available_platform_agents(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
查看所有可用的平台 Agent 模板
渠道管理员可以查看平台上所有可用的 Agent 模板,
并查看自己是否已获得使用权限和配额情况。
模板数据从 Agent Manager 动态获取。
权限:view:resources (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if role != "super_admin" and not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 从 Agent Manager 获取平台模板(传递数据库会话以获取管理员配置)
platform_templates = await _get_platform_templates_from_agent_manager(db)
if not platform_templates:
logger.warning("no_platform_templates_available", channel_id=str(channel_id))
return SuccessResponse(
data={"templates": [], "warning": "无法从 Agent Manager 获取模板列表"}
)
# 获取渠道已有的平台 Agent 配额
quota_result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == channel_id,
PlatformAgentQuota.target_type == "channel"
)
)
)
quota_map = {q.template_name: q for q in quota_result.scalars().all()}
# 获取渠道下每个模板已分配给租户的配额总和(与 allocate_tenant_resources 保持一致)
tenant_quota_result = await db.execute(
select(
PlatformAgentQuota.template_name,
func.sum(PlatformAgentQuota.pod_quota).label("total_allocated")
)
.select_from(PlatformAgentQuota)
.join(User, PlatformAgentQuota.target_id == User.id)
.where(
and_(
User.channel_id == channel_id,
PlatformAgentQuota.target_type == "tenant"
)
)
.group_by(PlatformAgentQuota.template_name)
)
tenant_allocated_map = {row.template_name: row.total_allocated or 0 for row in tenant_quota_result.all()}
# 获取待审批的申请
pending_result = await db.execute(
select(ResourceApplication).where(
and_(
ResourceApplication.channel_id == channel_id,
ResourceApplication.resource_type == "platform_agent",
ResourceApplication.status == "pending"
)
)
)
pending_apps = {a.template_name for a in pending_result.scalars().all()}
data = []
for template_name, template in platform_templates.items():
quota = quota_map.get(template_name)
# 使用查询租户配额总和的方式计算剩余配额(与 allocate_tenant_resources 一致)
tenant_allocated = tenant_allocated_map.get(template_name, 0)
pod_remaining = (quota.pod_quota - tenant_allocated) if quota else 0
item = {
**template,
"hasAccess": quota is not None,
"podQuota": quota.pod_quota if quota else 0,
"podUsed": tenant_allocated, # 已分配给租户的配额总和
"podRemaining": pod_remaining,
"pendingApplication": template_name in pending_apps,
}
data.append(item)
return SuccessResponse(data={"templates": data})
@router.post("/applications/platform-agents", response_model=SuccessResponse)
async def apply_for_platform_agent(
req: ApplyPlatformAgentRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
申请平台 Agent
渠道管理员可以申请使用某个平台 Agent 模板,需要管理员审批后才能使用。
权限:view:applications (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:applications")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if role != "super_admin" and not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 从 Agent Manager 获取平台模板,验证模板是否存在
platform_templates = await _get_platform_templates_from_agent_manager()
if req.templateName not in platform_templates:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"平台 Agent 模板 '{req.templateName}' 不存在"
)
# 检查是否有待审批的申请
pending_result = await db.execute(
select(ResourceApplication).where(
and_(
ResourceApplication.channel_id == channel_id,
ResourceApplication.resource_type == "platform_agent",
ResourceApplication.template_name == req.templateName,
ResourceApplication.status == "pending"
)
)
)
if pending_result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="您已有一个待审批的申请"
)
# 创建申请
application = ResourceApplication(
channel_id=channel_id,
resource_type="platform_agent",
template_name=req.templateName,
requested_pod_quota=req.requestedPodQuota,
reason=req.reason,
status="pending",
)
db.add(application)
await db.commit()
await db.refresh(application)
template = platform_templates[req.templateName]
return SuccessResponse(
data={
"id": str(application.id),
"templateName": req.templateName,
"templateDisplayName": template.get("displayName", req.templateName),
"requestedPodQuota": req.requestedPodQuota,
"status": "pending",
},
message="申请已提交,等待管理员审批"
)
@router.get("/applications/platform-agents", response_model=SuccessResponse)
async def list_platform_agent_applications(
status_filter: Optional[str] = Query(None, alias="status", pattern="^(pending|approved|rejected)$"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
查看渠道的平台 Agent 申请列表
权限:view:applications (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:applications")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if role == "super_admin":
query = select(ResourceApplication).where(
ResourceApplication.resource_type == "platform_agent"
)
elif channel_id:
query = select(ResourceApplication).where(
and_(
ResourceApplication.channel_id == channel_id,
ResourceApplication.resource_type == "platform_agent"
)
)
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
if status_filter:
query = query.where(ResourceApplication.status == status_filter)
query = query.order_by(desc(ResourceApplication.created_at))
result = await db.execute(query)
applications = result.scalars().all()
# 获取渠道信息
channel_ids = list(set(str(a.channel_id) for a in applications))
channels_result = await db.execute(
select(Channel).where(Channel.id.in_(channel_ids))
)
channels_map = {str(c.id): c for c in channels_result.scalars().all()}
data = []
for app in applications:
channel = channels_map.get(str(app.channel_id))
data.append({
"id": str(app.id),
"channelId": str(app.channel_id),
"channelName": channel.name if channel else "未知",
"resourceType": app.resource_type,
"templateName": app.template_name,
"templateDisplayName": app.template_name,
"requestedPodQuota": app.requested_pod_quota,
"approvedPodQuota": app.approved_pod_quota,
"reason": app.reason,
"status": app.status,
"reviewReason": app.review_reason,
"reviewedAt": app.reviewed_at.isoformat() if app.reviewed_at else None,
"createdAt": app.created_at.isoformat(),
})
return SuccessResponse(data={"applications": data})
# ============= 平台 Agent 配额管理 =============
@router.get("/platform-agents", response_model=SuccessResponse)
async def list_channel_platform_agent_quotas(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
查看渠道的平台 Agent 配额
返回渠道已获得的所有平台 Agent 配额信息。
权限:view:resources (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if role != "super_admin" and not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 获取渠道的平台 Agent 配额
result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == channel_id,
PlatformAgentQuota.target_type == "channel"
)
)
)
quotas = result.scalars().all()
# 查询所有模板配置,用于获取管理员设置的CPU和内存
template_configs_result = await db.execute(select(PlatformAgentTemplateConfig))
template_configs = {config.template_name: config for config in template_configs_result.scalars().all()}
data = []
for quota in quotas:
# 从模板配置中获取管理员设置的CPU和内存限制
template_config = template_configs.get(quota.template_name)
cpu_limit = template_config.cpu_limit if template_config and template_config.cpu_limit else "100m"
memory_limit = template_config.memory_limit if template_config and template_config.memory_limit else "256Mi"
data.append({
"templateName": quota.template_name,
"templateDisplayName": quota.template_name,
"podQuota": quota.pod_quota,
"podUsed": quota.pod_used,
"podRemaining": quota.pod_quota - quota.pod_used,
"cpuLimit": cpu_limit,
"memoryLimit": memory_limit,
"allocatedAt": quota.allocated_at.isoformat() if quota.allocated_at else None,
})
return SuccessResponse(data={"quotas": data})
@router.post("/tenants/{tenant_id}/platform-agents", response_model=SuccessResponse)
async def allocate_platform_agent_to_tenant(
tenant_id: str,
req: AllocatePlatformAgentRequest,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
分配平台 Agent 给租户
渠道管理员可以将自己的平台 Agent 配额分配给租户。
权限:manage:resources (channel_admin, billing_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能操作该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "manage:resources")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
tenant_result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = tenant_result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 从 Agent Manager 获取平台模板,验证模板是否存在
platform_templates = await _get_platform_templates_from_agent_manager()
if req.templateName not in platform_templates:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"平台 Agent 模板 '{req.templateName}' 不存在"
)
# 获取渠道的配额(使用行锁防止并发更新)
channel_quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_id == channel_id,
PlatformAgentQuota.target_type == "channel",
PlatformAgentQuota.template_name == req.templateName
)
)
.with_for_update() # 行锁
)
channel_quota = channel_quota_result.scalar_one_or_none()
if not channel_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"渠道没有 '{req.templateName}' 的配额"
)
# 计算已分配给其他租户的配额
other_tenants_quota_result = await db.execute(
select(func.sum(PlatformAgentQuota.pod_quota).label("total"))
.select_from(PlatformAgentQuota)
.join(User, PlatformAgentQuota.target_id == User.id)
.where(
and_(
User.channel_id == channel_id,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.template_name == req.templateName,
PlatformAgentQuota.target_id != tenant_id
)
)
)
other_quota = other_tenants_quota_result.scalar() or 0
# 查找或创建租户配额记录(使用行锁防止并发更新)
tenant_quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_id == tenant_id,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.template_name == req.templateName
)
)
.with_for_update() # 行锁
)
tenant_quota = tenant_quota_result.scalar_one_or_none()
# 追加模式:req.podQuota 表示要追加的配额量
current_tenant_quota = tenant_quota.pod_quota if tenant_quota else 0
quota_delta = req.podQuota # 追加量
new_quota = current_tenant_quota + quota_delta # 新的总配额
# 检查追加量是否超过渠道剩余配额
# 渠道剩余 = 渠道配额 - 其他租户已分配 - 当前租户已分配
remaining = channel_quota.pod_quota - other_quota - current_tenant_quota
if quota_delta > remaining:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"配额超出渠道剩余配额。渠道剩余: {remaining},请求追加: {quota_delta}"
)
old_pod_used = tenant_quota.pod_used if tenant_quota else 0
if tenant_quota:
# 追加模式:在现有配额基础上增加
tenant_quota.pod_quota = new_quota
tenant_quota.allocated_at = datetime.utcnow()
logger.info(
f"追加平台 Agent 配额: template={req.templateName}, tenant={tenant_id}, "
f"追加量={quota_delta}, 新配额={new_quota}"
)
else:
# 注意:allocated_by 设置为 None,因为渠道管理员的 JWT sub 字段是渠道 ID 而非用户 ID
# 如果需要记录分配人,应该在 JWT 中添加实际的用户 ID 字段
tenant_quota = PlatformAgentQuota(
target_id=tenant_id,
target_type="tenant",
template_name=req.templateName,
pod_quota=new_quota,
pod_used=0,
allocated_by=None, # 渠道管理员的 sub 是渠道 ID,不是用户 ID
allocated_at=datetime.utcnow(),
)
db.add(tenant_quota)
logger.info(
f"创建平台 Agent 配额: template={req.templateName}, tenant={tenant_id}, "
f"配额={new_quota}"
)
# Bug 修复:更新渠道的 pod_used(分配给租户的配额视为渠道已使用的配额)
channel_quota.pod_used = (channel_quota.pod_used or 0) + quota_delta
# ========== 重要修改:不再自动启动 Pods ==========
# 平台Agent改为和自定义Agent一样,只分配配额,由用户自行在租户端部署
# 用户通过 POST /api/user/platform-agents/deploy 接口部署
await db.commit()
# 使用已获取的模板信息
template = platform_templates.get(req.templateName, {})
return SuccessResponse(
data={
"tenantId": tenant_id,
"tenantName": tenant.name,
"templateName": req.templateName,
"templateDisplayName": template.get("displayName", req.templateName),
"podQuota": new_quota,
"podUsed": tenant_quota.pod_used if tenant_quota else 0,
"podRemaining": new_quota - (tenant_quota.pod_used if tenant_quota else 0),
},
message=f"平台 Agent 配额追加成功(追加量: {quota_delta}),租户可在租户端部署"
)
@router.get("/tenants/{tenant_id}/platform-agents/usage", response_model=SuccessResponse)
async def get_tenant_platform_agent_usage(
tenant_id: str,
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
查看租户的平台 Agent 使用情况
权限:view:resources (channel_admin, billing_admin, operations_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数,且只能查看该渠道下的租户
- 其他管理员自动使用所属渠道
"""
_verify_permission(principal, "view:resources")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证租户存在且属于指定渠道
tenant_result = await db.execute(
select(User).where(
and_(
User.id == tenant_id,
User.channel_id == channel_id
)
)
)
tenant = tenant_result.scalar_one_or_none()
if not tenant:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
# 获取租户的平台 Agent 配额
result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == tenant_id,
PlatformAgentQuota.target_type == "tenant"
)
)
)
quotas = result.scalars().all()
data = []
for quota in quotas:
data.append({
"templateName": quota.template_name,
"templateDisplayName": quota.template_name,
"podQuota": quota.pod_quota,
"podUsed": quota.pod_used,
"podRemaining": quota.pod_quota - quota.pod_used,
"allocatedAt": quota.allocated_at.isoformat() if quota.allocated_at else None,
})
return SuccessResponse(
data={
"tenantId": tenant_id,
"tenantName": tenant.name,
"quotas": data,
}
)
# ============= Agent 计费统计 =============
@router.get("/agent-billing/stats", response_model=SuccessResponse)
async def get_channel_agent_billing_stats(
startTime: str = Query(..., description="开始时间 (ISO 8601)"),
endTime: str = Query(..., description="结束时间 (ISO 8601)"),
agentType: Optional[str] = Query(None, description="Agent 类型: platform/custom"),
templateName: Optional[str] = Query(None, description="模板名称(平台 Agent)"),
tenantId: Optional[str] = Query(None, description="租户 ID"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取渠道的 Agent 计费统计
返回渠道下所有租户的 Agent 使用费用统计,支持按类型、模板、租户筛选。
权限:view:billing (channel_admin, billing_admin, operations_admin)
"""
from app.billing import get_channel_agent_billing_stats
_verify_permission(principal, "view:billing")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if role != "super_admin" and not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 解析时间(移除时区信息,使用 naive datetime)
try:
start_dt = datetime.fromisoformat(startTime.replace("Z", "+00:00")).replace(tzinfo=None)
end_dt = datetime.fromisoformat(endTime.replace("Z", "+00:00")).replace(tzinfo=None)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的时间格式,请使用 ISO 8601 格式"
)
# 获取统计数据
stats = await get_channel_agent_billing_stats(
channel_id=str(channel_id),
start_date=start_dt,
end_date=end_dt,
db=db,
)
return SuccessResponse(data=stats)
@router.get("/agent-billing/history", response_model=SuccessResponse)
async def get_channel_agent_billing_history(
startTime: str = Query(..., description="开始时间 (ISO 8601)"),
endTime: str = Query(..., description="结束时间 (ISO 8601)"),
agentType: Optional[str] = Query(None, description="Agent 类型: platform/custom"),
templateName: Optional[str] = Query(None, description="模板名称"),
tenantId: Optional[str] = Query(None, description="租户 ID"),
page: int = Query(1, ge=1, description="页码"),
pageSize: int = Query(20, ge=1, le=100, description="每页数量"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取渠道的 Agent 计费历史记录
返回渠道下所有租户的 Agent 计费详细记录,支持分页和筛选。
权限:view:billing (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:billing")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if role != "super_admin" and not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 解析时间(移除时区信息,使用 naive datetime)
try:
start_dt = datetime.fromisoformat(startTime.replace("Z", "+00:00")).replace(tzinfo=None)
end_dt = datetime.fromisoformat(endTime.replace("Z", "+00:00")).replace(tzinfo=None)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的时间格式,请使用 ISO 8601 格式"
)
# 获取渠道下的租户
tenants_result = await db.execute(
select(User).where(User.channel_id == channel_id)
)
tenants = {str(t.id): t for t in tenants_result.scalars().all()}
tenant_ids = list(tenants.keys())
# 如果指定了租户 ID,验证是否属于该渠道
if tenantId:
if tenantId not in tenant_ids:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="租户不存在或不属于该渠道"
)
tenant_ids = [tenantId]
# 构建查询 - 使用 AgentBillingRecord
query = select(AgentBillingRecord).where(
and_(
AgentBillingRecord.user_id.in_(tenant_ids),
AgentBillingRecord.period_start >= start_dt,
AgentBillingRecord.period_start <= end_dt,
)
)
if agentType:
query = query.where(AgentBillingRecord.agent_type == agentType)
if templateName:
query = query.where(AgentBillingRecord.template_name == templateName)
# 获取总数
count_query = select(func.count()).select_from(query.subquery())
total_result = await db.execute(count_query)
total = total_result.scalar() or 0
# 分页查询
query = query.order_by(desc(AgentBillingRecord.period_start))
query = query.offset((page - 1) * pageSize).limit(pageSize)
result = await db.execute(query)
records = result.scalars().all()
data = []
for record in records:
tenant = tenants.get(str(record.user_id))
data.append({
"id": str(record.id),
"timestamp": record.period_start.isoformat() if record.period_start else None,
"tenantId": str(record.user_id),
"tenantName": tenant.name if tenant else "未知",
"agentType": record.agent_type,
"templateName": record.template_name,
"agentName": record.agent_name,
"durationSeconds": record.duration_seconds,
"cpuSeconds": record.cpu_seconds,
"memoryGbSeconds": record.memory_gb_seconds,
"requestCount": record.request_count,
"cost": float(record.cost),
"periodStart": record.period_start.isoformat() if record.period_start else None,
"periodEnd": record.period_end.isoformat() if record.period_end else None,
})
return SuccessResponse(
data={
"records": data,
"pagination": {
"page": page,
"pageSize": pageSize,
"total": total,
"totalPages": (total + pageSize - 1) // pageSize,
}
}
)
@router.get("/agent-billing/tenant-summary", response_model=SuccessResponse)
async def get_channel_tenant_agent_billing_summary(
startTime: str = Query(..., description="开始时间 (ISO 8601)"),
endTime: str = Query(..., description="结束时间 (ISO 8601)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取渠道下各租户的 Agent 计费汇总
返回每个租户的 Agent 使用费用汇总,便于渠道管理员了解各租户的使用情况。
权限:view:billing (channel_admin, billing_admin, operations_admin)
"""
_verify_permission(principal, "view:billing")
role = _get_role(principal)
channel_id = _get_channel_id(principal)
if role != "super_admin" and not channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 解析时间(移除时区信息,使用 naive datetime)
try:
start_dt = datetime.fromisoformat(startTime.replace("Z", "+00:00")).replace(tzinfo=None)
end_dt = datetime.fromisoformat(endTime.replace("Z", "+00:00")).replace(tzinfo=None)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的时间格式,请使用 ISO 8601 格式"
)
# 获取渠道下的租户
tenants_result = await db.execute(
select(User).where(User.channel_id == channel_id)
)
tenants = {str(t.id): t for t in tenants_result.scalars().all()}
tenant_ids = list(tenants.keys())
# 按租户统计 - 使用 AgentBillingRecord
tenant_stats_result = await db.execute(
select(
AgentBillingRecord.user_id,
AgentBillingRecord.agent_type,
func.count(AgentBillingRecord.id).label("count"),
func.sum(AgentBillingRecord.duration_seconds).label("total_duration"),
func.sum(AgentBillingRecord.request_count).label("total_requests"),
func.sum(AgentBillingRecord.cost).label("total_cost"),
)
.where(
and_(
AgentBillingRecord.user_id.in_(tenant_ids),
AgentBillingRecord.period_start >= start_dt,
AgentBillingRecord.period_start <= end_dt,
)
)
.group_by(AgentBillingRecord.user_id, AgentBillingRecord.agent_type)
)
# 整理数据
tenant_data = {}
for row in tenant_stats_result.all():
tenant_id = str(row.user_id)
if tenant_id not in tenant_data:
tenant = tenants.get(tenant_id)
tenant_data[tenant_id] = {
"tenantId": tenant_id,
"tenantName": tenant.name if tenant else "未知",
"platformAgent": {
"count": 0,
"totalDuration": 0,
"totalRequests": 0,
"totalCost": 0,
},
"customAgent": {
"count": 0,
"totalDuration": 0,
"totalRequests": 0,
"totalCost": 0,
},
"total": {
"count": 0,
"totalDuration": 0,
"totalRequests": 0,
"totalCost": 0,
},
}
agent_type_key = "platformAgent" if row.agent_type == "platform" else "customAgent"
tenant_data[tenant_id][agent_type_key] = {
"count": row.count,
"totalDuration": row.total_duration or 0,
"totalRequests": int(row.total_requests or 0),
"totalCost": float(row.total_cost or 0),
}
# 更新总计
tenant_data[tenant_id]["total"]["count"] += row.count
tenant_data[tenant_id]["total"]["totalDuration"] += row.total_duration or 0
tenant_data[tenant_id]["total"]["totalRequests"] += int(row.total_requests or 0)
tenant_data[tenant_id]["total"]["totalCost"] += float(row.total_cost or 0)
# 计算渠道总计
channel_total = {
"platformAgent": {"count": 0, "totalDuration": 0, "totalRequests": 0, "totalCost": 0},
"customAgent": {"count": 0, "totalDuration": 0, "totalRequests": 0, "totalCost": 0},
"total": {"count": 0, "totalDuration": 0, "totalRequests": 0, "totalCost": 0},
}
for tenant_stats in tenant_data.values():
for key in ["platformAgent", "customAgent", "total"]:
channel_total[key]["count"] += tenant_stats[key]["count"]
channel_total[key]["totalDuration"] += tenant_stats[key]["totalDuration"]
channel_total[key]["totalRequests"] += tenant_stats[key]["totalRequests"]
channel_total[key]["totalCost"] += tenant_stats[key]["totalCost"]
return SuccessResponse(
data={
"tenants": list(tenant_data.values()),
"channelTotal": channel_total,
"period": {
"startTime": start_dt.isoformat(),
"endTime": end_dt.isoformat(),
}
}
)
# ============= 渠道租户资源分配查看 =============
@router.get("/tenants/resources/summary", response_model=SuccessResponse)
async def get_channel_tenants_resources_summary(
channel_id_param: Optional[str] = Query(None, alias="channel_id", description="渠道ID(超级管理员必填)"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
查看渠道下所有租户被分配的资源汇总
返回渠道下每个租户的:
- 自定义 Agent 配额(cpu_quota, memory_quota, cpu_used, memory_used, agent_count)
- 平台 Agent 配额(各模板的 pod_quota, pod_used)
- 模型配额(各模型的 rpm, tpm)
权限:view:tenants (channel_admin, billing_admin, operations_admin, super_admin)
注意:
- 超级管理员必须提供 channel_id 参数
- 其他管理员自动使用自己所属的渠道
"""
_verify_permission(principal, "view:tenants")
role = _get_role(principal)
user_channel_id = _get_channel_id(principal)
# 确定目标渠道ID
if role == "super_admin":
if not channel_id_param:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="超级管理员必须提供 channel_id 参数"
)
try:
channel_id = uuid.UUID(channel_id_param)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的渠道ID格式"
)
elif user_channel_id:
channel_id = user_channel_id
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取渠道ID"
)
# 验证渠道存在
channel_result = await db.execute(
select(Channel).where(Channel.id == channel_id)
)
channel = channel_result.scalar_one_or_none()
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="渠道不存在"
)
# 获取渠道下所有租户
tenants_result = await db.execute(
select(User).where(User.channel_id == channel_id)
)
tenants = tenants_result.scalars().all()
# 获取所有租户的余额(从 Balance 表)
tenant_ids = [tenant.id for tenant in tenants]
balance_result = await db.execute(
select(Balance).where(Balance.user_id.in_(tenant_ids))
)
balances = {b.user_id: float(b.eu_balance) for b in balance_result.scalars().all()}
# 构建租户资源数据
tenants_resources = []
for tenant in tenants:
tenant_data = {
"tenantId": str(tenant.id),
"tenantName": tenant.name,
"tenantEmail": tenant.email,
"status": tenant.status,
"subscriptionTier": tenant.subscription_tier,
"balance": balances.get(tenant.id, 0.0), # 从 Balance 表获取余额
"createdAt": tenant.created_at.isoformat() if tenant.created_at else None,
}
# 1. 获取自定义 Agent 配额
custom_quota_result = await db.execute(
select(TenantCustomAgentQuota).where(
TenantCustomAgentQuota.tenant_id == tenant.id
)
)
custom_quota = custom_quota_result.scalar_one_or_none()
if custom_quota:
tenant_data["customAgentQuota"] = {
"cpuQuota": float(custom_quota.cpu_quota),
"memoryQuota": float(custom_quota.memory_quota),
"cpuUsed": float(custom_quota.cpu_used),
"memoryUsed": float(custom_quota.memory_used),
"agentCount": custom_quota.agent_count,
}
else:
tenant_data["customAgentQuota"] = None
# 2. 获取平台 Agent 配额
platform_quotas_result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == tenant.id,
PlatformAgentQuota.target_type == "tenant"
)
)
)
platform_quotas = platform_quotas_result.scalars().all()
tenant_data["platformAgents"] = [
{
"templateName": pq.template_name,
"podQuota": pq.pod_quota,
"podUsed": pq.pod_used,
"cpuPerPod": pq.cpu_per_pod,
"memoryPerPod": pq.memory_per_pod,
}
for pq in platform_quotas
]
# 3. 获取模型配额(从 TenantModelKey 获取)
model_keys_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == tenant.id,
TenantModelKey.status == "active"
)
)
)
model_keys = model_keys_result.scalars().all()
tenant_data["models"] = [
{
"modelName": mk.model_name,
"rpmLimit": mk.rpm_limit,
"tpmLimit": mk.tpm_limit,
"maxBudget": float(mk.max_budget) if mk.max_budget else None,
"budgetDuration": mk.budget_duration,
}
for mk in model_keys
]
tenants_resources.append(tenant_data)
# 汇总统计
summary = {
"totalTenants": len(tenants),
"tenantsWithCustomAgents": len([t for t in tenants_resources if t["customAgentQuota"]]),
"tenantsWithPlatformAgents": len([t for t in tenants_resources if t["platformAgents"]]),
"tenantsWithModels": len([t for t in tenants_resources if t["models"]]),
"totalCustomAgentCpuQuota": sum(
t["customAgentQuota"]["cpuQuota"] for t in tenants_resources if t["customAgentQuota"]
),
"totalCustomAgentMemoryQuota": sum(
t["customAgentQuota"]["memoryQuota"] for t in tenants_resources if t["customAgentQuota"]
),
"totalCustomAgentCpuUsed": sum(
t["customAgentQuota"]["cpuUsed"] for t in tenants_resources if t["customAgentQuota"]
),
"totalCustomAgentMemoryUsed": sum(
t["customAgentQuota"]["memoryUsed"] for t in tenants_resources if t["customAgentQuota"]
),
"totalPlatformAgentPodQuota": sum(
sum(pa["podQuota"] for pa in t["platformAgents"]) for t in tenants_resources
),
"totalPlatformAgentPodUsed": sum(
sum(pa["podUsed"] for pa in t["platformAgents"]) for t in tenants_resources
),
"totalModelsAllocated": sum(
len(t["models"]) for t in tenants_resources
),
}
# 获取渠道自身的资源配额,用于对比
channel_custom_quota_result = await db.execute(
select(ChannelCustomAgentQuota).where(
ChannelCustomAgentQuota.channel_id == channel_id
)
)
channel_custom_quota = channel_custom_quota_result.scalar_one_or_none()
channel_quota = {
"customAgentQuota": {
"cpuQuota": float(channel_custom_quota.cpu_quota) if channel_custom_quota else 0,
"memoryQuota": float(channel_custom_quota.memory_quota) if channel_custom_quota else 0,
"cpuAllocated": float(channel_custom_quota.cpu_allocated) if channel_custom_quota else 0,
"memoryAllocated": float(channel_custom_quota.memory_allocated) if channel_custom_quota else 0,
} if channel_custom_quota else None
}
# 获取渠道的平台 Agent 配额
channel_platform_quotas_result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == channel_id,
PlatformAgentQuota.target_type == "channel"
)
)
)
channel_platform_quotas = channel_platform_quotas_result.scalars().all()
channel_quota["platformAgents"] = [
{
"templateName": cpq.template_name,
"podQuota": cpq.pod_quota,
"podUsed": cpq.pod_used,
"cpuPerPod": cpq.cpu_per_pod,
"memoryPerPod": cpq.memory_per_pod,
}
for cpq in channel_platform_quotas
]
# 获取渠道分配的模型
channel_models_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == channel_id,
ResourceAllocation.target_type == "channel",
ResourceAllocation.resource_type == "model"
)
)
)
channel_models = channel_models_result.scalars().all()
channel_quota["models"] = [str(cm.resource_id) for cm in channel_models]
return SuccessResponse(
data={
"channelId": str(channel_id),
"channelName": channel.name,
"channelQuota": channel_quota,
"tenants": tenants_resources,
"summary": summary,
},
message="获取渠道租户资源分配成功"
)