d8dfun/ai-mcp-web-ui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(auth): 修复 Admin MCP JWT 认证流程和调试工具

Browse Source 
主要修复:
- 修复 admin-login 端点 token 提取 (access_token 字段)
- 修复用户信息提取 (user.username 而非直接 username)
- 修复流式聊天中 tool_to_server_map 参数传递

新增功能:
- 添加 /api/test-mcp 测试端点 (GET/POST)
- 添加完整的调试日志追踪 token 传递路径

调试日志覆盖:
- 前端 X-MCP-Tokens header 接收
- ConversationManager token 初始化
- ToolHandler server_id 查找
- MCPClient Authorization header 构建

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude AI
2026-03-18 05:42:50 +00:00
parent 13e55d0638
commit 8e1d831340
6 changed files with 218 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
backend/app_fastapi.py
View File

@@ -134,6 +134,11 @@ async def chat(request: Request):
else:
parsed_tokens = mcp_tokens
# DEBUG: 打印收到的 token
print(f"[DEBUG /api/chat] Received mcp_tokens: {list(parsed_tokens.keys()) if parsed_tokens else 'None'}")
for k, v in parsed_tokens.items():
print(f"[DEBUG /api/chat] {k}: {v[:30] if v else 'None'}...")
# 创建对话管理器（带 token）
conv_manager = ConversationManager(
api_key=ANTHROPIC_API_KEY,
@@ -194,11 +199,16 @@ async def generate_chat_stream(
if isinstance(mcp_tokens, str):
try:
parsed_tokens = json_module.loads(mcp_tokens)
except:
except Exception as e:
parsed_tokens = {}
else:
parsed_tokens = mcp_tokens
# DEBUG: 打印解析后的 token
print(f"[DEBUG generate_chat_stream] Parsed mcp_tokens keys: {list(parsed_tokens.keys())}")
for k, v in parsed_tokens.items():
print(f"[DEBUG generate_chat_stream] {k}: {v[:30] if v else 'None'}...")
# 创建对话管理器（带 token）
conv_manager = ConversationManager(
api_key=ANTHROPIC_API_KEY,
@@ -327,7 +337,8 @@ async def generate_chat_stream(
yield f"event: tool_call\ndata: {json_module.dumps({'tool': tool_block['name'], 'args': tool_block['input'], 'tool_id': tool_block['id']})}\n\n"
tool_results = await conv_manager.tool_handler.process_tool_use_blocks(
tool_use_blocks
tool_use_blocks,
conv_manager._tool_to_server_map
)
for tr in tool_results:
@@ -389,6 +400,10 @@ async def chat_stream(request: Request):
session_id = request.headers.get('X-Session-ID')
mcp_tokens = request.headers.get('X-MCP-Tokens') # MCP tokens (JSON string)
# DEBUG: 打印收到的 token
print(f"[DEBUG /api/chat/stream] mcp_tokens type: {type(mcp_tokens)}")
print(f"[DEBUG /api/chat/stream] mcp_tokens value: {mcp_tokens[:150] if mcp_tokens else 'None'}...")
if not message:
raise HTTPException(status_code=400, detail="Message is required")
@@ -507,11 +522,15 @@ async def login(request: Request):
result = response.json()
session_id = str(uuid.uuid4())
# Novel Platform API 返回 access_token 和 user 对象
access_token = result.get("access_token")
user_info = result.get("user", {})
# 存储会话信息
auth_sessions[session_id] = {
"username": result.get("username", email),
"username": user_info.get("username") or user_info.get("email", email),
"email": email,
"token": result.get("token"),
"token": access_token,
"refresh_token": result.get("refresh_token"),
"server": target_server.get("name")
}
@@ -519,9 +538,9 @@ async def login(request: Request):
return {
"success": True,
"session_id": session_id,
"username": result.get("username", email),
"username": user_info.get("username") or user_info.get("email", email),
"server": target_server.get("name"),
"token": result.get("token")
"token": access_token
}
else:
raise HTTPException(
@@ -571,10 +590,14 @@ async def admin_login(request: Request):
result = response.json()
session_id = str(uuid.uuid4())
# Novel Platform API 返回 access_token 和 user 对象
access_token = result.get("access_token")
user_info = result.get("user", {})
auth_sessions[session_id] = {
"username": result.get("username", email),
"username": user_info.get("username") or user_info.get("email", email),
"email": email,
"token": result.get("token"),
"token": access_token,
"refresh_token": result.get("refresh_token"),
"server": target_server.get("name"),
"role": "admin"
@@ -583,10 +606,10 @@ async def admin_login(request: Request):
return {
"success": True,
"session_id": session_id,
"username": result.get("username", email),
"username": user_info.get("username") or user_info.get("email", email),
"server": target_server.get("name"),
"role": "admin",
"token": result.get("token")
"token": access_token
}
else:
raise HTTPException(
@@ -630,6 +653,157 @@ async def auth_status(x_session_id: Optional[str] = Header(None, alias='X-Sessio
return {"authenticated": False}
# ========== 测试 API ==========
@app.get("/api/test-mcp")
async def test_mcp_get(
tool_name: str = "get_system_stats",
server_id: str = "novel-platform-admin",
auth_token: Optional[str] = None
):
"""
GET 方式测试 MCP 工具调用（用于 curl 测试）
参数:
- tool_name: 工具名称 (默认: get_system_stats)
- server_id: 服务器 ID (默认: novel-platform-admin)
- auth_token: JWT token (通过 query 参数或 header 传递)
"""
try:
# 如果 query 参数没有 token，尝试从 header 获取
if not auth_token:
# 这个处理会在实际请求时通过 FastAPI 的 Header 参数处理
pass
print("\n" + "="*60)
print("[TEST-MCP GET] MCP 工具调用测试")
print("="*60)
print(f"[TEST-MCP GET] server_id: {server_id}")
print(f"[TEST-MCP GET] tool_name: {tool_name}")
print(f"[TEST-MCP GET] auth_token present: {bool(auth_token)}")
if auth_token:
print(f"[TEST-MCP GET] auth_token (前50字符): {auth_token[:50]}...")
print("="*60 + "\n")
# 创建 MCP 客户端
client = MCPClient(
server_id=server_id,
session_id="test-session",
auth_token=auth_token
)
# 调用工具（无参数）
print(f"[TEST-MCP GET] 开始调用工具...")
result = await client.call_tool(tool_name, {})
print(f"[TEST-MCP GET] 调用完成")
print(f"[TEST-MCP GET] success: {result.get('success', False)}")
print("="*60 + "\n")
return {
"success": True,
"server_id": server_id,
"tool_name": tool_name,
"result": result
}
except Exception as e:
import traceback
print(f"[TEST-MCP GET] 异常: {e}")
traceback.print_exc()
return JSONResponse(
status_code=500,
content={
"success": False,
"error": str(e),
"traceback": traceback.format_exc()
}
)
@app.post("/api/test-mcp")
async def test_mcp_call(request: Request):
"""
直接测试 MCP 工具调用（绕过 CORS，用于调试）
请求体:
{
"server_id": "novel-platform-admin", // 可选，默认使用 admin
"tool_name": "get_system_stats", // 工具名称
"arguments": {}, // 工具参数
"auth_token": "jwt-token" // JWT 认证 token
}
"""
try:
data = await request.json()
server_id = data.get('server_id', 'novel-platform-admin')
tool_name = data.get('tool_name', '')
arguments = data.get('arguments', {})
auth_token = data.get('auth_token')
print("\n" + "="*60)
print("[TEST-MCP] MCP 工具调用测试")
print("="*60)
print(f"[TEST-MCP] server_id: {server_id}")
print(f"[TEST-MCP] tool_name: {tool_name}")
print(f"[TEST-MCP] arguments: {arguments}")
print(f"[TEST-MCP] auth_token present: {bool(auth_token)}")
if auth_token:
print(f"[TEST-MCP] auth_token (前50字符): {auth_token[:50]}...")
print("="*60 + "\n")
if not tool_name:
raise HTTPException(status_code=400, detail="tool_name is required")
# 创建 MCP 客户端
client = MCPClient(
server_id=server_id,
session_id="test-session",
auth_token=auth_token
)
# 调用工具
print(f"[TEST-MCP] 开始调用工具...")
result = await client.call_tool(tool_name, arguments)
print(f"[TEST-MCP] 调用结果:")
print(f"[TEST-MCP] success: {result.get('success', False)}")
print(f"[TEST-MCP] has_error: {'error' in result}")
if 'error' in result:
print(f"[TEST-MCP] error: {result['error']}")
else:
result_preview = result.get('result', '')[:100]
print(f"[TEST-MCP] result (预览): {result_preview}...")
print("="*60 + "\n")
return {
"success": True,
"server_id": server_id,
"tool_name": tool_name,
"arguments": arguments,
"result": result,
"debug": {
"auth_token_present": bool(auth_token),
"auth_token_length": len(auth_token) if auth_token else 0
}
}
except HTTPException:
raise
except Exception as e:
import traceback
print(f"[TEST-MCP] 异常: {e}")
traceback.print_exc()
return JSONResponse(
status_code=500,
content={
"success": False,
"error": str(e),
"traceback": traceback.format_exc()
}
)
# ========== 主程序入口 ==========
if __name__ == '__main__':

2
backend/config.py
View File

@@ -23,7 +23,7 @@ MCP_SERVERS = {
"name": "Novel Platform Admin MCP",
"url": "https://d8d-ai-vscode-8080-223-238-template-6-group.dev.d8d.fun/admin-mcp/",
"auth_type": "jwt",
"login_url": "/api/v1/auth/admin-login",
"login_url": "/api/v1/auth/login",
"base_url": "https://d8d-ai-vscode-8080-223-238-template-6-group.dev.d8d.fun",
"enabled": True
},

6
backend/conversation_manager.py
View File

@@ -41,6 +41,12 @@ class ConversationManager:
self.model = model
self.session_id = session_id
self.mcp_tokens = mcp_tokens or {} # MCP 服务器 token 映射
# DEBUG: 打印接收到的 token
print(f"[DEBUG ConversationManager.__init__] mcp_tokens keys: {list(self.mcp_tokens.keys())}")
for k, v in self.mcp_tokens.items():
print(f"[DEBUG ConversationManager.__init__] {k}: {v[:30] if v else 'None'}...")
self.tool_handler = ToolCallHandler(session_id=session_id, mcp_tokens=mcp_tokens)
self._cached_tools = None
self._tool_to_server_map = {} # 工具名到服务器 ID 的映射

13
backend/mcp_client.py
View File

@@ -106,6 +106,17 @@ class MCPClient:
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
# DEBUG: 打印调用详情
print(f"[DEBUG MCPClient.call_tool] tool_name: {tool_name}")
print(f"[DEBUG MCPClient.call_tool] server_id: {self.server_id}")
print(f"[DEBUG MCPClient.call_tool] url: {url}")
print(f"[DEBUG MCPClient.call_tool] auth_token present: {bool(self.auth_token)}")
if self.auth_token:
print(f"[DEBUG MCPClient.call_tool] auth_token: {self.auth_token[:30]}...")
print(f"[DEBUG MCPClient.call_tool] Authorization header: {headers['Authorization'][:50]}...")
else:
print(f"[DEBUG MCPClient.call_tool] NO auth_token!")
payload = {
"jsonrpc": "2.0",
"id": 2,
@@ -118,6 +129,8 @@ class MCPClient:
response = await client.post(url, json=payload, headers=headers)
print(f"[DEBUG MCPClient.call_tool] response status: {response.status_code}")
if response.status_code == 200:
# 解析 SSE 响应
json_text = parse_sse_response(response.text)

12
backend/tool_handler.py
View File

@@ -56,6 +56,14 @@ class ToolCallHandler:
if server_id and server_id in self.mcp_tokens:
auth_token = self.mcp_tokens[server_id]
# DEBUG: 打印工具调用详情
print(f"[DEBUG ToolCallHandler.process_tool_use_block] tool: {tool_name}")
print(f"[DEBUG ToolCallHandler.process_tool_use_block] server_id: {server_id}")
print(f"[DEBUG ToolCallHandler.process_tool_use_block] auth_token present: {bool(auth_token)}")
if auth_token:
print(f"[DEBUG ToolCallHandler.process_tool_use_block] auth_token: {auth_token[:30]}...")
print(f"[DEBUG ToolCallHandler.process_tool_use_block] available tokens: {list(self.mcp_tokens.keys())}")
# 调用 MCP 工具（带 token）
client = MCPClient(session_id=self.session_id, server_id=server_id, auth_token=auth_token)
result = await client.call_tool(tool_name, tool_input)
@@ -87,6 +95,10 @@ class ToolCallHandler:
if block.get("type") == "tool_use"
]
# DEBUG: 打印工具映射和可用 tokens
print(f"[DEBUG ToolCallHandler.process_tool_use_blocks] tool_to_server_map keys: {list(tool_to_server_map.keys()) if tool_to_server_map else 'None'}")
print(f"[DEBUG ToolCallHandler.process_tool_use_blocks] self.mcp_tokens keys: {list(self.mcp_tokens.keys())}")
if not tool_use_blocks:
return []

4
frontend/index.html
View File

@@ -514,9 +514,9 @@
},
"novel-platform-admin": {
"name": "Novel Platform Admin MCP",
"url": "https://d8d-ai-vscode-8080-223-238-template-6-group.dev.d8d.fun/mcp/",
"url": "https://d8d-ai-vscode-8080-223-238-template-6-group.dev.d8d.fun/admin-mcp/",
"auth_type": "jwt",
"login_api": "/api/v1/auth/admin-login",
"login_api": "/api/v1/auth/login",
"base_url": "https://d8d-ai-vscode-8080-223-238-template-6-group.dev.d8d.fun"
}
};