taijibaga/taiji-AI-PAD

Fork 0

forked from xiaohei/taiji-AI-PAD

Files

T

zhanggangyong 40447f86f6 更新开发者平台

2026-03-16 04:05:44 +00:00

24 KiB

Raw Permalink Blame History

开发者平台 API 设计方案

版本：v1.3
日期：2026-03-16
状态：✅ 已实现

1. 概述

1.1 目标

为已注册的租户用户提供 API Key 认证方式访问现有的 /api/user/* 接口，使开发者能够通过程序化方式（而非 Web UI）使用平台能力。

1.2 核心需求

需求	说明
API Key 认证	支持通过 API Key 访问现有接口，与 JWT Token 认证并行
现有接口复用	不新增业务接口，复用现有 `/api/user/*` 接口
计费不变	继续使用现有 EU 计费模式

1.3 架构说明

┌─────────────────────────────────────────────────────────────────┐
│                        现有架构                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────┐                    ┌─────────────────────────┐ │
│  │   Web UI    │──JWT Token──────▶ │                         │ │
│  │   前端      │                    │    /api/user/*          │ │
│  └─────────────┘                    │    现有接口             │ │
│                                     │                         │ │
│  ┌─────────────┐                    │  - 平台 Agent 管理      │ │
│  │  开发者     │──API Key ────────▶│  - 自定义 Agent 管理    │ │
│  │  程序调用   │   [新增]           │  - 外部工具管理         │ │
│  └─────────────┘                    │  - 工作流管理           │ │
│                                     │  - 计费/使用量查询      │ │
│                                     └─────────────────────────┘ │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │                    Agent 直接调用                            ││
│  │                                                             ││
│  │  开发者 ──────▶ Agent 域名（domain_url）──────▶ Agent Pod   ││
│  │                 https://my-agent.taiji-ai.com               ││
│  └─────────────────────────────────────────────────────────────┘│
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

2. 需要开发的内容

2.1 新增功能

功能	说明	优先级
API Key 认证支持	修改 `require_auth` 依赖，支持 API Key 认证	P0
API Key 管理接口	新增创建/列表/删除 API Key 的接口	P0
限流中间件	基于 API Key 的请求限流	P1

2.2 不需要开发的内容

❌ Agent 调用代理接口（用户直接使用 domain_url 调用）
❌ LLM 调用接口（用户通过 Agent 或直接使用 LiteLLM）
❌ 新的业务接口（复用现有 /api/user/* 接口）

3. 现有接口分析

3.1 `/api/user/*` 接口清单（需要开放给 API Key 认证）

根据 services/mcp-server/app/routes/user.py 分析，共 42 个接口：

3.1.1 仪表盘与统计（4 个）

方法	路径	功能描述	开发者需要
GET	`/dashboard/stats`	获取仪表盘统计数据	✅
GET	`/dashboard/billing-overview`	获取计费概览	✅
GET	`/agents/activity`	获取 Agent 活动数据	✅
GET	`/tools/stats`	获取工具统计数据	✅

3.1.2 工具管理（3 个）

方法	路径	功能描述	开发者需要
GET	`/tools`	获取用户工具列表	✅
POST	`/tools/create`	创建工具	✅
PUT	`/tools/{tool_id}`	更新工具	✅

3.1.3 网关管理（4 个）

方法	路径	功能描述	开发者需要
POST	`/gateway/select`	选择网关类型	⚠️ 可选
POST	`/gateway/api/create`	创建网关 API	⚠️ 可选
GET	`/gateway/apis`	获取网关 API 列表	⚠️ 可选
GET	`/gateway/monitoring`	获取网关监控数据	⚠️ 可选

3.1.4 配额管理（1 个）

方法	路径	功能描述	开发者需要
GET	`/custom-agent-quota`	获取自定义 Agent 配额	✅

3.1.5 平台 Agent 管理（5 个）

方法	路径	功能描述	开发者需要
GET	`/platform-agents/available`	获取可用平台 Agent 模板	✅
POST	`/platform-agents/deploy`	部署平台 Agent	✅
POST	`/platform-agents/use`	使用/启动平台 Agent	✅
GET	`/platform-agents/instances`	获取用户的 Agent 实例列表	✅
GET	`/agents/platform`	获取平台 Agent 列表	✅

3.1.6 自定义 Agent 管理（8 个）

方法	路径	功能描述	开发者需要
GET	`/custom-agents/templates`	获取自定义 Agent 模板	✅
POST	`/custom-agents`	创建自定义 Agent	✅
DELETE	`/custom-agents/{name}`	删除自定义 Agent	✅
PUT	`/custom-agents/{name}/scale`	扩缩容自定义 Agent	✅
GET	`/custom-agents`	获取自定义 Agent 列表	✅
GET	`/custom-agents/{name}/logs`	获取 Agent 日志	✅
POST	`/custom-agents/{name}/restart`	重启 Agent	✅
POST	`/agents/deploy`	部署 Agent（通用）	✅

3.1.7 工作流管理（4 个）

方法	路径	功能描述	开发者需要
POST	`/workflows/create`	创建工作流	✅
GET	`/workflows`	获取工作流列表	✅
POST	`/workflows/{workflow_id}/run`	运行工作流	✅
DELETE	`/workflows/{workflow_id}`	删除工作流	✅

3.1.8 模型管理（3 个）

方法	路径	功能描述	开发者需要
GET	`/models`	获取用户模型列表	✅
GET	`/models/available`	获取可用模型列表	✅
GET	`/models/usage/stats`	获取模型使用统计	✅

3.1.9 计费管理（3 个）

方法	路径	功能描述	开发者需要
GET	`/billing/balance`	获取 EU 余额	✅
POST	`/billing/recharge`	充值（需要支付）	⚠️ 可选
GET	`/billing/history`	获取计费历史	✅

3.1.10 Agent 计费（2 个）

方法	路径	功能描述	开发者需要
GET	`/agent-billing/stats`	获取 Agent 计费统计	✅
GET	`/agent-billing/history`	获取 Agent 计费历史	✅

3.1.11 用户资料（2 个）

方法	路径	功能描述	开发者需要
GET	`/profile`	获取用户资料	✅
PUT	`/profile`	更新用户资料	⚠️ 可选

3.1.12 资源信息（2 个）

方法	路径	功能描述	开发者需要
GET	`/resources/info`	获取用户资源信息	✅
GET	`/resources/agents`	获取用户 Agent 资源	✅

3.2 `/api/user/external-tools/*` 接口清单

根据 services/mcp-server/app/routes/external_tools.py 分析，共 7 个接口：

方法	路径	功能描述	开发者需要
POST	`/external-tools`	创建外部数据工具	✅
POST	`/external-tools/upload`	上传 JSON 文件创建工具	✅
GET	`/external-tools`	获取工具列表	✅
GET	`/external-tools/{tool_id}`	获取工具详情	✅
PUT	`/external-tools/{tool_id}`	更新工具配置	✅
DELETE	`/external-tools/{tool_id}`	删除工具	✅
POST	`/external-tools/{tool_id}/test`	测试工具连接	✅

3.3 `/api/user/toolkits/*` 接口清单

根据 services/mcp-server/app/routes/external_tools.py 分析，共 5 个接口：

方法	路径	功能描述	开发者需要
POST	`/toolkits`	创建工具集	✅
GET	`/toolkits`	获取工具集列表	✅
GET	`/toolkits/{toolkit_id}`	获取工具集详情	✅
PUT	`/toolkits/{toolkit_id}`	更新工具集	✅
DELETE	`/toolkits/{toolkit_id}`	删除工具集	✅

3.4 `/api/auth/keys/*` 接口清单（需要新增）

方法	路径	功能描述	状态
GET	`/keys/info`	获取 API 密钥信息	✅ 已有
POST	`/keys/regenerate`	重新生成 API 密钥	✅ 已有
GET	`/keys`	获取密钥列表	❌ 需新增
POST	`/keys`	创建新密钥	❌ 需新增
DELETE	`/keys/{key_id}`	删除密钥	❌ 需新增

4. 接口汇总

4.1 需要开放的现有接口（54 个）

模块	接口数量	说明
仪表盘与统计	4	全部开放
工具管理	3	全部开放
网关管理	4	可选开放
配额管理	1	全部开放
平台 Agent 管理	5	全部开放
自定义 Agent 管理	8	全部开放
工作流管理	4	全部开放
模型管理	3	全部开放
计费管理	3	全部开放
Agent 计费	2	全部开放
用户资料	2	可选开放
资源信息	2	全部开放
外部数据工具	7	全部开放
工具集	5	全部开放
合计	54

4.2 需要新增的接口（3 个）

模块	接口数量	说明
API Key 管理	3	列表/创建/删除

5. 实现方案

5.1 认证模块修改

修改 services/mcp-server/app/auth.py：

# 现有的 require_auth 依赖需要修改为支持双重认证

async def require_auth(
    authorization: Optional[str] = Header(None),
    x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
    db: AsyncSession = Depends(get_db)
) -> dict:
    """
    认证依赖，支持 JWT Token 和 API Key 两种方式
    
    认证方式：
    1. Authorization: Bearer <jwt_token>  - JWT Token 认证
    2. Authorization: Bearer sk-xxx       - API Key 认证
    3. X-API-Key: sk-xxx                  - API Key 认证
    """
    # 1. 尝试从 Authorization 头获取
    if authorization and authorization.startswith("Bearer "):
        token = authorization[7:]
        
        # 判断是 JWT 还是 API Key
        if token.startswith("sk-"):
            # API Key 认证
            return await verify_api_key(token, db)
        else:
            # JWT Token 认证（现有逻辑）
            return await verify_jwt_token(token)
    
    # 2. 尝试从 X-API-Key 头获取
    if x_api_key and x_api_key.startswith("sk-"):
        return await verify_api_key(x_api_key, db)
    
    # 3. 认证失败
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="未认证或认证失败"
    )


async def verify_api_key(api_key: str, db: AsyncSession) -> dict:
    """
    验证 API Key 并返回用户信息
    """
    # 提取前缀用于快速查找
    prefix = api_key[:8]
    
    # 查询匹配的 API Key
    result = await db.execute(
        select(APIKey)
        .where(APIKey.api_key_prefix == prefix)
        .where(APIKey.is_active == True)
    )
    key_record = result.scalar_one_or_none()
    
    if not key_record:
        raise HTTPException(status_code=401, detail="无效的 API Key")
    
    # 验证完整 Key（使用 bcrypt）
    if not verify_password(api_key, key_record.api_key_hash):
        raise HTTPException(status_code=401, detail="无效的 API Key")
    
    # 检查过期时间
    if key_record.expires_at and key_record.expires_at < datetime.utcnow():
        raise HTTPException(status_code=401, detail="API Key 已过期")
    
    # 更新最后使用时间和请求计数
    key_record.last_used = datetime.utcnow()
    key_record.total_requests = (key_record.total_requests or 0) + 1
    await db.commit()
    
    # 查询用户信息
    user_result = await db.execute(
        select(User).where(User.id == key_record.user_id)
    )
    user = user_result.scalar_one_or_none()
    
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    
    # 返回与 JWT 认证相同格式的 principal
    return {
        "user_id": str(user.id),
        "email": user.email,
        "role": user.role,
        "channel_id": str(user.channel_id) if user.channel_id else None,
        "claims": {
            "sub": str(user.id),
            "email": user.email,
            "role": user.role,
            "channelId": str(user.channel_id) if user.channel_id else None,
        },
        "auth_type": "api_key",
        "api_key_id": str(key_record.id),
    }

5.2 API Key 管理接口

在 services/mcp-server/app/routes/auth.py 新增：

@router.get("/keys", response_model=SuccessResponse)
async def list_api_keys(
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """获取用户的 API 密钥列表"""
    user_id = principal.get("user_id")
    
    result = await db.execute(
        select(APIKey)
        .where(APIKey.user_id == user_id)
        .order_by(APIKey.created_at.desc())
    )
    keys = result.scalars().all()
    
    return SuccessResponse(
        data={
            "keys": [
                {
                    "id": str(key.id),
                    "name": key.name or "默认密钥",
                    "prefix": key.api_key_prefix + "...",
                    "is_active": key.is_active,
                    "created_at": key.created_at.isoformat(),
                    "last_used": key.last_used.isoformat() if key.last_used else None,
                    "expires_at": key.expires_at.isoformat() if key.expires_at else None,
                    "total_requests": key.total_requests or 0,
                }
                for key in keys
            ]
        }
    )


@router.post("/keys", response_model=SuccessResponse)
async def create_api_key(
    name: str = Query(default="API Key", description="密钥名称"),
    expires_in_days: Optional[int] = Query(default=None, description="过期天数，不填则永不过期"),
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    创建新的 API 密钥
    
    注意：密钥只在创建时显示一次，请妥善保管
    """
    user_id = principal.get("user_id")
    
    # 生成新密钥
    raw_key = f"sk-{secrets.token_urlsafe(32)}"
    key_hash = get_password_hash(raw_key)
    
    expires_at = None
    if expires_in_days:
        expires_at = datetime.utcnow() + timedelta(days=expires_in_days)
    
    new_key = APIKey(
        user_id=user_id,
        api_key_hash=key_hash,
        api_key_prefix=raw_key[:8],
        name=name,
        key_hash=key_hash,
        prefix=raw_key[:8],
        is_active=True,
        expires_at=expires_at,
    )
    db.add(new_key)
    await db.commit()
    await db.refresh(new_key)
    
    return SuccessResponse(
        data={
            "id": str(new_key.id),
            "name": name,
            "key": raw_key,  # 只在创建时返回完整密钥
            "prefix": raw_key[:8] + "...",
            "expires_at": expires_at.isoformat() if expires_at else None,
        },
        message="API 密钥创建成功，请妥善保管，密钥只显示一次"
    )


@router.delete("/keys/{key_id}", response_model=SuccessResponse)
async def delete_api_key(
    key_id: str,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """删除 API 密钥"""
    user_id = principal.get("user_id")
    
    try:
        key_uuid = uuid.UUID(key_id)
    except ValueError:
        raise HTTPException(status_code=400, detail="无效的密钥 ID")
    
    result = await db.execute(
        select(APIKey)
        .where(APIKey.id == key_uuid)
        .where(APIKey.user_id == user_id)
    )
    key = result.scalar_one_or_none()
    
    if not key:
        raise HTTPException(status_code=404, detail="密钥不存在")
    
    await db.delete(key)
    await db.commit()
    
    return SuccessResponse(
        data={"id": key_id},
        message="API 密钥已删除"
    )

6. 实现任务清单

Phase 1: 认证扩展（P0）✅ 已完成

修改 app/auth.py 中的 require_auth 函数
- 添加 API Key 检测逻辑
- 实现 verify_api_key() 函数（已存在，已增强）
- 确保返回格式与 JWT 认证一致
更新 authenticate_request 函数（中间件使用）

Phase 2: API Key 管理接口（P0）✅ 已完成

在 app/routes/auth.py 新增接口
- GET /api/auth/keys - 获取密钥列表
- POST /api/auth/keys - 创建新密钥
- DELETE /api/auth/keys/{key_id} - 删除密钥

Phase 3: 限流中间件（P1）✅ 已完成

实现基于 API Key 的限流
- 每分钟请求数限制（默认 60）
- 每日请求数限制（默认 10000）
- 返回限流响应头
创建 app/rate_limiter.py 模块
在 app/application.py 中注册中间件

Phase 4: 文档与测试 ✅ 已完成

更新 OpenAPI 文档（自动生成）
编写 API 使用指南
创建测试脚本 test_developer_api.py

7. 使用示例

7.1 创建 API Key

# 使用 JWT Token 登录后创建 API Key
curl -X POST "https://api.taiji-ai.com/api/auth/keys?name=MyAppKey" \
  -H "Authorization: Bearer <jwt_token>"

响应：

{
  "success": true,
  "data": {
    "id": "key_abc123",
    "name": "MyAppKey",
    "key": "sk-a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0",
    "prefix": "sk-a1b2c...",
    "expires_at": null
  },
  "message": "API 密钥创建成功，请妥善保管，密钥只显示一次"
}

7.2 使用 API Key 调用接口

# 部署平台 Agent
curl -X POST "https://api.taiji-ai.com/api/user/platform-agents/deploy" \
  -H "Authorization: Bearer sk-a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0" \
  -H "Content-Type: application/json" \
  -d '{
    "template": "code-reviewer",
    "name": "my-code-reviewer"
  }'

响应：

{
  "success": true,
  "data": {
    "name": "my-code-reviewer",
    "status": "Running",
    "domain": "my-code-reviewer.taiji-ai.com",
    "domainUrl": "https://my-code-reviewer.taiji-ai.com"
  }
}

7.3 直接调用 Agent

# 使用返回的域名直接调用 Agent（不经过 MCP Server）
curl -X POST "https://my-code-reviewer.taiji-ai.com/review" \
  -H "Content-Type: application/json" \
  -d '{
    "code": "def hello():\n    print(\"Hello, World!\")",
    "language": "python"
  }'

7.4 查询使用量

# 获取 EU 余额
curl -X GET "https://api.taiji-ai.com/api/user/billing/balance" \
  -H "Authorization: Bearer sk-a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0"

# 获取计费历史
curl -X GET "https://api.taiji-ai.com/api/user/billing/history?page=1&page_size=20" \
  -H "Authorization: Bearer sk-a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0"

8. 安全考虑

8.1 API Key 安全

API Key 使用 bcrypt 哈希存储
只在创建时显示完整 Key
支持设置过期时间
支持随时删除/禁用

8.2 限流保护

默认每分钟 60 次请求
默认每日 10,000 次请求
超限返回 429 状态码

8.3 审计日志

记录所有 API Key 的使用
记录创建/删除操作

更新日志

日期	版本	变更内容
2026-03-16	v1.0	初始设计方案
2026-03-16	v1.1	简化方案：复用现有接口，只扩展 API Key 认证
2026-03-16	v1.2	详细分析现有接口，确定开放清单
2026-03-16	v1.3	✅ 功能实现完成

实现说明

已实现功能

✅ 认证扩展

修改了 app/auth.py 中的 require_auth 和 authenticate_request 函数
支持从 Authorization: Bearer sk-xxx 自动识别 API Key
支持从 X-API-Key: sk-xxx 识别 API Key
与 JWT Token 认证返回相同格式的 principal

✅ API Key 管理接口

GET /api/auth/keys - 获取用户所有密钥列表，包含使用统计
POST /api/auth/keys - 创建新密钥，支持设置名称和过期时间
DELETE /api/auth/keys/{key_id} - 删除指定密钥

✅ 限流中间件

创建了 app/rate_limiter.py 模块
实现了基于内存的滑动窗口限流算法
每分钟 60 次请求限制
每日 10,000 次请求限制
自动返回限流响应头（X-RateLimit-*）
超限返回 429 状态码和重试时间

✅ 文档与测试

创建了完整的使用指南：Docs/开发者平台API使用指南.md
创建了测试脚本：services/mcp-server/test_developer_api.py
OpenAPI 文档自动包含新接口

实现文件清单

文件	说明	状态
`services/mcp-server/app/auth.py`	扩展认证函数支持 API Key	✅ 已修改
`services/mcp-server/app/routes/auth.py`	新增 API Key 管理接口	✅ 已修改
`services/mcp-server/app/rate_limiter.py`	限流中间件实现	✅ 已创建
`services/mcp-server/app/application.py`	注册限流中间件	✅ 已修改
`services/mcp-server/test_developer_api.py`	功能测试脚本	✅ 已创建
`Docs/开发者平台API使用指南.md`	用户使用文档	✅ 已创建

测试方式

启动服务
```
cd services/mcp-server
python main.py
```

运行测试脚本

# 安装依赖
pip install httpx

# 运行测试
python test_developer_api.py

# 或指定服务器地址
python test_developer_api.py http://localhost:8000

手动测试

# 登录获取 JWT Token
curl -X POST "http://localhost:8000/api/auth/login" \
  -H "Content-Type: application/json" \
  -d '{"email": "test@example.com", "password": "test123456", "role": "user"}'

# 创建 API Key
curl -X POST "http://localhost:8000/api/auth/keys?name=TestKey" \
  -H "Authorization: Bearer <jwt_token>"

# 使用 API Key 调用接口
curl -X GET "http://localhost:8000/api/user/profile" \
  -H "Authorization: Bearer sk-xxx"

技术要点

认证流程
- 优先检查 X-API-Key header
- 其次检查 Authorization: Bearer header
- 如果 token 以 sk- 开头，识别为 API Key
- 否则作为 JWT Token 处理
API Key 验证
- 使用 bcrypt 哈希存储
- 通过前 8 个字符快速查找
- 验证完整 Key 的哈希值
- 检查是否过期和是否激活
- 更新最后使用时间和请求计数
限流实现
- 使用滑动窗口算法（内存实现）
- 自动清理过期的时间戳
- 按日期重置每日计数
- 返回详细的限流信息
安全考虑
- API Key 只在创建时显示一次
- 支持设置过期时间
- 支持随时禁用/删除
- 记录使用统计和审计日志

后续优化建议

📋 未来可选功能

支持 Redis 作为限流后端（分布式部署）
支持自定义限流配额（不同用户不同限制）
API Key 权限范围控制（scopes）
支持 IP 白名单
Webhook 回调（Key 过期提醒）
使用统计仪表板

24 KiB Raw Permalink Blame History Unescape Escape