From 47d38ca173bc5aac061bec49552c8dc172b22766 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Fri, 17 Jul 2026 09:43:43 +0800 Subject: [PATCH] =?UTF-8?q?fix(MCP):=20=E4=BF=AE=E5=A4=8D=E5=AF=BC?= =?UTF-8?q?=E5=85=A5MCP=E6=97=B6=E4=BC=A0=E8=BE=93=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E4=B8=A2=E5=A4=B1=E5=AF=BC=E8=87=B4=E6=97=A0=E6=B3=95=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 前端 MarketPage 提交时随 download_url 发送粘贴的 MCP 代码(handleSubmit+handleCreateApp) - 后端新增 parse_mcp_config 解析标准 JSON 配置, 提取 url/headers/transport - upload_app 与 _sync 使用解析器落库 mcp_servers 并自动连通性测试 - _test_mcp_server 连接测试携带 headers 鉴权头 - 优化 MCP 代码输入框占位符为真实配置格式 --- backend/python/app/api/admin/mcp_admin.py | 4588 +++++++++++++++++---- backend/python/app/api/market.py | 1639 ++++---- frontend/src/pages/front/MarketPage.tsx | 62 +- 3 files changed, 4548 insertions(+), 1741 deletions(-) diff --git a/backend/python/app/api/admin/mcp_admin.py b/backend/python/app/api/admin/mcp_admin.py index 1b48b2f..e085456 100644 --- a/backend/python/app/api/admin/mcp_admin.py +++ b/backend/python/app/api/admin/mcp_admin.py @@ -1,906 +1,3682 @@ -""" - -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() - # 注意：已存在记录时保留其 is_enabled（后台禁用态不被同步覆盖） - 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=(ma.status != "removed"), - - 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 - - # 将源头 MarketApp(type='mcp')状态置为 removed，避免 _sync 在下次查询时又把 mcp_servers 复活（market_apps 被 app_installations 等外键引用，不能直接删除） - try: - from ...models import MarketApp - market_app = db.query(MarketApp).filter(MarketApp.id == server_id).first() - if market_app: - market_app.status = "removed" - market_app.updated_at = datetime.now() - db.add(market_app) - except Exception: - pass - 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() - # 同步源头 MarketApp 状态：禁用 -> status='removed'，启用 -> status='published' - try: - from ...models import MarketApp - market_app = db.query(MarketApp).filter(MarketApp.id == server_id).first() - if market_app: - market_app.status = "removed" if not server.is_enabled else "published" - market_app.updated_at = datetime.now() - db.add(market_app) - except Exception: - pass - - 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", - - } +""" + + + + + + + +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 _normalize_transport(t: str) -> str: + """Map various transport spellings to the canonical stored value.""" + if not t: + return "streamable_http" + t = str(t).lower().replace("-", "_") + if "stdio" in t: + return "stdio" + if "sse" in t: + return "sse" + if "http" in t or "stream" in t: + return "streamable_http" + return "streamable_http" + + +def parse_mcp_config(raw: str) -> dict: + """Parse a pasted MCP config (standard JSON or free text) into + {url, headers, transport_type}. Best-effort; never raises. + + Handles: + - Standard mcp config: {"mcpServers": {"name": {"url":..., "headers":..., "type":...}}} + - Alias "servers" key + - Top-level single-server object {"url":..., "type":..., "headers":...} + - Free text containing a bare URL (regex fallback) + """ + result = {"url": "", "headers": {}, "transport_type": "streamable_http"} + if not raw: + return result + raw = raw.strip() + try: + cfg = json.loads(raw) + if isinstance(cfg, dict): + servers = None + if isinstance(cfg.get("mcpServers"), dict): + servers = cfg["mcpServers"] + elif isinstance(cfg.get("servers"), dict): + servers = cfg["servers"] + elif "url" in cfg or "type" in cfg or "transport" in cfg: + servers = {"_default": cfg} + if servers: + entry = next(iter(servers.values())) + if isinstance(entry, dict): + result["url"] = entry.get("url") or result["url"] + h = entry.get("headers") or {} + if isinstance(h, dict): + result["headers"] = {str(k): str(v) for k, v in h.items()} + t = str(entry.get("type") or entry.get("transport") or "").lower().replace("-", "_") + if t: + result["transport_type"] = _normalize_transport(t) + except Exception: + pass + if not result["url"]: + result["url"] = _extract_url_from_text(raw) or "" + if not isinstance(result["headers"], dict): + result["headers"] = {} + return result + + +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("-", "_") + req_headers = {} + try: + if server.headers: + import json as _json + _h = _json.loads(server.headers) if isinstance(server.headers, str) else server.headers + if isinstance(_h, dict): + req_headers = {str(k): str(v) for k, v in _h.items()} + except Exception: + req_headers = {} + + + + + + + + + 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, headers=req_headers, timeout=10.0, follow_redirects=True) + + + + if resp.status_code == 405: + + + + resp = httpx.post( + url, + headers=req_headers, + 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, headers=req_headers, 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() + + + + # 注意：已存在记录时保留其 is_enabled（后台禁用态不被同步覆盖） + + + + db.commit() + + + + continue + + + + + + + + parsed = parse_mcp_config(ma.skill_content or "") + + + + + + + + + transport = parsed["transport_type"] or ma.mcp_transport or "streamable_http" + + + + + + + + code = ma.skill_content or "" + + + + + + + + url = parsed["url"] 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=json.dumps(parsed["headers"], ensure_ascii=False) if parsed["headers"] else "{}", + + + + + + + + is_enabled=(ma.status != "removed"), + + + + + + + + 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 + + + + + + + + # 将源头 MarketApp(type='mcp')状态置为 removed，避免 _sync 在下次查询时又把 mcp_servers 复活（market_apps 被 app_installations 等外键引用，不能直接删除） + + + try: + + + + from ...models import MarketApp + + + + market_app = db.query(MarketApp).filter(MarketApp.id == server_id).first() + + + + if market_app: + + + + market_app.status = "removed" + + + market_app.updated_at = datetime.now() + + + db.add(market_app) + + + except Exception: + + + + pass + + + + 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() + + + + # 同步源头 MarketApp 状态：禁用 -> status='removed'，启用 -> status='published' + + + + try: + + + + from ...models import MarketApp + + + + market_app = db.query(MarketApp).filter(MarketApp.id == server_id).first() + + + + if market_app: + + + + market_app.status = "removed" if not server.is_enabled else "published" + + + + market_app.updated_at = datetime.now() + + + + db.add(market_app) + + + + except Exception: + + + + pass + + + + + + + + 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/market.py b/backend/python/app/api/market.py index e005840..a222055 100644 --- a/backend/python/app/api/market.py +++ b/backend/python/app/api/market.py @@ -1,818 +1,821 @@ -from fastapi import APIRouter, Depends, HTTPException, Query -from sqlalchemy.orm import Session -from sqlalchemy import and_, desc, asc, func -from pydantic import BaseModel, Field -from typing import Optional -import uuid -from datetime import datetime -import httpx -from urllib.parse import urlparse -import re - -from ..database import get_db -from ..models import ( - User, MarketApp, AppPurchase, AppReview, AppInstallation, Order, - AdBanner, MCPServer, gen_uuid -) -from ..auth import get_current_user, get_current_user_id, get_current_user_id_optional -from ..middleware.log_service import LogService -from app.api.admin.mcp_admin import _test_mcp_server - -router = APIRouter() - - -def _extract_url_from_text(text: Optional[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 - - -# ===== Agent 接入前连通性 / 鉴权校验 ===== -def _normalize_dify_url(endpoint: str) -> str: - ep = endpoint.rstrip("/") - if ep.endswith("/chat-messages"): - return ep - if ep.endswith("/v1/chat-messages"): - return ep - if ep.endswith("/v1"): - return ep + "/chat-messages" - return ep + "/v1/chat-messages" - - -def _short_err(e: Exception) -> str: - return str(e)[:160] - - -def _post_follow(url: str, headers: dict, json_body: dict, timeout: float = 15.0): - """POST 并手动跟随 301/302/307/308 重定向(**保持 POST 方法与请求体**)。 - - 背景：Dify 若填写 http:// 地址会 301 永久重定向到 https://；httpx 内建 - follow_redirects 对 301/302 会把 POST 降级为 GET(浏览器行为)，导致对 - /v1/chat-messages 变成 GET → 405 method_not_allowed。故手动跟随并保持 POST。 - 容器内 env 有坏代理(502)，必须 trust_env=False 直连。 - """ - cur = url - last = None - for _ in range(4): - last = httpx.post(cur, headers=headers, json=json_body, timeout=timeout, - trust_env=False, follow_redirects=False) - if last.status_code in (301, 302, 307, 308): - loc = last.headers.get("location") - if not loc: - return last - cur = loc if loc.startswith("http") else str(httpx.URL(cur).join(loc)) - continue - return last - return last - - -def verify_agent_connectivity(platform: str, endpoint: str, api_key: str) -> dict: - """ - 接入 Agent 前的连通性 + 鉴权校验。 - 返回 {ok, verified, detail} - ok: 是否通过（dify=鉴权通过；n8n/coze/其他=地址可达） - verified: 是否做了真正的鉴权验证（dify 为 True，其余为 False） - """ - ep = (endpoint or "").strip() - if not ep or not ep.startswith(("http://", "https://")): - return {"ok": False, "verified": False, "detail": "Endpoint 必须以 http:// 或 https:// 开头"} - platform = (platform or "dify").lower() - - # 关键：容器内 env 有坏代理(502)，必须 trust_env=False 直连 - if platform == "dify": - url = _normalize_dify_url(ep) - payload = { - "inputs": {}, - "query": "连通性测试", - "user": "__verify__", - "response_mode": "blocking", - } - headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} - try: - r = _post_follow(url, headers, payload, timeout=15.0) - except Exception as e: - return {"ok": False, "verified": False, "detail": f"无法连接 Dify 地址：{_short_err(e)}"} - if r.status_code == 200: - return {"ok": True, "verified": True, "detail": "连通且 API Key 有效"} - if r.status_code in (401, 403): - return {"ok": False, "verified": False, "detail": "API Key 无效（401/403）"} - if r.status_code == 404: - return {"ok": False, "verified": False, - "detail": "地址路径不正确（404），应为 .../v1/chat-messages"} - if r.status_code in (301, 302, 307, 308): - loc = r.headers.get("location") or "" - return {"ok": False, "verified": False, - "detail": f"地址被重定向（{r.status_code}）到 {loc or '未知'}；" - f"请直接填写重定向后的地址（通常把 http:// 改为 https://）"} - if r.status_code == 400: - # 400 需区分两种情况： - # (a) 到达应用且鉴权通过，但 Dify 应用/工作流内部执行报错(invalid_param + Run failed) - # (b) 纯参数错误 - body = (r.text or "") - low = body.lower() - if ("run failed" in low or "error code" in low or "workflow" in low - or "node" in low or "-500" in body): - return {"ok": True, "verified": True, - "detail": "连通且 API Key 有效；但你的 Dify 应用内部执行报错：" - f"{body[:180]}。这是你自建 Dify 应用/工作流本身的问题" - "（例如代码节点沙箱缺少 python3），请在 Dify 侧修复。"} - return {"ok": False, "verified": False, "detail": f"参数错误（400）：{body[:180]}"} - return {"ok": False, "verified": False, "detail": f"Dify 返回状态码 {r.status_code}"} - - # n8n / coze / 其他：仅做可达性探测（HEAD，避免触发 webhook 副作用） - try: - r = httpx.head(ep, timeout=12.0, trust_env=False, follow_redirects=True) - if r.status_code < 500: - return {"ok": True, "verified": False, - "detail": f"地址可连通（HTTP {r.status_code}），运行时再做协议级鉴权"} - return {"ok": False, "verified": False, "detail": f"服务返回 {r.status_code}，地址可能不可用"} - except Exception as e: - return {"ok": False, "verified": False, "detail": f"地址不可达：{_short_err(e)}"} - - -class VerifyAgentRequest(BaseModel): - platform: Optional[str] = None - endpoint: str - api_key: str - - -@router.post("/apps/verify-agent") -def verify_agent( - req: VerifyAgentRequest, - current_user: User = Depends(get_current_user), -): - """接入 Agent 前手动测试连通性 / 鉴权，供前台「测试连接」按钮调用。""" - result = verify_agent_connectivity(req.platform or "dify", req.endpoint, req.api_key) - return {"code": 0, "data": result, "message": "ok"} - - -@router.get("/ads") -def list_public_ads(db: Session = Depends(get_db)): - """ - 前台首页「精选推荐」广告位：返回已上架(active)广告，按排序升序，无需登录。 - 供 LandingPage 轮播调用，无鉴权依赖。 - """ - ads = ( - db.query(AdBanner) - .filter(AdBanner.status == "active") - .order_by(AdBanner.sort_order.asc()) - .all() - ) - return [ - { - "id": a.id, - "title": a.title, - "image_url": a.image_url, - "link_app_id": a.link_app_id, - "link_app_name": a.link_app_name, - "sort_order": a.sort_order, - "click_count": a.click_count, - "status": a.status, - "created_at": a.created_at.isoformat() if a.created_at else None, - } - for a in ads - ] - - -class UploadAppRequest(BaseModel): - name: str = Field(min_length=1, max_length=200) - description: Optional[str] = None - icon: Optional[str] = None - type: Optional[str] = None # skill, agent, mcp - source: Optional[str] = None - price: float = Field(default=0, ge=0) - download_url: Optional[str] = None - category: Optional[str] = None - tags: Optional[str] = None - is_member_free: Optional[bool] = False - join_promotion: Optional[bool] = False - agent_endpoint: Optional[str] = None - api_key: Optional[str] = None - platform: Optional[str] = None - - -class PurchaseAppRequest(BaseModel): - amount: Optional[float] = None - - -class ReviewAppRequest(BaseModel): - rating: int = Field(ge=1, le=5) - content: Optional[str] = None - - -@router.get("/apps") -def list_apps( - type: Optional[str] = Query(None), - source: Optional[str] = Query(None), - category: Optional[str] = Query(None), - search: Optional[str] = Query(None), - sort: str = Query(default="latest"), - page: int = Query(default=1, ge=1), - page_size: int = Query(default=20, ge=1, le=100), - db: Session = Depends(get_db), - user_id: Optional[str] = Depends(get_current_user_id_optional), -): - query = db.query(MarketApp).filter(MarketApp.status == "published") - - if type: - query = query.filter(MarketApp.type == type) - if source: - query = query.filter(MarketApp.source == source) - if category: - query = query.filter(MarketApp.type == category) - if search: - query = query.filter( - and_( - MarketApp.name.ilike(f"%{search}%") - | MarketApp.description.ilike(f"%{search}%") - ) - ) - - if sort == "popular": - query = query.order_by(desc(MarketApp.install_count)) - elif sort == "rating": - query = query.order_by(desc(MarketApp.rating_avg)) - else: - query = query.order_by(desc(MarketApp.created_at)) - - total = query.count() - apps = query.offset((page - 1) * page_size).limit(page_size).all() - - # 如果用户已登录，获取该用户已安装的应用ID集合（确保按用户隔离） - installed_app_ids = set() - if user_id: - installations = db.query(AppInstallation.app_id).filter( - AppInstallation.user_id == user_id - ).all() - installed_app_ids = {inst[0] for inst in installations} - - # 查询已安装状态时始终带 user_id 过滤 - - result = [] - for app in apps: - # 用户自己创建/导入的应用应始终视为已安装（避免安装记录缺失导致显示未安装） - is_user_owned = bool(user_id) and app.source == "user" and app.publisher_id == user_id - item = { - "id": app.id, - "name": app.name, - "description": app.description, - "icon": app.icon, - "type": app.type, - "source": app.source, - "publisher_id": app.publisher_id, - "publisher_name": app.publisher_name, - "is_member_free": app.is_member_free if app.is_member_free else False, - "join_promotion": app.join_promotion if app.join_promotion else False, - "status": app.status, - "price": float(app.price) if app.price else 0, - "install_count": app.install_count or 0, - "rating_avg": float(app.rating_avg) if app.rating_avg else 0, - "review_count": app.review_count or 0, - "created_at": str(app.created_at) if app.created_at else None, - "installed": (app.id in installed_app_ids or is_user_owned) if user_id else False, - } - result.append(item) - - return { - "code": 0, - "data": { - "total": total, - "page": page, - "page_size": page_size, - "items": result, - }, - "message": "ok", - } - - -@router.get("/apps/{app_id}") -def get_app( - app_id: str, - db: Session = Depends(get_db), - current_user: Optional[User] = Depends(get_current_user), -): - app = db.query(MarketApp).filter(MarketApp.id == app_id).first() - if not app: - raise HTTPException(status_code=404, detail="应用不存在") - - # 检查当前用户是否已购买（按用户隔离） - purchased = False - installed = False - if current_user: - purchase = db.query(AppPurchase).filter( - and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id) - ).first() - if purchase: - purchased = True - installed = purchase.status == "installed" - # 通过 AppInstallation 表检查是否已安装（按用户隔离） - installation = db.query(AppInstallation).filter( - and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id) - ).first() - if installation: - installed = True - - return { - "code": 0, - "data": { - "id": app.id, "name": app.name, "description": app.description, - "icon": app.icon, "type": app.type, "source": app.source, - "type": app.type, - "price": float(app.price) if app.price else 0, - "install_count": app.install_count or 0, - "rating_avg": float(app.rating_avg) if app.rating_avg else 0, - "review_count": app.review_count or 0, - "publisher_id": app.publisher_id, - "download_url": app.download_url, - "detail": app.detail, - "created_at": str(app.created_at), - "purchased": purchased, - "installed": installed, - }, - "message": "ok", - } - - -@router.post("/apps") -def upload_app( - req: UploadAppRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - # 接入 Agent 时先做连通性 / 鉴权校验，避免把无效连接存进库 - if (req.type or "") == "agent" and req.agent_endpoint and req.api_key: - v = verify_agent_connectivity(req.platform or "dify", req.agent_endpoint, req.api_key) - if not v["ok"]: - raise HTTPException( - status_code=400, - detail=f"Agent 连接校验未通过：{v['detail']}", - ) - - # source 判定：管理员（admin 角色）通过前台市场导入/开发的 Agent/Skill/MCP - # 统一标记为官方（official），其余用户按请求或默认为 user。 - if current_user.role == "admin": - app_source = "official" - elif req.source and req.source in ("official", "user"): - app_source = req.source - else: - app_source = "user" - - app = MarketApp( - id=gen_uuid(), - name=req.name, - description=req.description, - icon=req.icon, - type=req.type if req.type in ("skill","agent","mcp") else "skill", - source=app_source, - platform=req.platform, - price=req.price, - install_count=0, - rating_avg=0, - review_count=0, - publisher_id=current_user.id, - publisher_name=current_user.username, - is_member_free=req.is_member_free if req.is_member_free is not None else False, - join_promotion=req.join_promotion if req.join_promotion is not None else False, - agent_endpoint=req.agent_endpoint, - agent_api_key=req.api_key, - status="published", # 用户创建的应用直接发布 - created_at=datetime.now(), - updated_at=datetime.now(), - ) - db.add(app) - db.commit() - db.refresh(app) - - # MCP 应用同步注册到后台 MCP Client 管理（mcp_servers 表） - if app.type == "mcp": - transport = "streamable_http" - code = req.download_url or "" - if code: - app.skill_content = code - app.mcp_transport = transport - db.commit() - db.refresh(app) - - existing_server = db.query(MCPServer).filter(MCPServer.id == app.id).first() - if not existing_server: - url = _extract_url_from_text(code) or _extract_url_from_text(req.description or "") or "" - server = MCPServer( - id=app.id, - name=app.name, - description=app.description or "", - transport_type=transport, - command="", - args="{}", - env_vars="{}", - url=url, - headers="{}", - is_enabled=True, - status="disconnected", - tools_count=0, - resources_count=0, - created_at=app.created_at or datetime.now(), - updated_at=app.updated_at or datetime.now(), - ) - db.add(server) - db.commit() - - # 导入后自动做一次连通性测试，状态会同步为 connected / error - _test_mcp_server(server, db) - - # 用户创建/导入的应用自动标记为已安装 - existing_install = db.query(AppInstallation).filter( - AppInstallation.user_id == current_user.id, - AppInstallation.app_id == app.id, - ).first() - if not existing_install: - installation = AppInstallation( - user_id=current_user.id, - app_id=app.id, - installed_at=datetime.now(), - ) - db.add(installation) - db.commit() - - LogService.log_operation( - user_id=current_user.id, username=current_user.username, action="upload_app", - target="market_app", target_id=app.id, - detail=f"上传应用 {app.name}", - ) - - return { - "code": 0, - "data": { - "id": app.id, "name": app.name, "status": app.status, - "created_at": str(app.created_at), - }, - "message": "提交成功，等待审核", - } - - -@router.post("/apps/{app_id}/purchase") -def purchase_app( - app_id: str, - req: PurchaseAppRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - app = db.query(MarketApp).filter( - and_(MarketApp.id == app_id, MarketApp.status == "published") - ).first() - if not app: - raise HTTPException(status_code=404, detail="应用不存在或未上架") - - # 检查是否已购买 - existing = db.query(AppPurchase).filter( - and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id) - ).first() - if existing and existing.status in ("purchased", "installed"): - raise HTTPException(status_code=400, detail="您已购买该应用") - - amount = req.amount or app.price - - # 创建订单 - order = Order( - id=gen_uuid(), - user_id=current_user.id, - order_type="app", - item_id=app_id, - item_name=app.name, - amount=amount, - status="pending", - created_at=datetime.now(), - ) - db.add(order) - db.flush() - - # 创建购买记录 - purchase = AppPurchase( - id=gen_uuid(), - user_id=current_user.id, - app_id=app_id, - order_id=order.id, - status="purchased", - purchased_at=datetime.now(), - ) - db.add(purchase) - - # 更新应用下载量 - app.install_count = (app.install_count or 0) + 1 - app.updated_at = datetime.now() - - db.commit() - - return { - "code": 0, - "data": { - "order_id": order.id, - "purchase_id": purchase.id, - "amount": amount, - "status": "purchased", - }, - "message": "购买成功", - } - - -@router.post("/apps/{app_id}/install") -def install_app( - app_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - app = db.query(MarketApp).filter(MarketApp.id == app_id).first() - if not app: - raise HTTPException(status_code=404, detail="应用不存在") - - # 检查是否已安装（通过 AppInstallation 表） - existing_install = db.query(AppInstallation).filter( - and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id) - ).first() - if existing_install: - raise HTTPException(status_code=400, detail="应用已安装") - - # 检查是否已购买（付费应用需要先购买） - purchase = db.query(AppPurchase).filter( - and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id) - ).first() - - # 对于付费应用，必须已购买才能安装 - if float(app.price or 0) > 0: - if not purchase: - raise HTTPException(status_code=403, detail="请先购买应用") - if purchase.status == "installed": - raise HTTPException(status_code=400, detail="应用已安装") - - # 创建安装记录 - installation = AppInstallation( - id=gen_uuid(), - user_id=current_user.id, - app_id=app_id, - installed_at=datetime.now(), - ) - db.add(installation) - - # 如果有购买记录，更新状态为已安装 - if purchase: - purchase.status = "installed" - purchase.installed_at = datetime.now() - - # 更新应用安装量 - app.install_count = (app.install_count or 0) + 1 - app.updated_at = datetime.now() - - db.commit() - - return { - "code": 0, - "data": {"installation_id": installation.id, "status": "installed"}, - "message": "安装成功", - } - - -@router.delete("/apps/{app_id}/uninstall") -def uninstall_app( - app_id: str, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - """卸载应用 - 删除当前用户的安装记录（按用户隔离）""" - app = db.query(MarketApp).filter(MarketApp.id == app_id).first() - if not app: - raise HTTPException(status_code=404, detail="应用不存在") - - # 查找当前用户的安装记录（按用户隔离） - installation = db.query(AppInstallation).filter( - and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id) - ).first() - if not installation: - raise HTTPException(status_code=400, detail="应用未安装") - - # 删除安装记录 - db.delete(installation) - - # 问题4修复: 将AppPurchase状态从installed重置为purchased - purchase = db.query(AppPurchase).filter( - and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id) - ).first() - if purchase and purchase.status == "installed": - purchase.status = "purchased" - - # 更新应用安装量（减1，最小为0） - if app.install_count and app.install_count > 0: - app.install_count = app.install_count - 1 - app.updated_at = datetime.now() - - db.commit() - - LogService.log_operation( - user_id=current_user.id, username=current_user.username, action="uninstall_app", - target="market_app", target_id=app_id, - detail=f"卸载应用 {app.name}", - ) - - return { - "code": 0, - "message": "卸载成功", - } - - -@router.post("/apps/{app_id}/review") -def review_app( - app_id: str, - req: ReviewAppRequest, - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - app = db.query(MarketApp).filter(MarketApp.id == app_id).first() - if not app: - raise HTTPException(status_code=404, detail="应用不存在") - - # 检查是否已评价 - existing = db.query(AppReview).filter( - and_(AppReview.app_id == app_id, AppReview.user_id == current_user.id) - ).first() - if existing: - # 更新已有评价 - old_rating = existing.rating - existing.rating = req.rating - existing.content = req.content - existing.updated_at = datetime.now() - - # 重新计算平均分 (总数不变，减去旧评分加新评分) - if app.review_count and app.review_count > 0: - total_rating = float(app.rating_avg or 0) * app.review_count - total_rating = total_rating - old_rating + req.rating - app.rating_avg = total_rating / app.review_count - else: - app.rating_avg = req.rating - app.review_count = 1 - else: - # 创建新评价 - review = AppReview( - id=gen_uuid(), - app_id=app_id, - user_id=current_user.id, - rating=req.rating, - content=req.content, - created_at=datetime.now(), - updated_at=datetime.now(), - ) - db.add(review) - - # 重新计算平均分 - app.review_count = (app.review_count or 0) + 1 - if app.review_count == 1: - app.rating_avg = req.rating - else: - total_rating = float(app.rating_avg or 0) * (app.review_count - 1) + req.rating - app.rating_avg = round(total_rating / app.review_count, 1) - - app.updated_at = datetime.now() - db.commit() - - return { - "code": 0, - "data": { - "rating_avg": float(app.rating_avg or 0), - "review_count": app.review_count or 0, - }, - "message": "评价成功", - } - - -@router.get("/my-apps") -def get_my_apps( - status: Optional[str] = Query(None), - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - query = db.query(AppPurchase).filter(AppPurchase.user_id == current_user.id) - purchases = query.order_by(desc(AppPurchase.created_at)).all() - - result = [] - for p in purchases: - app = db.query(MarketApp).filter(MarketApp.id == p.app_id).first() - if app: - result.append({ - "id": p.id, - "app_id": app.id, - "name": app.name, - "icon": app.icon, - "type": app.type, - "price": float(app.price or 0), - "purchased_at": str(p.created_at) if p.created_at else None, - }) - - return {"code": 0, "data": result, "message": "ok"} - - -@router.get("/reviews/{app_id}") -def get_reviews( - app_id: str, - page: int = Query(default=1, ge=1), - page_size: int = Query(default=20, ge=1, le=100), - db: Session = Depends(get_db), -): - query = db.query(AppReview).filter(AppReview.app_id == app_id) - total = query.count() - - reviews = ( - query.order_by(desc(AppReview.created_at)) - .offset((page - 1) * page_size) - .limit(page_size) - .all() - ) - - result = [] - for r in reviews: - user = db.query(User).filter(User.id == r.user_id).first() - result.append({ - "id": r.id, - "user_id": r.user_id, - "username": user.username if user else "unknown", - "avatar": user.avatar if user else None, - "rating": r.rating, - "content": r.content, - "created_at": str(r.created_at) if r.created_at else None, - }) - - return { - "code": 0, - "data": {"total": total, "page": page, "page_size": page_size, "items": result}, - "message": "ok", - } - - -# ============================================================ -# Agent部署资源查询接口 -# ============================================================ - -@router.get("/agent-resources") -def get_agent_resources( - db: Session = Depends(get_db), - current_user: User = Depends(get_current_user), -): - """Agent部署时需要的资源列表：可用的应用、知识库、文档""" - # 用户已发布的应用 - my_apps = db.query(MarketApp).filter( - MarketApp.publisher_id == current_user.id, - MarketApp.status == "published", - ).order_by(desc(MarketApp.updated_at)).all() - - apps_result = [] - for app in my_apps: - apps_result.append({ - "id": app.id, - "name": app.name, - "type": app.type, - "description": app.description, - "install_count": app.install_count or 0, - }) - - # 用户知识库 - from ..models import KnowledgeBase, KnowledgeDocument - kb_list = db.query(KnowledgeBase).filter( - KnowledgeBase.user_id == current_user.id - ).order_by(desc(KnowledgeBase.updated_at)).all() - - kb_result = [] - for kb in kb_list: - doc_count = db.query(KnowledgeDocument).filter( - KnowledgeDocument.knowledge_base_id == kb.id - ).count() - kb_result.append({ - "id": kb.id, - "name": kb.name, - "description": kb.description, - "document_count": doc_count, - }) - - # 用户已安装的应用（从AppPurchase） - installed = db.query(AppPurchase).filter( - AppPurchase.user_id == current_user.id - ).all() - installed_ids = [p.app_id for p in installed] - installed_apps = [] - if installed_ids: - iapps = db.query(MarketApp).filter(MarketApp.id.in_(installed_ids)).all() - for ia in iapps: - installed_apps.append({ - "id": ia.id, - "name": ia.name, - "type": ia.type, - "description": ia.description, - }) - - return { - "code": 0, - "data": { - "my_apps": apps_result, - "knowledge_bases": kb_result, - "installed_apps": installed_apps, - }, - "message": "ok", - } - +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session +from sqlalchemy import and_, desc, asc, func +from pydantic import BaseModel, Field +from typing import Optional +import uuid +from datetime import datetime +import httpx +from urllib.parse import urlparse +import re +import json + +from ..database import get_db +from ..models import ( + User, MarketApp, AppPurchase, AppReview, AppInstallation, Order, + AdBanner, MCPServer, gen_uuid +) +from ..auth import get_current_user, get_current_user_id, get_current_user_id_optional +from ..middleware.log_service import LogService +from app.api.admin.mcp_admin import _test_mcp_server, parse_mcp_config + +router = APIRouter() + + +def _extract_url_from_text(text: Optional[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 + + +# ===== Agent 接入前连通性 / 鉴权校验 ===== +def _normalize_dify_url(endpoint: str) -> str: + ep = endpoint.rstrip("/") + if ep.endswith("/chat-messages"): + return ep + if ep.endswith("/v1/chat-messages"): + return ep + if ep.endswith("/v1"): + return ep + "/chat-messages" + return ep + "/v1/chat-messages" + + +def _short_err(e: Exception) -> str: + return str(e)[:160] + + +def _post_follow(url: str, headers: dict, json_body: dict, timeout: float = 15.0): + """POST 并手动跟随 301/302/307/308 重定向(**保持 POST 方法与请求体**)。 + + 背景：Dify 若填写 http:// 地址会 301 永久重定向到 https://；httpx 内建 + follow_redirects 对 301/302 会把 POST 降级为 GET(浏览器行为)，导致对 + /v1/chat-messages 变成 GET → 405 method_not_allowed。故手动跟随并保持 POST。 + 容器内 env 有坏代理(502)，必须 trust_env=False 直连。 + """ + cur = url + last = None + for _ in range(4): + last = httpx.post(cur, headers=headers, json=json_body, timeout=timeout, + trust_env=False, follow_redirects=False) + if last.status_code in (301, 302, 307, 308): + loc = last.headers.get("location") + if not loc: + return last + cur = loc if loc.startswith("http") else str(httpx.URL(cur).join(loc)) + continue + return last + return last + + +def verify_agent_connectivity(platform: str, endpoint: str, api_key: str) -> dict: + """ + 接入 Agent 前的连通性 + 鉴权校验。 + 返回 {ok, verified, detail} + ok: 是否通过（dify=鉴权通过；n8n/coze/其他=地址可达） + verified: 是否做了真正的鉴权验证（dify 为 True，其余为 False） + """ + ep = (endpoint or "").strip() + if not ep or not ep.startswith(("http://", "https://")): + return {"ok": False, "verified": False, "detail": "Endpoint 必须以 http:// 或 https:// 开头"} + platform = (platform or "dify").lower() + + # 关键：容器内 env 有坏代理(502)，必须 trust_env=False 直连 + if platform == "dify": + url = _normalize_dify_url(ep) + payload = { + "inputs": {}, + "query": "连通性测试", + "user": "__verify__", + "response_mode": "blocking", + } + headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} + try: + r = _post_follow(url, headers, payload, timeout=15.0) + except Exception as e: + return {"ok": False, "verified": False, "detail": f"无法连接 Dify 地址：{_short_err(e)}"} + if r.status_code == 200: + return {"ok": True, "verified": True, "detail": "连通且 API Key 有效"} + if r.status_code in (401, 403): + return {"ok": False, "verified": False, "detail": "API Key 无效（401/403）"} + if r.status_code == 404: + return {"ok": False, "verified": False, + "detail": "地址路径不正确（404），应为 .../v1/chat-messages"} + if r.status_code in (301, 302, 307, 308): + loc = r.headers.get("location") or "" + return {"ok": False, "verified": False, + "detail": f"地址被重定向（{r.status_code}）到 {loc or '未知'}；" + f"请直接填写重定向后的地址（通常把 http:// 改为 https://）"} + if r.status_code == 400: + # 400 需区分两种情况： + # (a) 到达应用且鉴权通过，但 Dify 应用/工作流内部执行报错(invalid_param + Run failed) + # (b) 纯参数错误 + body = (r.text or "") + low = body.lower() + if ("run failed" in low or "error code" in low or "workflow" in low + or "node" in low or "-500" in body): + return {"ok": True, "verified": True, + "detail": "连通且 API Key 有效；但你的 Dify 应用内部执行报错：" + f"{body[:180]}。这是你自建 Dify 应用/工作流本身的问题" + "（例如代码节点沙箱缺少 python3），请在 Dify 侧修复。"} + return {"ok": False, "verified": False, "detail": f"参数错误（400）：{body[:180]}"} + return {"ok": False, "verified": False, "detail": f"Dify 返回状态码 {r.status_code}"} + + # n8n / coze / 其他：仅做可达性探测（HEAD，避免触发 webhook 副作用） + try: + r = httpx.head(ep, timeout=12.0, trust_env=False, follow_redirects=True) + if r.status_code < 500: + return {"ok": True, "verified": False, + "detail": f"地址可连通（HTTP {r.status_code}），运行时再做协议级鉴权"} + return {"ok": False, "verified": False, "detail": f"服务返回 {r.status_code}，地址可能不可用"} + except Exception as e: + return {"ok": False, "verified": False, "detail": f"地址不可达：{_short_err(e)}"} + + +class VerifyAgentRequest(BaseModel): + platform: Optional[str] = None + endpoint: str + api_key: str + + +@router.post("/apps/verify-agent") +def verify_agent( + req: VerifyAgentRequest, + current_user: User = Depends(get_current_user), +): + """接入 Agent 前手动测试连通性 / 鉴权，供前台「测试连接」按钮调用。""" + result = verify_agent_connectivity(req.platform or "dify", req.endpoint, req.api_key) + return {"code": 0, "data": result, "message": "ok"} + + +@router.get("/ads") +def list_public_ads(db: Session = Depends(get_db)): + """ + 前台首页「精选推荐」广告位：返回已上架(active)广告，按排序升序，无需登录。 + 供 LandingPage 轮播调用，无鉴权依赖。 + """ + ads = ( + db.query(AdBanner) + .filter(AdBanner.status == "active") + .order_by(AdBanner.sort_order.asc()) + .all() + ) + return [ + { + "id": a.id, + "title": a.title, + "image_url": a.image_url, + "link_app_id": a.link_app_id, + "link_app_name": a.link_app_name, + "sort_order": a.sort_order, + "click_count": a.click_count, + "status": a.status, + "created_at": a.created_at.isoformat() if a.created_at else None, + } + for a in ads + ] + + +class UploadAppRequest(BaseModel): + name: str = Field(min_length=1, max_length=200) + description: Optional[str] = None + icon: Optional[str] = None + type: Optional[str] = None # skill, agent, mcp + source: Optional[str] = None + price: float = Field(default=0, ge=0) + download_url: Optional[str] = None + category: Optional[str] = None + tags: Optional[str] = None + is_member_free: Optional[bool] = False + join_promotion: Optional[bool] = False + agent_endpoint: Optional[str] = None + api_key: Optional[str] = None + platform: Optional[str] = None + + +class PurchaseAppRequest(BaseModel): + amount: Optional[float] = None + + +class ReviewAppRequest(BaseModel): + rating: int = Field(ge=1, le=5) + content: Optional[str] = None + + +@router.get("/apps") +def list_apps( + type: Optional[str] = Query(None), + source: Optional[str] = Query(None), + category: Optional[str] = Query(None), + search: Optional[str] = Query(None), + sort: str = Query(default="latest"), + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), + db: Session = Depends(get_db), + user_id: Optional[str] = Depends(get_current_user_id_optional), +): + query = db.query(MarketApp).filter(MarketApp.status == "published") + + if type: + query = query.filter(MarketApp.type == type) + if source: + query = query.filter(MarketApp.source == source) + if category: + query = query.filter(MarketApp.type == category) + if search: + query = query.filter( + and_( + MarketApp.name.ilike(f"%{search}%") + | MarketApp.description.ilike(f"%{search}%") + ) + ) + + if sort == "popular": + query = query.order_by(desc(MarketApp.install_count)) + elif sort == "rating": + query = query.order_by(desc(MarketApp.rating_avg)) + else: + query = query.order_by(desc(MarketApp.created_at)) + + total = query.count() + apps = query.offset((page - 1) * page_size).limit(page_size).all() + + # 如果用户已登录，获取该用户已安装的应用ID集合（确保按用户隔离） + installed_app_ids = set() + if user_id: + installations = db.query(AppInstallation.app_id).filter( + AppInstallation.user_id == user_id + ).all() + installed_app_ids = {inst[0] for inst in installations} + + # 查询已安装状态时始终带 user_id 过滤 + + result = [] + for app in apps: + # 用户自己创建/导入的应用应始终视为已安装（避免安装记录缺失导致显示未安装） + is_user_owned = bool(user_id) and app.source == "user" and app.publisher_id == user_id + item = { + "id": app.id, + "name": app.name, + "description": app.description, + "icon": app.icon, + "type": app.type, + "source": app.source, + "publisher_id": app.publisher_id, + "publisher_name": app.publisher_name, + "is_member_free": app.is_member_free if app.is_member_free else False, + "join_promotion": app.join_promotion if app.join_promotion else False, + "status": app.status, + "price": float(app.price) if app.price else 0, + "install_count": app.install_count or 0, + "rating_avg": float(app.rating_avg) if app.rating_avg else 0, + "review_count": app.review_count or 0, + "created_at": str(app.created_at) if app.created_at else None, + "installed": (app.id in installed_app_ids or is_user_owned) if user_id else False, + } + result.append(item) + + return { + "code": 0, + "data": { + "total": total, + "page": page, + "page_size": page_size, + "items": result, + }, + "message": "ok", + } + + +@router.get("/apps/{app_id}") +def get_app( + app_id: str, + db: Session = Depends(get_db), + current_user: Optional[User] = Depends(get_current_user), +): + app = db.query(MarketApp).filter(MarketApp.id == app_id).first() + if not app: + raise HTTPException(status_code=404, detail="应用不存在") + + # 检查当前用户是否已购买（按用户隔离） + purchased = False + installed = False + if current_user: + purchase = db.query(AppPurchase).filter( + and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id) + ).first() + if purchase: + purchased = True + installed = purchase.status == "installed" + # 通过 AppInstallation 表检查是否已安装（按用户隔离） + installation = db.query(AppInstallation).filter( + and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id) + ).first() + if installation: + installed = True + + return { + "code": 0, + "data": { + "id": app.id, "name": app.name, "description": app.description, + "icon": app.icon, "type": app.type, "source": app.source, + "type": app.type, + "price": float(app.price) if app.price else 0, + "install_count": app.install_count or 0, + "rating_avg": float(app.rating_avg) if app.rating_avg else 0, + "review_count": app.review_count or 0, + "publisher_id": app.publisher_id, + "download_url": app.download_url, + "detail": app.detail, + "created_at": str(app.created_at), + "purchased": purchased, + "installed": installed, + }, + "message": "ok", + } + + +@router.post("/apps") +def upload_app( + req: UploadAppRequest, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + # 接入 Agent 时先做连通性 / 鉴权校验，避免把无效连接存进库 + if (req.type or "") == "agent" and req.agent_endpoint and req.api_key: + v = verify_agent_connectivity(req.platform or "dify", req.agent_endpoint, req.api_key) + if not v["ok"]: + raise HTTPException( + status_code=400, + detail=f"Agent 连接校验未通过：{v['detail']}", + ) + + # source 判定：管理员（admin 角色）通过前台市场导入/开发的 Agent/Skill/MCP + # 统一标记为官方（official），其余用户按请求或默认为 user。 + if current_user.role == "admin": + app_source = "official" + elif req.source and req.source in ("official", "user"): + app_source = req.source + else: + app_source = "user" + + app = MarketApp( + id=gen_uuid(), + name=req.name, + description=req.description, + icon=req.icon, + type=req.type if req.type in ("skill","agent","mcp") else "skill", + source=app_source, + platform=req.platform, + price=req.price, + install_count=0, + rating_avg=0, + review_count=0, + publisher_id=current_user.id, + publisher_name=current_user.username, + is_member_free=req.is_member_free if req.is_member_free is not None else False, + join_promotion=req.join_promotion if req.join_promotion is not None else False, + agent_endpoint=req.agent_endpoint, + agent_api_key=req.api_key, + status="published", # 用户创建的应用直接发布 + created_at=datetime.now(), + updated_at=datetime.now(), + ) + db.add(app) + db.commit() + db.refresh(app) + + # MCP 应用同步注册到后台 MCP Client 管理（mcp_servers 表） + if app.type == "mcp": + code = req.download_url or "" + if code: + app.skill_content = code + parsed = parse_mcp_config(code) + transport = parsed["transport_type"] + app.mcp_transport = transport + db.commit() + db.refresh(app) + + existing_server = db.query(MCPServer).filter(MCPServer.id == app.id).first() + if not existing_server: + url = parsed["url"] or _extract_url_from_text(req.description or "") or "" + headers_json = json.dumps(parsed["headers"], ensure_ascii=False) if parsed["headers"] else "{}" + server = MCPServer( + id=app.id, + name=app.name, + description=app.description or "", + transport_type=transport, + command="", + args="{}", + env_vars="{}", + url=url, + headers=headers_json, + is_enabled=True, + status="disconnected", + tools_count=0, + resources_count=0, + created_at=app.created_at or datetime.now(), + updated_at=app.updated_at or datetime.now(), + ) + db.add(server) + db.commit() + + # 导入后自动做一次连通性测试，状态会同步为 connected / error + _test_mcp_server(server, db) + + # 用户创建/导入的应用自动标记为已安装 + existing_install = db.query(AppInstallation).filter( + AppInstallation.user_id == current_user.id, + AppInstallation.app_id == app.id, + ).first() + if not existing_install: + installation = AppInstallation( + user_id=current_user.id, + app_id=app.id, + installed_at=datetime.now(), + ) + db.add(installation) + db.commit() + + LogService.log_operation( + user_id=current_user.id, username=current_user.username, action="upload_app", + target="market_app", target_id=app.id, + detail=f"上传应用 {app.name}", + ) + + return { + "code": 0, + "data": { + "id": app.id, "name": app.name, "status": app.status, + "created_at": str(app.created_at), + }, + "message": "提交成功，等待审核", + } + + +@router.post("/apps/{app_id}/purchase") +def purchase_app( + app_id: str, + req: PurchaseAppRequest, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + app = db.query(MarketApp).filter( + and_(MarketApp.id == app_id, MarketApp.status == "published") + ).first() + if not app: + raise HTTPException(status_code=404, detail="应用不存在或未上架") + + # 检查是否已购买 + existing = db.query(AppPurchase).filter( + and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id) + ).first() + if existing and existing.status in ("purchased", "installed"): + raise HTTPException(status_code=400, detail="您已购买该应用") + + amount = req.amount or app.price + + # 创建订单 + order = Order( + id=gen_uuid(), + user_id=current_user.id, + order_type="app", + item_id=app_id, + item_name=app.name, + amount=amount, + status="pending", + created_at=datetime.now(), + ) + db.add(order) + db.flush() + + # 创建购买记录 + purchase = AppPurchase( + id=gen_uuid(), + user_id=current_user.id, + app_id=app_id, + order_id=order.id, + status="purchased", + purchased_at=datetime.now(), + ) + db.add(purchase) + + # 更新应用下载量 + app.install_count = (app.install_count or 0) + 1 + app.updated_at = datetime.now() + + db.commit() + + return { + "code": 0, + "data": { + "order_id": order.id, + "purchase_id": purchase.id, + "amount": amount, + "status": "purchased", + }, + "message": "购买成功", + } + + +@router.post("/apps/{app_id}/install") +def install_app( + app_id: str, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + app = db.query(MarketApp).filter(MarketApp.id == app_id).first() + if not app: + raise HTTPException(status_code=404, detail="应用不存在") + + # 检查是否已安装（通过 AppInstallation 表） + existing_install = db.query(AppInstallation).filter( + and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id) + ).first() + if existing_install: + raise HTTPException(status_code=400, detail="应用已安装") + + # 检查是否已购买（付费应用需要先购买） + purchase = db.query(AppPurchase).filter( + and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id) + ).first() + + # 对于付费应用，必须已购买才能安装 + if float(app.price or 0) > 0: + if not purchase: + raise HTTPException(status_code=403, detail="请先购买应用") + if purchase.status == "installed": + raise HTTPException(status_code=400, detail="应用已安装") + + # 创建安装记录 + installation = AppInstallation( + id=gen_uuid(), + user_id=current_user.id, + app_id=app_id, + installed_at=datetime.now(), + ) + db.add(installation) + + # 如果有购买记录，更新状态为已安装 + if purchase: + purchase.status = "installed" + purchase.installed_at = datetime.now() + + # 更新应用安装量 + app.install_count = (app.install_count or 0) + 1 + app.updated_at = datetime.now() + + db.commit() + + return { + "code": 0, + "data": {"installation_id": installation.id, "status": "installed"}, + "message": "安装成功", + } + + +@router.delete("/apps/{app_id}/uninstall") +def uninstall_app( + app_id: str, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """卸载应用 - 删除当前用户的安装记录（按用户隔离）""" + app = db.query(MarketApp).filter(MarketApp.id == app_id).first() + if not app: + raise HTTPException(status_code=404, detail="应用不存在") + + # 查找当前用户的安装记录（按用户隔离） + installation = db.query(AppInstallation).filter( + and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id) + ).first() + if not installation: + raise HTTPException(status_code=400, detail="应用未安装") + + # 删除安装记录 + db.delete(installation) + + # 问题4修复: 将AppPurchase状态从installed重置为purchased + purchase = db.query(AppPurchase).filter( + and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id) + ).first() + if purchase and purchase.status == "installed": + purchase.status = "purchased" + + # 更新应用安装量（减1，最小为0） + if app.install_count and app.install_count > 0: + app.install_count = app.install_count - 1 + app.updated_at = datetime.now() + + db.commit() + + LogService.log_operation( + user_id=current_user.id, username=current_user.username, action="uninstall_app", + target="market_app", target_id=app_id, + detail=f"卸载应用 {app.name}", + ) + + return { + "code": 0, + "message": "卸载成功", + } + + +@router.post("/apps/{app_id}/review") +def review_app( + app_id: str, + req: ReviewAppRequest, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + app = db.query(MarketApp).filter(MarketApp.id == app_id).first() + if not app: + raise HTTPException(status_code=404, detail="应用不存在") + + # 检查是否已评价 + existing = db.query(AppReview).filter( + and_(AppReview.app_id == app_id, AppReview.user_id == current_user.id) + ).first() + if existing: + # 更新已有评价 + old_rating = existing.rating + existing.rating = req.rating + existing.content = req.content + existing.updated_at = datetime.now() + + # 重新计算平均分 (总数不变，减去旧评分加新评分) + if app.review_count and app.review_count > 0: + total_rating = float(app.rating_avg or 0) * app.review_count + total_rating = total_rating - old_rating + req.rating + app.rating_avg = total_rating / app.review_count + else: + app.rating_avg = req.rating + app.review_count = 1 + else: + # 创建新评价 + review = AppReview( + id=gen_uuid(), + app_id=app_id, + user_id=current_user.id, + rating=req.rating, + content=req.content, + created_at=datetime.now(), + updated_at=datetime.now(), + ) + db.add(review) + + # 重新计算平均分 + app.review_count = (app.review_count or 0) + 1 + if app.review_count == 1: + app.rating_avg = req.rating + else: + total_rating = float(app.rating_avg or 0) * (app.review_count - 1) + req.rating + app.rating_avg = round(total_rating / app.review_count, 1) + + app.updated_at = datetime.now() + db.commit() + + return { + "code": 0, + "data": { + "rating_avg": float(app.rating_avg or 0), + "review_count": app.review_count or 0, + }, + "message": "评价成功", + } + + +@router.get("/my-apps") +def get_my_apps( + status: Optional[str] = Query(None), + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + query = db.query(AppPurchase).filter(AppPurchase.user_id == current_user.id) + purchases = query.order_by(desc(AppPurchase.created_at)).all() + + result = [] + for p in purchases: + app = db.query(MarketApp).filter(MarketApp.id == p.app_id).first() + if app: + result.append({ + "id": p.id, + "app_id": app.id, + "name": app.name, + "icon": app.icon, + "type": app.type, + "price": float(app.price or 0), + "purchased_at": str(p.created_at) if p.created_at else None, + }) + + return {"code": 0, "data": result, "message": "ok"} + + +@router.get("/reviews/{app_id}") +def get_reviews( + app_id: str, + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), + db: Session = Depends(get_db), +): + query = db.query(AppReview).filter(AppReview.app_id == app_id) + total = query.count() + + reviews = ( + query.order_by(desc(AppReview.created_at)) + .offset((page - 1) * page_size) + .limit(page_size) + .all() + ) + + result = [] + for r in reviews: + user = db.query(User).filter(User.id == r.user_id).first() + result.append({ + "id": r.id, + "user_id": r.user_id, + "username": user.username if user else "unknown", + "avatar": user.avatar if user else None, + "rating": r.rating, + "content": r.content, + "created_at": str(r.created_at) if r.created_at else None, + }) + + return { + "code": 0, + "data": {"total": total, "page": page, "page_size": page_size, "items": result}, + "message": "ok", + } + + +# ============================================================ +# Agent部署资源查询接口 +# ============================================================ + +@router.get("/agent-resources") +def get_agent_resources( + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Agent部署时需要的资源列表：可用的应用、知识库、文档""" + # 用户已发布的应用 + my_apps = db.query(MarketApp).filter( + MarketApp.publisher_id == current_user.id, + MarketApp.status == "published", + ).order_by(desc(MarketApp.updated_at)).all() + + apps_result = [] + for app in my_apps: + apps_result.append({ + "id": app.id, + "name": app.name, + "type": app.type, + "description": app.description, + "install_count": app.install_count or 0, + }) + + # 用户知识库 + from ..models import KnowledgeBase, KnowledgeDocument + kb_list = db.query(KnowledgeBase).filter( + KnowledgeBase.user_id == current_user.id + ).order_by(desc(KnowledgeBase.updated_at)).all() + + kb_result = [] + for kb in kb_list: + doc_count = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb.id + ).count() + kb_result.append({ + "id": kb.id, + "name": kb.name, + "description": kb.description, + "document_count": doc_count, + }) + + # 用户已安装的应用（从AppPurchase） + installed = db.query(AppPurchase).filter( + AppPurchase.user_id == current_user.id + ).all() + installed_ids = [p.app_id for p in installed] + installed_apps = [] + if installed_ids: + iapps = db.query(MarketApp).filter(MarketApp.id.in_(installed_ids)).all() + for ia in iapps: + installed_apps.append({ + "id": ia.id, + "name": ia.name, + "type": ia.type, + "description": ia.description, + }) + + return { + "code": 0, + "data": { + "my_apps": apps_result, + "knowledge_bases": kb_result, + "installed_apps": installed_apps, + }, + "message": "ok", + } + diff --git a/frontend/src/pages/front/MarketPage.tsx b/frontend/src/pages/front/MarketPage.tsx index e48e754..385292f 100644 --- a/frontend/src/pages/front/MarketPage.tsx +++ b/frontend/src/pages/front/MarketPage.tsx @@ -541,7 +541,10 @@ const CreateToolModal: React.FC = ({ appType, onClose, onS publisherName: 'demo_user', status: 'published' as const, icon: appType === 'skill' ? 'S' : appType === 'agent' ? 'A' : 'M', - }); + // 关键修复：把粘贴的 MCP 传输代码随提交一起发给后端，否则后端拿不到代码、URL 为空、永远连不上 + mcpCode: appType === 'mcp' ? mcpCode.trim() : undefined, + mcpTransport: appType === 'mcp' ? mcpTransport : undefined, + } as any); }, 1500); }, 2000); }; @@ -721,13 +724,17 @@ const CreateToolModal: React.FC = ({ appType, onClose, onS