fix: 显式选中技能/MCP 时优先按工具资料回答 + 媒体任务超时/日志

- _build_system_prompt 末尾增加【最高优先级覆盖规则】：用户明确选择技能/MCP 并询问其介绍/功能/用法时，
  必须只基于该工具资料作答，禁止引用平台知识库中技能市场/MCP/平台能力的通用描述。
- _generate_media_task 增加启动/结束日志，对图像生成加 90s 超时、视频 300s 超时，避免异步任务挂死导致前端一直显示“生成中”。

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",
+
+    }
+

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"
+            "若用户问题与已选工具本身无关，则按常规方式回答。"
+        )
+
+