feat: 添加API密钥管理功能,可通过 API_KEY 代替 Token 访问接口

- 引入了用于API密钥管理的新路由,包含列出、创建、更新、删除和重新生成API密钥的端点。
- 在后端实现了API密钥生成逻辑和验证。
- 增强了认证中间件以支持API密钥认证以及JWT认证。
- 创建了用于在前端管理API密钥的新Vue组件,包括用于创建、显示和删除密钥的用户界面。
This commit is contained in:
Wenjie Zhang 2026-03-23 09:27:01 +08:00
parent 927da60836
commit 74229c05a4
12 changed files with 1162 additions and 15 deletions

2
.gitignore vendored
View File

@ -51,7 +51,7 @@ cache
*.local.py
*.local.js
*.local.yaml
*.local/*
*.pdf
src/data

View File

@ -75,6 +75,9 @@ class User(Base):
# 关联部门
department = relationship("Department", back_populates="users")
# 关联 API Keys
api_keys = relationship("APIKey", back_populates="user", cascade="all, delete-orphan")
def to_dict(self, include_password: bool = False) -> dict[str, Any]:
result = {
"id": self.id,
@ -592,6 +595,53 @@ class SubAgent(Base):
return spec
class APIKey(Base):
"""API Key 模型"""
__tablename__ = "api_keys"
id = Column(Integer, primary_key=True, autoincrement=True)
key_hash = Column(String(64), nullable=False, unique=True, index=True)
key_prefix = Column(String(16), nullable=False)
name = Column(String(100), nullable=False)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
department_id = Column(Integer, ForeignKey("departments.id"), nullable=True, index=True)
expires_at = Column(DateTime, nullable=True)
is_enabled = Column(Boolean, nullable=False, default=True)
last_used_at = Column(DateTime, nullable=True)
created_by = Column(String(64), nullable=False)
created_at = Column(DateTime, default=utc_now_naive)
# 关联
user = relationship("User", back_populates="api_keys")
department = relationship("Department")
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"key_prefix": self.key_prefix,
"name": self.name,
"user_id": self.user_id,
"department_id": self.department_id,
"expires_at": format_utc_datetime(self.expires_at),
"is_enabled": bool(self.is_enabled),
"last_used_at": format_utc_datetime(self.last_used_at),
"created_by": self.created_by,
"created_at": format_utc_datetime(self.created_at),
}
def is_valid(self) -> bool:
"""检查 Key 是否有效"""
if not self.is_enabled:
return False
if self.expires_at and utc_now_naive() > self.expires_at:
return False
return True
class AgentRun(Base):
"""AgentRun table - 运行任务表"""

View File

@ -12,6 +12,7 @@ from server.routers.subagent_router import subagents_router
from server.routers.system_router import system
from server.routers.task_router import tasks
from server.routers.tool_router import tools
from server.routers.apikey_router import apikey_router
_LITE_MODE = os.environ.get("LITE_MODE", "").lower() in ("true", "1")
@ -28,6 +29,7 @@ router.include_router(mcp) # /api/system/mcp-servers/*
router.include_router(skills) # /api/system/skills/*
router.include_router(subagents_router) # /api/system/subagents/*
router.include_router(tools) # /api/system/tools/*
router.include_router(apikey_router) # /api/apikey/*
if not _LITE_MODE:
from server.routers.graph_router import graph

View File

@ -0,0 +1,223 @@
"""API Key 管理路由"""
import hashlib
import secrets
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from yuxi.storage.postgres.models_business import User, APIKey
from server.utils.auth_middleware import get_db, get_required_user, get_superadmin_user
from yuxi.utils.datetime_utils import coerce_any_to_utc_datetime, utc_now_naive
apikey_router = APIRouter(prefix="/apikey", tags=["apikey"])
def generate_api_key() -> tuple[str, str, str]:
"""生成新的 API Key
Returns: (full_key, key_hash, key_prefix)
- full_key: 完整密钥仅在创建时返回一次
- key_hash: 存储到数据库的哈希值
- key_prefix: 保存前缀用于显示
"""
random_part = secrets.token_hex(24)
full_key = f"yxkey_{random_part}"
key_hash = hashlib.sha256(full_key.encode()).hexdigest()
key_prefix = full_key[:12]
return full_key, key_hash, key_prefix
class APIKeyCreate(BaseModel):
name: str
user_id: int | None = None
department_id: int | None = None
expires_at: str | None = None
class APIKeyUpdate(BaseModel):
name: str | None = None
expires_at: str | None = None
is_enabled: bool | None = None
class APIKeyResponse(BaseModel):
id: int
key_prefix: str
name: str
user_id: int | None
department_id: int | None
expires_at: str | None
is_enabled: bool
last_used_at: str | None
created_by: str
created_at: str
class APIKeyCreateResponse(BaseModel):
api_key: APIKeyResponse
secret: str
@apikey_router.get("/", response_model=dict)
async def list_api_keys(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=500),
current_user: User = Depends(get_superadmin_user),
db: AsyncSession = Depends(get_db),
):
"""列出所有 API Keys"""
result = await db.execute(select(APIKey).order_by(APIKey.created_at.desc()).offset(skip).limit(limit))
api_keys = result.scalars().all()
total_result = await db.execute(select(func.count(APIKey.id)))
total = total_result.scalar()
return {
"api_keys": [key.to_dict() for key in api_keys],
"total": total,
}
@apikey_router.post("/", response_model=APIKeyCreateResponse)
async def create_api_key(
data: APIKeyCreate,
current_user: User = Depends(get_required_user),
db: AsyncSession = Depends(get_db),
):
"""创建新 API Keysecret 仅在此处返回一次)"""
# 生成 Key
full_key, key_hash, key_prefix = generate_api_key()
# 验证关联用户
if data.user_id:
result = await db.execute(select(User).filter(User.id == data.user_id))
user = result.scalar_one_or_none()
if not user or user.is_deleted:
raise HTTPException(status_code=404, detail="关联的用户不存在")
else:
# 自动绑定为当前登录用户
data.user_id = current_user.id
# 解析过期时间(转换为 naive datetime 以匹配数据库字段)
expires_at = None
if data.expires_at:
aware_dt = coerce_any_to_utc_datetime(data.expires_at)
if aware_dt:
expires_at = aware_dt.replace(tzinfo=None)
# 创建记录
api_key = APIKey(
key_hash=key_hash,
key_prefix=key_prefix,
name=data.name,
user_id=data.user_id,
department_id=data.department_id,
expires_at=expires_at,
created_by=str(current_user.id),
)
db.add(api_key)
await db.commit()
await db.refresh(api_key)
return APIKeyCreateResponse(
api_key=APIKeyResponse(**api_key.to_dict()),
secret=full_key,
)
@apikey_router.get("/{api_key_id}", response_model=dict)
async def get_api_key(
api_key_id: int,
current_user: User = Depends(get_superadmin_user),
db: AsyncSession = Depends(get_db),
):
"""获取单个 API Key"""
result = await db.execute(select(APIKey).filter(APIKey.id == api_key_id))
api_key = result.scalar_one_or_none()
if not api_key:
raise HTTPException(status_code=404, detail="API Key 不存在")
return {"api_key": api_key.to_dict()}
@apikey_router.put("/{api_key_id}", response_model=dict)
async def update_api_key(
api_key_id: int,
data: APIKeyUpdate,
current_user: User = Depends(get_superadmin_user),
db: AsyncSession = Depends(get_db),
):
"""更新 API Key"""
result = await db.execute(select(APIKey).filter(APIKey.id == api_key_id))
api_key = result.scalar_one_or_none()
if not api_key:
raise HTTPException(status_code=404, detail="API Key 不存在")
if data.name is not None:
api_key.name = data.name
if data.expires_at is not None:
aware_dt = coerce_any_to_utc_datetime(data.expires_at)
api_key.expires_at = aware_dt.replace(tzinfo=None) if aware_dt else None
if data.is_enabled is not None:
api_key.is_enabled = data.is_enabled
await db.commit()
await db.refresh(api_key)
return {"api_key": api_key.to_dict()}
@apikey_router.delete("/{api_key_id}", response_model=dict)
async def delete_api_key(
api_key_id: int,
current_user: User = Depends(get_superadmin_user),
db: AsyncSession = Depends(get_db),
):
"""删除 API Key"""
result = await db.execute(select(APIKey).filter(APIKey.id == api_key_id))
api_key = result.scalar_one_or_none()
if not api_key:
raise HTTPException(status_code=404, detail="API Key 不存在")
await db.delete(api_key)
await db.commit()
return {"success": True}
@apikey_router.post("/{api_key_id}/regenerate", response_model=APIKeyCreateResponse)
async def regenerate_api_key(
api_key_id: int,
current_user: User = Depends(get_superadmin_user),
db: AsyncSession = Depends(get_db),
):
"""重新生成 API Key 密钥secret 仅在此处返回一次)"""
result = await db.execute(select(APIKey).filter(APIKey.id == api_key_id))
api_key = result.scalar_one_or_none()
if not api_key:
raise HTTPException(status_code=404, detail="API Key 不存在")
# 生成新密钥
full_key, key_hash, key_prefix = generate_api_key()
api_key.key_hash = key_hash
api_key.key_prefix = key_prefix
await db.commit()
await db.refresh(api_key)
return APIKeyCreateResponse(
api_key=APIKeyResponse(**api_key.to_dict()),
secret=full_key,
)

View File

@ -1,13 +1,16 @@
import hashlib
import re
from fastapi import Depends, HTTPException, status
from fastapi import Depends, Header, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from yuxi.storage.postgres.manager import pg_manager
from yuxi.storage.postgres.models_business import User
from yuxi.storage.postgres.models_business import User, APIKey
from server.utils.auth_utils import AuthUtils
from yuxi.utils.datetime_utils import utc_now_naive
# 定义OAuth2密码承载器指定token URL
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token", auto_error=False)
@ -29,20 +32,76 @@ async def get_db():
yield db
async def _verify_api_key(key: str, db: AsyncSession) -> tuple[User | None, APIKey | None]:
"""验证 API Key 并返回关联用户和 APIKey 对象"""
key_hash = hashlib.sha256(key.encode()).hexdigest()
result = await db.execute(select(APIKey).filter(APIKey.key_hash == key_hash))
api_key = result.scalar_one_or_none()
if api_key is None:
return None, None
if not api_key.is_enabled:
return None, None
if api_key.expires_at and utc_now_naive() > api_key.expires_at:
return None, None
if api_key.user_id:
result = await db.execute(select(User).filter(User.id == api_key.user_id))
user = result.scalar_one_or_none()
if user and not user.is_deleted:
return user, api_key
if api_key.department_id:
result = await db.execute(
select(User).filter(User.department_id == api_key.department_id, User.role.in_(["admin", "superadmin"]))
)
user = result.scalar_one_or_none()
if user and not user.is_deleted:
return user, api_key
result = await db.execute(select(User).filter(User.role == "superadmin", User.is_deleted == 0).limit(1))
user = result.scalar_one_or_none()
if user:
return user, api_key
return None, None
# 获取当前用户(异步版本)
async def get_current_user(token: str | None = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
async def get_current_user(
authorization: str | None = Header(None),
db: AsyncSession = Depends(get_db),
):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的凭证",
headers={"WWW-Authenticate": "Bearer"},
)
# 允许无token访问公开路径
if token is None:
if authorization is None:
return None
if not authorization.startswith("Bearer "):
return None
token = authorization.split("Bearer ")[1]
if not token:
return None
# 根据 token 前缀判断认证方式
if token.startswith("yxkey_"):
# API Key 认证
user, api_key_obj = await _verify_api_key(token, db)
if user is not None and api_key_obj is not None:
api_key_obj.last_used_at = utc_now_naive()
await db.commit()
return user
# JWT Token 认证
try:
# 验证token
payload = AuthUtils.verify_access_token(token)
user_id = payload.get("sub")
if user_id is None:
@ -50,17 +109,12 @@ async def get_current_user(token: str | None = Depends(oauth2_scheme), db: Async
except JWTError:
raise credentials_exception
except ValueError as e:
# 捕获AuthUtils.verify_access_token可能抛出的ValueError
# 例如令牌过期或无效
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=str(e), # 将错误信息直接传递给客户端
detail=str(e),
headers={"WWW-Authenticate": "Bearer"},
)
# 查找用户(异步版本)
from sqlalchemy import select
result = await db.execute(select(User).filter(User.id == int(user_id)))
user = result.scalar_one_or_none()
if user is None:

View File

@ -0,0 +1,234 @@
"""
Integration tests for API Key router endpoints.
"""
from __future__ import annotations
import pytest
pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
async def test_list_api_keys_requires_auth(test_client):
"""List API keys should require authentication."""
response = await test_client.get("/api/apikey/")
assert response.status_code == 401
async def test_list_api_keys_requires_admin(test_client, admin_headers):
"""List API keys should require admin privileges."""
response = await test_client.get("/api/apikey/", headers=admin_headers)
assert response.status_code == 200, response.text
data = response.json()
assert "api_keys" in data
assert "total" in data
async def test_create_api_key(test_client, admin_headers):
"""Admin should be able to create a new API key."""
payload = {
"name": "Test API Key",
}
response = await test_client.post("/api/apikey/", json=payload, headers=admin_headers)
assert response.status_code == 200, response.text
data = response.json()
assert "api_key" in data
assert "secret" in data
assert data["api_key"]["name"] == "Test API Key"
assert data["api_key"]["key_prefix"].startswith("yxkey_")
# Note: The "****" suffix is added by the frontend, not stored in backend
assert data["api_key"]["key_prefix"] == "yxkey_144cba" or data["api_key"]["key_prefix"].startswith("yxkey_")
# Secret should start with the prefix
assert data["secret"].startswith(data["api_key"]["key_prefix"][:6])
return data
async def test_get_api_key(test_client, admin_headers):
"""Admin should be able to get a single API key."""
# First create a key
create_response = await test_client.post(
"/api/apikey/", json={"name": "Get Test"}, headers=admin_headers
)
assert create_response.status_code == 200
created = create_response.json()["api_key"]
# Then retrieve it
response = await test_client.get(f"/api/apikey/{created['id']}", headers=admin_headers)
assert response.status_code == 200, response.text
data = response.json()
assert data["api_key"]["id"] == created["id"]
assert data["api_key"]["name"] == "Get Test"
async def test_update_api_key(test_client, admin_headers):
"""Admin should be able to update an API key."""
# Create a key
create_response = await test_client.post(
"/api/apikey/", json={"name": "Update Test"}, headers=admin_headers
)
assert create_response.status_code == 200
created = create_response.json()["api_key"]
# Update it
response = await test_client.put(
f"/api/apikey/{created['id']}",
json={"name": "Updated Name", "is_enabled": False},
headers=admin_headers,
)
assert response.status_code == 200, response.text
data = response.json()
assert data["api_key"]["name"] == "Updated Name"
assert data["api_key"]["is_enabled"] is False
async def test_delete_api_key(test_client, admin_headers):
"""Admin should be able to delete an API key."""
# Create a key
create_response = await test_client.post(
"/api/apikey/", json={"name": "Delete Test"}, headers=admin_headers
)
assert create_response.status_code == 200
created = create_response.json()["api_key"]
# Delete it
response = await test_client.delete(f"/api/apikey/{created['id']}", headers=admin_headers)
assert response.status_code == 200, response.text
assert response.json()["success"] is True
# Verify it's gone
get_response = await test_client.get(f"/api/apikey/{created['id']}", headers=admin_headers)
assert get_response.status_code == 404
async def test_regenerate_api_key(test_client, admin_headers):
"""Admin should be able to regenerate an API key."""
# Create a key
create_response = await test_client.post(
"/api/apikey/", json={"name": "Regenerate Test"}, headers=admin_headers
)
assert create_response.status_code == 200
original_secret = create_response.json()["secret"]
created = create_response.json()["api_key"]
# Regenerate it
response = await test_client.post(f"/api/apikey/{created['id']}/regenerate", headers=admin_headers)
assert response.status_code == 200, response.text
data = response.json()
assert "secret" in data
assert data["secret"] != original_secret
assert data["api_key"]["key_prefix"] != original_secret[:12]
async def test_api_key_auth_chat_endpoint(test_client, admin_headers):
"""Test that API Key can be used to authenticate to chat endpoint via Bearer token."""
# Create an API key
create_response = await test_client.post(
"/api/apikey/", json={"name": "Chat Auth Test"}, headers=admin_headers
)
assert create_response.status_code == 200
api_key_secret = create_response.json()["secret"]
created = create_response.json()["api_key"]
try:
# Get default agent
agent_response = await test_client.get("/api/chat/default_agent", headers=admin_headers)
assert agent_response.status_code == 200
agent_id = agent_response.json().get("default_agent_id")
if not agent_id:
pytest.skip("No default agent configured")
# Call chat endpoint with API Key using Bearer format (streaming response)
async with test_client.stream(
"POST",
f"/api/chat/agent/{agent_id}",
json={"query": "Hello"},
headers={"Authorization": f"Bearer {api_key_secret}"},
) as response:
assert response.status_code == 200, response.text
assert response.headers.get("content-type") == "application/json"
finally:
# Cleanup: delete the test API key
await test_client.delete(f"/api/apikey/{created['id']}", headers=admin_headers)
async def test_api_key_auth_requires_valid_key(test_client):
"""Test that invalid API Key is rejected."""
# Call chat endpoint with invalid API Key
response = await test_client.post(
"/api/chat/agent/ChatbotAgent",
json={"query": "Hello"},
headers={"Authorization": "Bearer yxkey_invalid_key_that_does_not_exist"},
)
assert response.status_code == 401, response.text
async def test_api_key_auth_requires_bearer_prefix(test_client, admin_headers):
"""Test that API Key must be prefixed with 'Bearer '."""
# Create an API key
admin_response = await test_client.post(
"/api/apikey/", json={"name": "Prefix Test"}, headers=admin_headers
)
assert admin_response.status_code == 200
api_key_secret = admin_response.json()["secret"]
created = admin_response.json()["api_key"]
try:
# Call without Bearer prefix should fail
response = await test_client.post(
"/api/chat/agent/ChatbotAgent",
json={"query": "Hello"},
headers={"Authorization": api_key_secret}, # Missing "Bearer " prefix
)
assert response.status_code == 401, response.text
finally:
# Cleanup: delete the test API key
await test_client.delete(f"/api/apikey/{created['id']}", headers=admin_headers)
async def test_jwt_still_works_after_apikey_auth(test_client, admin_headers):
"""Test that JWT Bearer tokens still work after API Key changes."""
# Get default agent
agent_response = await test_client.get("/api/chat/default_agent", headers=admin_headers)
assert agent_response.status_code == 200
agent_id = agent_response.json().get("default_agent_id")
if not agent_id:
pytest.skip("No default agent configured")
# Call chat with JWT Bearer token (admin_headers) - streaming response
async with test_client.stream(
"POST",
f"/api/chat/agent/{agent_id}",
json={"query": "Hello"},
headers=admin_headers,
) as response:
assert response.status_code == 200, response.text
assert response.headers.get("content-type") == "application/json"
async def test_api_key_auto_binds_to_current_user(test_client, admin_headers):
"""Test that API Key created without user_id is auto-bound to creator."""
# Create API key as admin
create_response = await test_client.post(
"/api/apikey/", json={"name": "Auto Bind Test"}, headers=admin_headers
)
assert create_response.status_code == 200
created = create_response.json()["api_key"]
try:
# Verify user_id is set (auto-bound to admin)
assert created["user_id"] is not None, "API Key should be auto-bound to creator"
# Verify the key can be used for auth
api_key_secret = create_response.json()["secret"]
async with test_client.stream(
"POST",
"/api/chat/agent/ChatbotAgent",
json={"query": "Hello"},
headers={"Authorization": f"Bearer {api_key_secret}"},
) as response:
assert response.status_code == 200, response.text
finally:
# Cleanup: delete the test API key
await test_client.delete(f"/api/apikey/{created['id']}", headers=admin_headers)

View File

@ -54,7 +54,8 @@ export default defineConfig({
{ text: '文档解析', link: '/advanced/document-processing' },
{ text: '品牌自定义', link: '/advanced/branding' },
{ text: '其他配置', link: '/advanced/misc' },
{ text: '生产部署', link: '/advanced/deployment' }
{ text: '生产部署', link: '/advanced/deployment' },
{ text: 'API Key 外部集成', link: '/advanced/api-key-integration' }
]
},
{

View File

@ -0,0 +1,81 @@
# API Key 外部集成
Yuxi 平台提供了 API Key 认证机制,允许外部系统在无需用户登录的情况下调用智能体对话接口。本文档详细介绍 API Key 的使用方法、接口调用方式以及安全注意事项。
## API Key 概述
API Key 是一种用于身份验证的密钥字符串,外部系统可以通过它在请求头中携带凭据来访问 Yuxi 的对话接口。与传统的用户名密码登录方式相比API Key 更加适合用于系统间的自动化调用场景。Yuxi 的 API Key 以 `yxkey_` 为前缀,长度为 56 个字符,采用 SHA-256 哈希存储,确保密钥本身不会在数据库中明文保存。系统会记录每个 API Key 的最后使用时间,方便管理员追踪使用情况。
## 创建 API Key
登录系统后,进入 API Key 管理界面,可以创建新的密钥。创建时需要为 API Key 设置一个名称,用于标识其用途,例如"外部客服系统"或"数据同步服务"。创建的 API Key 会自动绑定到当前登录用户,绑定后的 API Key 在调用接口时会以该用户的身份执行操作。API Key 还支持设置过期时间,过期后该密钥将自动失效。
需要特别注意的是,创建 API Key 时返回的完整密钥secret只会显示一次务必在创建时将其安全保存。如果遗失需要通过"重新生成"功能生成新的密钥,原有的密钥将立即失效。
## 接口调用方式
外部系统通过 HTTP 请求调用 Yuxi 的对话接口,需要在请求头中携带 API Key。接口地址为 `POST /api/chat/agent/{agent_id}`,其中 `{agent_id}` 是目标智能体的标识符。请求头需要包含 `Authorization` 字段,值格式为 `Bearer <api_key>`,其中 `<api_key>` 是创建 API Key 时获取的完整密钥。请求体为 JSON 格式,包含 `query` 字段表示用户问题,以及可选的 `config``meta` 字段用于配置对话参数。
以下是一个典型的 Python 调用示例:
```python
import requests
import json
url = "http://your-yuxi-server/api/chat/agent/agent_id"
headers = {
"Authorization": "Bearer yxkey_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"Content-Type": "application/json"
}
payload = {
"query": "你好,请介绍一下你自己",
"config": {},
"meta": {}
}
response = requests.post(url, headers=headers, json=payload, stream=True)
for line in response.iter_lines():
if line:
print(line.decode('utf-8'))
```
该接口返回的是流式响应Server-Sent Events每个事件是一行 JSON 数据,包含对话的增量内容。客户端需要逐行解析并处理这些事件来构建完整的对话结果。
## 响应格式
接口返回的流式响应采用 JSON Lines 格式,每行代表一个事件。常见的事件类型包括:
`event: data` 表示数据事件,携带实际的对话内容。`event: error` 表示错误事件,当对话过程中发生错误时会收到此类事件。`event: done` 表示完成事件,标志对话结束。
每次调用都会在响应中包含 `request_id`,这是本次对话的唯一标识符,可用于日志追踪和问题排查。如果需要在多轮对话中使用同一个会话,可以通过 `config.thread_id` 参数指定线程 ID系统会将同一线程的消息串联起来形成连贯的对话上下文。
## 认证方式
Yuxi 的 API 接口统一支持两种认证方式:
1. **API Key 认证**:使用 `Authorization: Bearer <api_key>` 格式,其中 API Key 必须以 `yxkey_` 前缀开头
2. **JWT Token 认证**:使用 `Authorization: Bearer <jwt_token>` 格式
系统根据 token 的前缀自动判断认证方式。以 `yxkey_` 开头的 token 被视为 API Key其他 token 则作为 JWT Token 处理。这种设计使得同一个接口可以同时支持外部系统(使用 API Key和内部前端应用使用用户登录态调用。
## 安全注意事项
保管好 API Key 密钥是最重要的安全原则。由于 API Key 一旦泄露就可能被滥用,建议不要将密钥硬编码在代码中,而是通过环境变量或配置中心来管理。如果怀疑密钥泄露,应立即在管理界面禁用该 API Key 并重新生成。启用密钥过期功能是一种良好的安全实践,可以设置较短的有效期并定期轮换。
在生产环境中,建议为不同的外部系统创建独立的 API Key这样可以在某个密钥泄露时快速定位问题并限制影响范围。同时建议在管理界面定期查看 API Key 的使用记录,检查是否存在异常调用情况。
关于权限控制API Key 的权限等同于其绑定的用户在系统中的角色。如果 API Key 绑定到特定用户,则该用户的所有权限都会体现在 API Key 的操作中,因此务必妥善保管。
## 常见问题
**Q: API Key 认证失败返回什么错误?**
A: 认证失败时返回 401 Unauthorized 错误,错误信息为"无效的凭证"。请检查请求头中 `Authorization` 字段的格式是否正确,是否包含完整的密钥,且密钥必须以 `yxkey_` 开头。
**Q: 可以同时使用 API Key 和 JWT Token 吗?**
A: 不可以。系统根据 token 前缀自动判断认证方式。以 `yxkey_` 开头的 token 使用 API Key 认证,其他 token 使用 JWT 认证。
**Q: API Key 是否有调用频率限制?**
A: 目前没有单独的频率限制,但 API Key 的行为等同于其绑定的用户身份,因此会受到用户角色相关的一些限制。
**Q: 对话返回的内容是乱码怎么办?**
A: 确保客户端正确处理了 UTF-8 编码。流式响应中可能包含中文字符,需要使用正确的编码方式解析。如果在终端显示乱码,可以检查终端的编码设置。

View File

@ -6,7 +6,6 @@
### 看板
- 集成 LangFuse (观望) 添加用户日志与用户反馈模块,可以在 AgentView 中查看信息
- 系统层面添加 apikey在智能体、知识库调用中支持 apikey 以支持外部调用
- 部分场景应该使用默认模型作为默认值而不是空值
- 检索测试中,添加问答
- 集成 Memory基于 deepagents 的文件后端实现
@ -26,6 +25,7 @@
<!-- 添加到这里 -->
- 新增 API Key 管理功能,支持外部系统通过 API Key 调用 Agent 对话接口(`POST /api/chat/agent/{agent_id}`)。统一使用 `Authorization: Bearer <api_key>` 认证API Key 以 `yxkey_` 开头。获取 API Key 即代表拥有绑定用户的所有接口访问权限。
- 将 后端代码 和 agents 解耦agents 作为单独的 package 使用
- 添加 subagents

View File

@ -0,0 +1,15 @@
import { apiSuperAdminGet, apiSuperAdminPost, apiSuperAdminPut, apiDelete } from './base'
export const apikeyApi = {
list: (skip = 0, limit = 100) => apiSuperAdminGet('/api/apikey/', { params: { skip, limit } }),
create: (data) => apiSuperAdminPost('/api/apikey/', data),
get: (id) => apiSuperAdminGet(`/api/apikey/${id}`),
update: (id, data) => apiSuperAdminPut(`/api/apikey/${id}`, data),
delete: (id) => apiDelete(`/api/apikey/${id}`),
regenerate: (id) => apiSuperAdminPost(`/api/apikey/${id}/regenerate`)
}

View File

@ -0,0 +1,464 @@
<template>
<div class="apikey-management">
<!-- 头部区域 -->
<div class="header-section">
<div class="header-content">
<h3 class="title">API Key 管理</h3>
<p class="description">
创建和管理 API Key用于外部系统调用 Agent 对话接口密钥仅显示一次请妥善保管
</p>
</div>
<a-button type="primary" @click="showCreateModal" class="add-btn">
<Plus :size="14" />
创建 API Key
</a-button>
</div>
<!-- 主内容区域 -->
<div class="content-section">
<a-spin :spinning="loading">
<div v-if="error" class="error-message">
<a-alert type="error" :message="error" show-icon />
</div>
<div class="cards-container">
<div v-if="apiKeys.length === 0" class="empty-state">
<a-empty description="暂无 API Key点击上方按钮创建一个" />
</div>
<div v-else class="apikey-cards-grid">
<div v-for="key in apiKeys" :key="key.id" class="apikey-card">
<div class="card-header">
<div class="key-info">
<KeyIcon size="18" class="key-icon" />
<div class="key-info-content">
<h4 class="key-name">{{ key.name }}</h4>
</div>
</div>
<code class="key-prefix">{{ key.key_prefix }}****</code>
</div>
<div class="card-content">
<div class="info-item">
<span class="info-label">过期时间:</span>
<span class="info-value">{{ key.expires_at || '永不过期' }}</span>
</div>
<div class="info-item">
<span class="info-label">最后使用:</span>
<span class="info-value">{{ formatTime(key.last_used_at) }}</span>
</div>
</div>
<div class="card-footer">
<div class="footer-left">
<span class="switch-label">{{ key.is_enabled ? '已启用' : '已禁用' }}</span>
<a-switch :checked="key.is_enabled" size="small" @change="toggleEnabled(key)" />
</div>
<div class="footer-actions">
<a-tooltip title="重新生成(获取完整密钥)">
<a-button
type="text"
size="small"
@click="regenerateKey(key)"
class="action-btn"
>
<RefreshCw :size="14" />
<span>重新生成</span>
</a-button>
</a-tooltip>
<a-popconfirm
title="确定要删除此 API Key 吗?此操作不可恢复。"
@confirm="deleteKey(key)"
ok-text="确定"
cancel-text="取消"
>
<a-tooltip title="删除">
<a-button type="text" size="small" danger class="action-btn">
<Trash2 :size="14" />
<span>删除</span>
</a-button>
</a-tooltip>
</a-popconfirm>
</div>
</div>
</div>
</div>
</div>
</a-spin>
</div>
<!-- 创建 Modal -->
<a-modal
v-model:open="createModalVisible"
title="创建 API Key"
@ok="handleCreate"
:confirmLoading="createLoading"
ok-text="创建"
cancel-text="取消"
>
<a-form layout="vertical" :model="createForm">
<a-form-item label="名称" required>
<a-input v-model:value="createForm.name" placeholder="如生产环境API" />
</a-form-item>
<a-form-item label="过期时间">
<a-date-picker
v-model:value="createForm.expires_at"
show-time
placeholder="留空表示永不过期"
style="width: 100%"
/>
</a-form-item>
</a-form>
</a-modal>
<!-- 密钥显示 Modal (创建后一次性显示) -->
<a-modal
v-model:open="secretModalVisible"
title="API Key 已创建"
:closable="true"
@cancel="secretModalVisible = false"
:footer="null"
width="520px"
>
<div class="secret-display">
<a-alert
type="warning"
message="请立即复制密钥,关闭后将无法再次查看完整密钥"
show-icon
class="secret-alert"
/>
<div class="secret-value-container">
<code class="secret-value">{{ createdSecret }}</code>
<a-button type="primary" @click="copySecret" class="copy-btn">
<Copy :size="14" />
复制
</a-button>
</div>
</div>
</a-modal>
</div>
</template>
<script setup>
import { ref, reactive, onMounted } from 'vue'
import { message } from 'ant-design-vue'
import { Plus, RefreshCw, Trash2, Copy } from 'lucide-vue-next'
import { Key as KeyIcon } from 'lucide-vue-next'
import { apikeyApi } from '@/apis/apikey_api'
const loading = ref(false)
const error = ref(null)
const apiKeys = ref([])
const createModalVisible = ref(false)
const secretModalVisible = ref(false)
const createLoading = ref(false)
const createdSecret = ref('')
const createForm = reactive({
name: '',
expires_at: null
})
const formatTime = (timeStr) => {
if (!timeStr) return '-'
const date = new Date(timeStr)
return date.toLocaleString('zh-CN', {
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit'
})
}
const loadApiKeys = async () => {
loading.value = true
error.value = null
try {
const res = await apikeyApi.list()
apiKeys.value = res.api_keys || []
} catch (e) {
error.value = e.message || '加载失败'
} finally {
loading.value = false
}
}
const showCreateModal = () => {
createForm.name = ''
createForm.expires_at = null
createModalVisible.value = true
}
const handleCreate = async () => {
if (!createForm.name.trim()) {
message.error('请输入名称')
return
}
createLoading.value = true
try {
const data = { name: createForm.name }
if (createForm.expires_at) {
data.expires_at = createForm.expires_at.format('YYYY-MM-DDTHH:mm:ss')
}
const res = await apikeyApi.create(data)
createdSecret.value = res.secret
createModalVisible.value = false
secretModalVisible.value = true
await loadApiKeys()
} catch (e) {
message.error(e.message || '创建失败')
} finally {
createLoading.value = false
}
}
const copySecret = async () => {
try {
await navigator.clipboard.writeText(createdSecret.value)
message.success('已复制到剪贴板')
} catch {
message.error('复制失败')
}
}
const regenerateKey = async (key) => {
try {
const res = await apikeyApi.regenerate(key.id)
createdSecret.value = res.secret
secretModalVisible.value = true
await loadApiKeys()
} catch (e) {
message.error(e.message || '重新生成失败')
}
}
const toggleEnabled = async (key) => {
try {
await apikeyApi.update(key.id, { is_enabled: !key.is_enabled })
message.success(key.is_enabled ? '已禁用' : '已启用')
await loadApiKeys()
} catch (e) {
message.error(e.message || '操作失败')
}
}
const deleteKey = async (key) => {
try {
await apikeyApi.delete(key.id)
message.success('删除成功')
await loadApiKeys()
} catch (e) {
message.error(e.message || '删除失败')
}
}
onMounted(() => {
loadApiKeys()
})
</script>
<style lang="less" scoped>
.apikey-management {
padding: 12px;
min-height: 50vh;
.header-section {
display: flex;
justify-content: space-between;
align-items: flex-start;
margin-bottom: 20px;
gap: 16px;
.header-content {
.title {
font-size: 16px;
font-weight: 600;
color: var(--gray-900);
margin: 0 0 4px 0;
}
.description {
font-size: 13px;
color: var(--gray-600);
margin: 0;
line-height: 1.4;
}
}
.add-btn {
flex-shrink: 0;
display: inline-flex;
align-items: center;
gap: 6px;
}
}
.content-section {
.error-message {
margin-bottom: 16px;
}
.cards-container {
.empty-state {
padding: 48px 0;
}
.apikey-cards-grid {
display: grid;
grid-template-columns: repeat(auto-fill, minmax(320px, 1fr));
gap: 12px;
}
.apikey-card {
background: var(--gray-0);
border: 1px solid var(--gray-150);
border-radius: 8px;
padding: 12px;
transition:
border-color 0.2s,
box-shadow 0.2s;
&:hover {
border-color: var(--gray-300);
}
.card-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 10px;
.key-info {
display: flex;
align-items: center;
gap: 10px;
.key-icon {
color: var(--main-600);
flex-shrink: 0;
}
.key-info-content {
.key-name {
font-size: 14px;
font-weight: 600;
color: var(--gray-900);
margin: 0;
}
}
}
.key-prefix {
font-family: 'Monaco', 'Consolas', monospace;
font-size: 12px;
color: var(--gray-600);
background: var(--gray-50);
padding: 2px 8px;
border-radius: 8px;
}
}
.card-content {
margin-bottom: 10px;
.info-item {
display: flex;
align-items: flex-start;
gap: 6px;
margin-bottom: 6px;
font-size: 13px;
&:last-child {
margin-bottom: 0;
}
.info-label {
color: var(--gray-600);
flex-shrink: 0;
}
.info-value {
color: var(--gray-900);
word-break: break-all;
}
&.half {
flex: 1;
}
}
}
.card-footer {
display: flex;
justify-content: space-between;
align-items: center;
padding-top: 8px;
border-top: 1px solid var(--gray-100);
.footer-left {
display: flex;
align-items: center;
gap: 8px;
.switch-label {
font-size: 12px;
color: var(--gray-600);
}
}
.footer-actions {
display: flex;
gap: 4px;
}
.action-btn {
font-size: 12px;
color: var(--gray-700);
display: inline-flex;
align-items: center;
gap: 4px;
&:hover {
color: var(--main-600);
}
}
}
}
}
}
}
.secret-display {
.secret-alert {
margin-bottom: 16px;
}
.secret-value-container {
display: flex;
gap: 8px;
align-items: stretch;
.secret-value {
flex: 1;
font-family: 'Monaco', 'Consolas', monospace;
font-size: 13px;
background: var(--gray-100);
border: 1px solid var(--gray-200);
border-radius: 6px;
padding: 12px;
word-break: break-all;
color: var(--gray-900);
}
.copy-btn {
flex-shrink: 0;
display: inline-flex;
align-items: center;
gap: 6px;
}
}
}
</style>

View File

@ -49,6 +49,15 @@
<TeamOutlined class="icon" />
<span>部门管理</span>
</div>
<div
class="sider-item"
:class="{ activesec: activeTab === 'apikey' }"
@click="activeTab = 'apikey'"
v-if="userStore.isSuperAdmin"
>
<KeyIcon class="icon" :size="14" />
<span>API Key</span>
</div>
</div>
<!-- 顶部导航 (Mobile) -->
@ -85,6 +94,14 @@
>
部门管理
</div>
<div
class="nav-item"
:class="{ active: activeTab === 'apikey' }"
@click="activeTab = 'apikey'"
v-if="userStore.isSuperAdmin"
>
API Key
</div>
</div>
<!-- 内容区域 -->
@ -105,6 +122,10 @@
<div v-show="activeTab === 'department'" v-if="userStore.isSuperAdmin">
<DepartmentManagementComponent />
</div>
<div v-show="activeTab === 'apikey'" v-if="userStore.isSuperAdmin">
<ApiKeyManagementComponent />
</div>
</div>
</div>
</div>
@ -115,10 +136,12 @@
import { ref, computed, watch } from 'vue'
import { useUserStore } from '@/stores/user'
import { SettingOutlined, CodeOutlined, UserOutlined, TeamOutlined } from '@ant-design/icons-vue'
import { Key as KeyIcon } from 'lucide-vue-next'
import BasicSettingsSection from '@/components/BasicSettingsSection.vue'
import ModelProvidersComponent from '@/components/ModelProvidersComponent.vue'
import UserManagementComponent from '@/components/UserManagementComponent.vue'
import DepartmentManagementComponent from '@/components/DepartmentManagementComponent.vue'
import ApiKeyManagementComponent from '@/components/ApiKeyManagementComponent.vue'
const props = defineProps({
visible: {