From d593d8dc42ae338754a420f4f1ddbcca260886e3 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Sat, 18 Jul 2026 13:52:26 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=98=BE=E5=BC=8F=E9=80=89=E4=B8=AD?= =?UTF-8?q?=E6=8A=80=E8=83=BD/MCP=20=E6=97=B6=E4=BC=98=E5=85=88=E6=8C=89?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E8=B5=84=E6=96=99=E5=9B=9E=E7=AD=94=20+=20?= =?UTF-8?q?=E5=AA=92=E4=BD=93=E4=BB=BB=E5=8A=A1=E8=B6=85=E6=97=B6/?= =?UTF-8?q?=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _build_system_prompt 末尾增加【最高优先级覆盖规则】：用户明确选择技能/MCP 并询问其介绍/功能/用法时， 必须只基于该工具资料作答，禁止引用平台知识库中技能市场/MCP/平台能力的通用描述。 - _generate_media_task 增加启动/结束日志，对图像生成加 90s 超时、视频 300s 超时，避免异步任务挂死导致前端一直显示“生成中”。 --- .../app/api/admin/mcp_admin.py.bak_1784203135 | 886 ++++++++++++++++++ backend/python/app/api/chat.py | 41 +- 2 files changed, 920 insertions(+), 7 deletions(-) create mode 100644 backend/python/app/api/admin/mcp_admin.py.bak_1784203135 diff --git a/backend/python/app/api/admin/mcp_admin.py.bak_1784203135 b/backend/python/app/api/admin/mcp_admin.py.bak_1784203135 new file mode 100644 index 0000000..8de2442 --- /dev/null +++ b/backend/python/app/api/admin/mcp_admin.py.bak_1784203135 @@ -0,0 +1,886 @@ +""" + +MCP (Model Context Protocol) Server Management API + +Phase 3.3: Manage MCP servers, discover tools/resources, test connections + +""" + +from fastapi import APIRouter, Depends, HTTPException, Query, Body + +from sqlalchemy import or_, func, desc, asc, text + +from sqlalchemy.orm import Session + +from pydantic import BaseModel, Field + +from typing import Optional, List, Dict, Any + +from datetime import datetime + +import uuid + +import json + +import logging + +import re + +from ...database import get_db + +from ...models import gen_uuid + +from ...auth import get_current_admin + +from ...middleware.log_service import LogService + +router = APIRouter() + +logger = logging.getLogger("aiflow") + +# ============================================================ + +# Pydantic Models + +# ============================================================ + +class MCPServerCreate(BaseModel): + + name: str + + description: Optional[str] = "" + + transport_type: str = "stdio" + + command: Optional[str] = "" + + args: Optional[Any] = None + + env_vars: Optional[Dict[str, str]] = None + + url: Optional[str] = "" + + headers: Optional[Dict[str, str]] = None + + is_enabled: Optional[bool] = True + +class MCPServerUpdate(BaseModel): + + name: Optional[str] = None + + description: Optional[str] = None + + transport_type: Optional[str] = None + + command: Optional[str] = None + + args: Optional[Any] = None + + env_vars: Optional[Dict[str, str]] = None + + url: Optional[str] = None + + headers: Optional[Dict[str, str]] = None + + is_enabled: Optional[bool] = None + +class MCPServerConnectionTest(BaseModel): + + transport_type: Optional[str] = None + + command: Optional[str] = None + + args: Optional[Any] = None + + env_vars: Optional[Dict[str, str]] = None + + url: Optional[str] = None + + headers: Optional[Dict[str, str]] = None + +# ============================================================ + +# Helper Functions + +# ============================================================ + +def _format_cst_time(dt): + + """Format datetime to CST string""" + + if dt is None: + + return None + + try: + + from datetime import timezone, timedelta + + CST = timezone(timedelta(hours=8)) + + if dt.tzinfo is None: + + dt = dt.replace(tzinfo=CST) + + return dt.astimezone(CST).strftime("%Y-%m-%d %H:%M:%S") + + except Exception: + + return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None + +def server_to_dict(server) -> dict: + + """Convert MCPServer model to dict""" + + return { + + "id": server.id, + + "name": server.name, + + "description": server.description or "", + + "transport_type": server.transport_type or "stdio", + + "transport_type_label": {"stdio": "STDIO", "sse": "SSE"}.get(server.transport_type, server.transport_type), + + "command": server.command or "", + + "args": json.loads(server.args) if server.args else {}, + + "env_vars": json.loads(server.env_vars) if server.env_vars else {}, + + "url": server.url or "", + + "headers": json.loads(server.headers) if server.headers else {}, + + "is_enabled": server.is_enabled if server.is_enabled is not None else True, + + "status": server.status or "disconnected", + + "last_connected_at": _format_cst_time(server.last_connected_at), + + "tools_count": server.tools_count or 0, + + "resources_count": server.resources_count or 0, + + "created_at": _format_cst_time(server.created_at), + + "updated_at": _format_cst_time(server.updated_at), + + } + +def _extract_url_from_text(text: str) -> Optional[str]: + + """Extract first http/https URL from text.""" + + if not text: + + return None + + match = re.search(r"https?://[^\s\"'<>\)\]\}]+", text) + + return match.group(0) if match else None + +def _test_mcp_server(server, db) -> dict: + """ + Test connectivity to an MCP server and persist the result. + Supports stdio, sse and streamable_http transports. + Returns a result dict with success/message/details. + """ + result = {"success": False, "message": "", "details": {}} + transport = (server.transport_type or "stdio").replace("-", "_") + + try: + if transport in ("streamable_http", "streamablehttp"): + url = server.url + if not url: + result["message"] = "Streamable HTTP endpoint URL is required" + else: + import httpx + try: + # A real streamable-http MCP endpoint accepts POST JSON-RPC. + # Some implementations also accept GET for SSE events. + # Try GET first (cheapest), fall back to a minimal POST. + resp = httpx.get(url, timeout=10.0, follow_redirects=True) + if resp.status_code == 405: + resp = httpx.post( + url, + json={"jsonrpc": "2.0", "method": "initialize", "id": 1}, + timeout=10.0, + follow_redirects=True, + ) + # Any HTTP response (including 4xx) means the host is reachable. + result["success"] = resp.status_code < 500 + result["message"] = f"HTTP responded with {resp.status_code}" + result["details"]["status_code"] = resp.status_code + result["details"]["content_type"] = resp.headers.get("content-type", "") + except httpx.ConnectError as e: + result["message"] = f"Connection refused - server may be down: {str(e)}" + except httpx.TimeoutException: + result["message"] = "Connection timed out" + except Exception as e: + result["message"] = f"HTTP request failed: {str(e)}" + + elif transport == "sse": + sse_url = server.url + if not sse_url: + result["message"] = "SSE URL is required" + else: + import httpx + try: + response = httpx.get(sse_url, timeout=10.0, follow_redirects=True) + result["success"] = response.status_code < 500 + result["message"] = f"SSE endpoint responded with {response.status_code}" + result["details"]["status_code"] = response.status_code + result["details"]["content_type"] = response.headers.get("content-type", "") + except httpx.ConnectError as e: + result["message"] = f"Connection refused - server may be down: {str(e)}" + except httpx.TimeoutException: + result["message"] = "Connection timed out" + except Exception as e: + result["message"] = f"SSE request failed: {str(e)}" + + elif transport == "stdio": + cmd = server.command + if not cmd: + result["message"] = "STDIO command is required" + else: + import subprocess + try: + subprocess.run([cmd, "--version"], capture_output=True, timeout=10) + result["success"] = True + result["message"] = "Command found and executable" + except FileNotFoundError: + result["message"] = "Command not found in PATH" + except subprocess.TimeoutExpired: + result["message"] = "Command timed out" + except Exception as e: + result["message"] = "Command test failed" + result["details"]["error"] = str(e) + + else: + result["message"] = f"Unsupported transport type: {transport}" + + # Persist status + if result["success"]: + server.status = "connected" + server.last_connected_at = datetime.now() + else: + server.status = "error" + server.updated_at = datetime.now() + db.commit() + + except Exception as e: + result["message"] = str(e) + result["details"]["error"] = str(e) + server.status = "error" + server.updated_at = datetime.now() + db.commit() + + return result +def _sync_market_mcp_to_servers(db: Session): + + """ + + Synchronize MarketApp MCP records (type='mcp') into mcp_servers table. + + Uses MarketApp.id as MCPServer.id so detail/edit endpoints can work directly. + + """ + + from ...models import MarketApp, MCPServer + + market_mcps = db.query(MarketApp).filter( + + MarketApp.type == "mcp", + + MarketApp.status == "published", + + ).all() + + existing_ids = {row[0] for row in db.query(MCPServer.id).all()} + + for ma in market_mcps: + + if ma.id in existing_ids: + # Update URL if the server record has no endpoint yet + existing = db.query(MCPServer).filter(MCPServer.id == ma.id).first() + if existing and not existing.url: + url = _extract_url_from_text(ma.skill_content or "") or _extract_url_from_text(ma.description or "") or "" + if not url and ma.agent_endpoint: + url = ma.agent_endpoint + if url: + existing.url = url + existing.updated_at = datetime.now() + db.commit() + continue + + transport = ma.mcp_transport or "streamable_http" + + code = ma.skill_content or "" + + url = _extract_url_from_text(code) or _extract_url_from_text(ma.description or "") or "" + + command = "" + + if not url and ma.agent_endpoint: + + url = ma.agent_endpoint + + now = datetime.now() + + server = MCPServer( + + id=ma.id, + + name=ma.name, + + description=ma.description or "", + + transport_type=transport, + + command=command, + + args="{}", + + env_vars="{}", + + url=url, + + headers="{}", + + is_enabled=True, + + status="disconnected", + + tools_count=0, + + resources_count=0, + + created_at=ma.created_at or now, + + updated_at=ma.updated_at or now, + + ) + + db.add(server) + + if market_mcps: + + db.commit() + +# ============================================================ + +# API Endpoints + +# ============================================================ + +@router.get("/stats/overview", summary="Get MCP server statistics") + +async def get_mcp_stats( + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """Get overview statistics for MCP servers""" + + from ...models import MCPServer + + _sync_market_mcp_to_servers(db) + + total = db.query(func.count(MCPServer.id)).scalar() or 0 + + enabled = db.query(func.count(MCPServer.id)).filter(MCPServer.is_enabled == True).scalar() or 0 + + connected = db.query(func.count(MCPServer.id)).filter(MCPServer.status == "connected").scalar() or 0 + + # By transport type + + by_transport = {} + + rows = db.query(MCPServer.transport_type, func.count(MCPServer.id)).group_by(MCPServer.transport_type).all() + + for t, c in rows: + + by_transport[t or "unknown"] = c + + # By status + + by_status = {} + + rows = db.query(MCPServer.status, func.count(MCPServer.id)).group_by(MCPServer.status).all() + + for s, c in rows: + + by_status[s or "unknown"] = c + + return { + + "code": 0, + + "data": { + + "total_servers": total, + + "enabled_servers": enabled, + + "connected_servers": connected, + + "by_transport": by_transport, + + "by_status": by_status, + + }, + + "message": "ok", + + } + +@router.get("/", summary="List MCP servers") + +async def list_mcp_servers( + + page: int = Query(1, ge=1), + + page_size: int = Query(20, ge=1, le=100), + + search: Optional[str] = Query(None), + + transport_type: Optional[str] = Query(None), + + status: Optional[str] = Query(None), + + is_enabled: Optional[bool] = Query(None), + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """List all MCP servers with pagination and filters""" + + from ...models import MCPServer + + _sync_market_mcp_to_servers(db) + + query = db.query(MCPServer) + + if search: + + query = query.filter( + + or_(MCPServer.name.contains(search), MCPServer.description.contains(search)) + + ) + + if transport_type: + + query = query.filter(MCPServer.transport_type == transport_type) + + if status: + + query = query.filter(MCPServer.status == status) + + if is_enabled is not None: + + query = query.filter(MCPServer.is_enabled == is_enabled) + + total = query.count() + + servers = query.order_by(desc(MCPServer.created_at)).offset((page - 1) * page_size).limit(page_size).all() + + return { + + "code": 0, + + "data": { + + "total": total, + + "page": page, + + "page_size": page_size, + + "items": [server_to_dict(s) for s in servers], + + }, + + "message": "ok", + + } + +@router.get("/{server_id}", summary="Get MCP server details") + +async def get_mcp_server( + + server_id: str, + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """Get a single MCP server by ID""" + + from ...models import MCPServer + + _sync_market_mcp_to_servers(db) + + server = db.query(MCPServer).filter(MCPServer.id == server_id).first() + + if not server: + + raise HTTPException(status_code=404, detail="MCP server not found") + + return {"code": 0, "data": server_to_dict(server), "message": "ok"} + +@router.post("/", summary="Create MCP server") + +async def create_mcp_server( + + data: MCPServerCreate, + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """Register a new MCP server""" + + from ...models import MCPServer + + # Check name uniqueness + + existing = db.query(MCPServer).filter(MCPServer.name == data.name).first() + + if existing: + + raise HTTPException(status_code=400, detail="MCP server with this name already exists") + + now = datetime.now() + + server = MCPServer( + + id=gen_uuid(), + + name=data.name, + + description=data.description or "", + + transport_type=data.transport_type, + + command=data.command or "", + + args=json.dumps(data.args) if data.args else "{}", + + env_vars=json.dumps(data.env_vars) if data.env_vars else "{}", + + url=data.url or "", + + headers=json.dumps(data.headers) if data.headers else "{}", + + is_enabled=data.is_enabled if data.is_enabled is not None else True, + + status="disconnected", + + created_at=now, + + updated_at=now, + + ) + + db.add(server) + + db.commit() + + db.refresh(server) + + logger.info(f"MCP server created: {server.name} ({server.id})") + + return {"code": 0, "data": server_to_dict(server), "message": "MCP server created"} + +@router.put("/{server_id}", summary="Update MCP server") + +async def update_mcp_server( + + server_id: str, + + data: MCPServerUpdate, + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """Update an existing MCP server""" + + from ...models import MCPServer + + server = db.query(MCPServer).filter(MCPServer.id == server_id).first() + + if not server: + + raise HTTPException(status_code=404, detail="MCP server not found") + + update_fields = data.model_dump(exclude_unset=True) + + # Handle JSON fields + + for json_field in ["args", "env_vars", "headers"]: + + if json_field in update_fields and update_fields[json_field] is not None: + + update_fields[json_field] = json.dumps(update_fields[json_field]) + + for key, value in update_fields.items(): + + if value is not None: + + setattr(server, key, value) + + server.updated_at = datetime.now() + + db.commit() + + db.refresh(server) + + logger.info(f"MCP server updated: {server.name} ({server.id})") + + return {"code": 0, "data": server_to_dict(server), "message": "MCP server updated"} + +@router.delete("/{server_id}", summary="Delete MCP server") + +async def delete_mcp_server( + + server_id: str, + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """Delete an MCP server""" + + from ...models import MCPServer + + server = db.query(MCPServer).filter(MCPServer.id == server_id).first() + + if not server: + + raise HTTPException(status_code=404, detail="MCP server not found") + + name = server.name + + db.delete(server) + + db.commit() + + logger.info(f"MCP server deleted: {name} ({server_id})") + + return {"code": 0, "data": None, "message": "MCP server deleted"} + +@router.put("/{server_id}/toggle", summary="Enable/Disable MCP server") + +async def toggle_mcp_server( + + server_id: str, + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """Toggle MCP server enabled state""" + + from ...models import MCPServer + + server = db.query(MCPServer).filter(MCPServer.id == server_id).first() + + if not server: + + raise HTTPException(status_code=404, detail="MCP server not found") + + server.is_enabled = not server.is_enabled + + server.updated_at = datetime.now() + + db.commit() + + logger.info(f"MCP server toggled: {server.name} -> enabled={server.is_enabled}") + + return { + + "code": 0, + + "data": {"id": server.id, "is_enabled": server.is_enabled}, + + "message": "MCP server toggled", + + } + +@router.post("/{server_id}/test", summary="Test MCP server connection") +async def test_mcp_connection( + server_id: str, + db: Session = Depends(get_db), + admin_user=Depends(get_current_admin), + test_data: Optional[MCPServerConnectionTest] = None, +): + """Test connection to an MCP server""" + from ...models import MCPServer + server = db.query(MCPServer).filter(MCPServer.id == server_id).first() + if not server: + raise HTTPException(status_code=404, detail="MCP server not found") + + # If test_data provides an override URL/transport, apply it temporarily + original_url = server.url + original_transport = server.transport_type + if test_data: + if test_data.transport_type: + server.transport_type = test_data.transport_type + if test_data.url: + server.url = test_data.url + if test_data.command: + server.command = test_data.command + + result = _test_mcp_server(server, db) + + # Restore original values if no override was explicitly persisted + if test_data and test_data.url is None: + server.url = original_url + if test_data and test_data.transport_type is None: + server.transport_type = original_transport + if test_data: + db.commit() + + return {"code": 0, "data": result, "message": "ok"} + +@router.get("/{server_id}/tools", summary="List MCP server tools") + +async def list_mcp_tools( + + server_id: str, + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """List tools exposed by an MCP server (placeholder/mock)""" + + from ...models import MCPServer + + server = db.query(MCPServer).filter(MCPServer.id == server_id).first() + + if not server: + + raise HTTPException(status_code=404, detail="MCP server not found") + + # In a real implementation, this would connect to the MCP server and call tools/list + + # For now, return mock data based on common MCP tools + + mock_tools = [] + + if server.transport_type == "stdio": + + mock_tools = [ + + {"name": "read_file", "description": "Read contents of a file"}, + + {"name": "write_file", "description": "Write contents to a file"}, + + {"name": "execute_command", "description": "Execute a shell command"}, + + {"name": "search_content", "description": "Search for content in files"}, + + ] + + elif server.transport_type == "sse": + + mock_tools = [ + + {"name": "web_search", "description": "Search the web"}, + + {"name": "fetch_url", "description": "Fetch content from URL"}, + + {"name": "api_request", "description": "Make API requests"}, + + ] + + return { + + "code": 0, + + "data": { + + "server_name": server.name, + + "tools": mock_tools, + + "count": len(mock_tools), + + }, + + "message": "ok", + + } + +@router.get("/{server_id}/resources", summary="List MCP server resources") + +async def list_mcp_resources( + + server_id: str, + + db: Session = Depends(get_db), + + admin_user=Depends(get_current_admin), + +): + + """List resources exposed by an MCP server (placeholder/mock)""" + + from ...models import MCPServer + + server = db.query(MCPServer).filter(MCPServer.id == server_id).first() + + if not server: + + raise HTTPException(status_code=404, detail="MCP server not found") + + mock_resources = [ + + {"uri": f"{server.name}://docs/readme", "name": "Documentation", "mime_type": "text/markdown"}, + + {"uri": f"{server.name}://config/settings", "name": "Configuration", "mime_type": "application/json"}, + + ] + + return { + + "code": 0, + + "data": { + + "server_name": server.name, + + "resources": mock_resources, + + "count": len(mock_resources), + + }, + + "message": "ok", + + } + diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index cd0e157..5130939 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -1940,22 +1940,36 @@ async def _generate_media_task(ai_msg_id: str, ai_model_id: str, prompt: str, us + + + + + logger.info(f"[_generate_media_task] 启动 media 生成 msg={ai_msg_id} model={ai_model_id} prompt={prompt[:60]!r}") + caps = [c.lower() for c in (ai_model.capabilities or [])] + is_video = "video" in caps + timeout_seconds = 300 if is_video else 90 try: + result = await asyncio.wait_for( + _generate_media(db, ai_model, prompt, user_id), + timeout=timeout_seconds + ) - result = await _generate_media(db, ai_model, prompt, user_id) - - - + except asyncio.TimeoutError: + logger.warning(f"[_generate_media_task] media 生成超时 msg={ai_msg_id} is_video={is_video} timeout={timeout_seconds}s") + _media_name = "视频" if is_video else "图像" + result = f"⚠️ 媒体生成超时：{_media_name}生成耗时超过 {timeout_seconds} 秒，请稍后重试。" except Exception as e: - - - + logger.warning(f"[_generate_media_task] media 生成异常 msg={ai_msg_id}: {e}", exc_info=True) result = f"⚠️ 媒体生成失败：{str(e)[:200]}" + + + + # 回写消息内容(占位 → 最终富文本/错误) @@ -5854,6 +5868,19 @@ def _build_system_prompt( ) + # 高优先级覆盖规则：显式选中工具时，回答必须基于该工具资料 + if (skills and not auto_discovered) or (mcp): + system_parts.append( + "【最高优先级覆盖规则】\n" + "用户在本轮已明确选择了工具（见上文【已选技能资料】或【用户已选择 MCP 应用】）。\n" + "如果用户的问题是在询问这些已选工具/技能/应用本身的介绍、功能、能做什么、怎么用、特点或用法，" + "你必须只基于上文提供的该工具具体资料作答；\n" + "不要引用《平台知识库》中关于技能市场、MCP 协议或平台整体能力的通用描述；" + "不要列举其它未选中的技能或 MCP 作为示例；也不要说“不知道”或“无法回答”。\n" + "若用户问题与已选工具本身无关，则按常规方式回答。" + ) + +