diff --git a/backend/python/app/api/admin/mcp_admin.py.bak_1784203135 b/backend/python/app/api/admin/mcp_admin.py.bak_1784203135 deleted file mode 100644 index 8de2442..0000000 --- a/backend/python/app/api/admin/mcp_admin.py.bak_1784203135 +++ /dev/null @@ -1,886 +0,0 @@ -""" - -MCP (Model Context Protocol) Server Management API - -Phase 3.3: Manage MCP servers, discover tools/resources, test connections - -""" - -from fastapi import APIRouter, Depends, HTTPException, Query, Body - -from sqlalchemy import or_, func, desc, asc, text - -from sqlalchemy.orm import Session - -from pydantic import BaseModel, Field - -from typing import Optional, List, Dict, Any - -from datetime import datetime - -import uuid - -import json - -import logging - -import re - -from ...database import get_db - -from ...models import gen_uuid - -from ...auth import get_current_admin - -from ...middleware.log_service import LogService - -router = APIRouter() - -logger = logging.getLogger("aiflow") - -# ============================================================ - -# Pydantic Models - -# ============================================================ - -class MCPServerCreate(BaseModel): - - name: str - - description: Optional[str] = "" - - transport_type: str = "stdio" - - command: Optional[str] = "" - - args: Optional[Any] = None - - env_vars: Optional[Dict[str, str]] = None - - url: Optional[str] = "" - - headers: Optional[Dict[str, str]] = None - - is_enabled: Optional[bool] = True - -class MCPServerUpdate(BaseModel): - - name: Optional[str] = None - - description: Optional[str] = None - - transport_type: Optional[str] = None - - command: Optional[str] = None - - args: Optional[Any] = None - - env_vars: Optional[Dict[str, str]] = None - - url: Optional[str] = None - - headers: Optional[Dict[str, str]] = None - - is_enabled: Optional[bool] = None - -class MCPServerConnectionTest(BaseModel): - - transport_type: Optional[str] = None - - command: Optional[str] = None - - args: Optional[Any] = None - - env_vars: Optional[Dict[str, str]] = None - - url: Optional[str] = None - - headers: Optional[Dict[str, str]] = None - -# ============================================================ - -# Helper Functions - -# ============================================================ - -def _format_cst_time(dt): - - """Format datetime to CST string""" - - if dt is None: - - return None - - try: - - from datetime import timezone, timedelta - - CST = timezone(timedelta(hours=8)) - - if dt.tzinfo is None: - - dt = dt.replace(tzinfo=CST) - - return dt.astimezone(CST).strftime("%Y-%m-%d %H:%M:%S") - - except Exception: - - return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None - -def server_to_dict(server) -> dict: - - """Convert MCPServer model to dict""" - - return { - - "id": server.id, - - "name": server.name, - - "description": server.description or "", - - "transport_type": server.transport_type or "stdio", - - "transport_type_label": {"stdio": "STDIO", "sse": "SSE"}.get(server.transport_type, server.transport_type), - - "command": server.command or "", - - "args": json.loads(server.args) if server.args else {}, - - "env_vars": json.loads(server.env_vars) if server.env_vars else {}, - - "url": server.url or "", - - "headers": json.loads(server.headers) if server.headers else {}, - - "is_enabled": server.is_enabled if server.is_enabled is not None else True, - - "status": server.status or "disconnected", - - "last_connected_at": _format_cst_time(server.last_connected_at), - - "tools_count": server.tools_count or 0, - - "resources_count": server.resources_count or 0, - - "created_at": _format_cst_time(server.created_at), - - "updated_at": _format_cst_time(server.updated_at), - - } - -def _extract_url_from_text(text: str) -> Optional[str]: - - """Extract first http/https URL from text.""" - - if not text: - - return None - - match = re.search(r"https?://[^\s\"'<>\)\]\}]+", text) - - return match.group(0) if match else None - -def _test_mcp_server(server, db) -> dict: - """ - Test connectivity to an MCP server and persist the result. - Supports stdio, sse and streamable_http transports. - Returns a result dict with success/message/details. - """ - result = {"success": False, "message": "", "details": {}} - transport = (server.transport_type or "stdio").replace("-", "_") - - try: - if transport in ("streamable_http", "streamablehttp"): - url = server.url - if not url: - result["message"] = "Streamable HTTP endpoint URL is required" - else: - import httpx - try: - # A real streamable-http MCP endpoint accepts POST JSON-RPC. - # Some implementations also accept GET for SSE events. - # Try GET first (cheapest), fall back to a minimal POST. - resp = httpx.get(url, timeout=10.0, follow_redirects=True) - if resp.status_code == 405: - resp = httpx.post( - url, - json={"jsonrpc": "2.0", "method": "initialize", "id": 1}, - timeout=10.0, - follow_redirects=True, - ) - # Any HTTP response (including 4xx) means the host is reachable. - result["success"] = resp.status_code < 500 - result["message"] = f"HTTP responded with {resp.status_code}" - result["details"]["status_code"] = resp.status_code - result["details"]["content_type"] = resp.headers.get("content-type", "") - except httpx.ConnectError as e: - result["message"] = f"Connection refused - server may be down: {str(e)}" - except httpx.TimeoutException: - result["message"] = "Connection timed out" - except Exception as e: - result["message"] = f"HTTP request failed: {str(e)}" - - elif transport == "sse": - sse_url = server.url - if not sse_url: - result["message"] = "SSE URL is required" - else: - import httpx - try: - response = httpx.get(sse_url, timeout=10.0, follow_redirects=True) - result["success"] = response.status_code < 500 - result["message"] = f"SSE endpoint responded with {response.status_code}" - result["details"]["status_code"] = response.status_code - result["details"]["content_type"] = response.headers.get("content-type", "") - except httpx.ConnectError as e: - result["message"] = f"Connection refused - server may be down: {str(e)}" - except httpx.TimeoutException: - result["message"] = "Connection timed out" - except Exception as e: - result["message"] = f"SSE request failed: {str(e)}" - - elif transport == "stdio": - cmd = server.command - if not cmd: - result["message"] = "STDIO command is required" - else: - import subprocess - try: - subprocess.run([cmd, "--version"], capture_output=True, timeout=10) - result["success"] = True - result["message"] = "Command found and executable" - except FileNotFoundError: - result["message"] = "Command not found in PATH" - except subprocess.TimeoutExpired: - result["message"] = "Command timed out" - except Exception as e: - result["message"] = "Command test failed" - result["details"]["error"] = str(e) - - else: - result["message"] = f"Unsupported transport type: {transport}" - - # Persist status - if result["success"]: - server.status = "connected" - server.last_connected_at = datetime.now() - else: - server.status = "error" - server.updated_at = datetime.now() - db.commit() - - except Exception as e: - result["message"] = str(e) - result["details"]["error"] = str(e) - server.status = "error" - server.updated_at = datetime.now() - db.commit() - - return result -def _sync_market_mcp_to_servers(db: Session): - - """ - - Synchronize MarketApp MCP records (type='mcp') into mcp_servers table. - - Uses MarketApp.id as MCPServer.id so detail/edit endpoints can work directly. - - """ - - from ...models import MarketApp, MCPServer - - market_mcps = db.query(MarketApp).filter( - - MarketApp.type == "mcp", - - MarketApp.status == "published", - - ).all() - - existing_ids = {row[0] for row in db.query(MCPServer.id).all()} - - for ma in market_mcps: - - if ma.id in existing_ids: - # Update URL if the server record has no endpoint yet - existing = db.query(MCPServer).filter(MCPServer.id == ma.id).first() - if existing and not existing.url: - url = _extract_url_from_text(ma.skill_content or "") or _extract_url_from_text(ma.description or "") or "" - if not url and ma.agent_endpoint: - url = ma.agent_endpoint - if url: - existing.url = url - existing.updated_at = datetime.now() - db.commit() - continue - - transport = ma.mcp_transport or "streamable_http" - - code = ma.skill_content or "" - - url = _extract_url_from_text(code) or _extract_url_from_text(ma.description or "") or "" - - command = "" - - if not url and ma.agent_endpoint: - - url = ma.agent_endpoint - - now = datetime.now() - - server = MCPServer( - - id=ma.id, - - name=ma.name, - - description=ma.description or "", - - transport_type=transport, - - command=command, - - args="{}", - - env_vars="{}", - - url=url, - - headers="{}", - - is_enabled=True, - - status="disconnected", - - tools_count=0, - - resources_count=0, - - created_at=ma.created_at or now, - - updated_at=ma.updated_at or now, - - ) - - db.add(server) - - if market_mcps: - - db.commit() - -# ============================================================ - -# API Endpoints - -# ============================================================ - -@router.get("/stats/overview", summary="Get MCP server statistics") - -async def get_mcp_stats( - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """Get overview statistics for MCP servers""" - - from ...models import MCPServer - - _sync_market_mcp_to_servers(db) - - total = db.query(func.count(MCPServer.id)).scalar() or 0 - - enabled = db.query(func.count(MCPServer.id)).filter(MCPServer.is_enabled == True).scalar() or 0 - - connected = db.query(func.count(MCPServer.id)).filter(MCPServer.status == "connected").scalar() or 0 - - # By transport type - - by_transport = {} - - rows = db.query(MCPServer.transport_type, func.count(MCPServer.id)).group_by(MCPServer.transport_type).all() - - for t, c in rows: - - by_transport[t or "unknown"] = c - - # By status - - by_status = {} - - rows = db.query(MCPServer.status, func.count(MCPServer.id)).group_by(MCPServer.status).all() - - for s, c in rows: - - by_status[s or "unknown"] = c - - return { - - "code": 0, - - "data": { - - "total_servers": total, - - "enabled_servers": enabled, - - "connected_servers": connected, - - "by_transport": by_transport, - - "by_status": by_status, - - }, - - "message": "ok", - - } - -@router.get("/", summary="List MCP servers") - -async def list_mcp_servers( - - page: int = Query(1, ge=1), - - page_size: int = Query(20, ge=1, le=100), - - search: Optional[str] = Query(None), - - transport_type: Optional[str] = Query(None), - - status: Optional[str] = Query(None), - - is_enabled: Optional[bool] = Query(None), - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """List all MCP servers with pagination and filters""" - - from ...models import MCPServer - - _sync_market_mcp_to_servers(db) - - query = db.query(MCPServer) - - if search: - - query = query.filter( - - or_(MCPServer.name.contains(search), MCPServer.description.contains(search)) - - ) - - if transport_type: - - query = query.filter(MCPServer.transport_type == transport_type) - - if status: - - query = query.filter(MCPServer.status == status) - - if is_enabled is not None: - - query = query.filter(MCPServer.is_enabled == is_enabled) - - total = query.count() - - servers = query.order_by(desc(MCPServer.created_at)).offset((page - 1) * page_size).limit(page_size).all() - - return { - - "code": 0, - - "data": { - - "total": total, - - "page": page, - - "page_size": page_size, - - "items": [server_to_dict(s) for s in servers], - - }, - - "message": "ok", - - } - -@router.get("/{server_id}", summary="Get MCP server details") - -async def get_mcp_server( - - server_id: str, - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """Get a single MCP server by ID""" - - from ...models import MCPServer - - _sync_market_mcp_to_servers(db) - - server = db.query(MCPServer).filter(MCPServer.id == server_id).first() - - if not server: - - raise HTTPException(status_code=404, detail="MCP server not found") - - return {"code": 0, "data": server_to_dict(server), "message": "ok"} - -@router.post("/", summary="Create MCP server") - -async def create_mcp_server( - - data: MCPServerCreate, - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """Register a new MCP server""" - - from ...models import MCPServer - - # Check name uniqueness - - existing = db.query(MCPServer).filter(MCPServer.name == data.name).first() - - if existing: - - raise HTTPException(status_code=400, detail="MCP server with this name already exists") - - now = datetime.now() - - server = MCPServer( - - id=gen_uuid(), - - name=data.name, - - description=data.description or "", - - transport_type=data.transport_type, - - command=data.command or "", - - args=json.dumps(data.args) if data.args else "{}", - - env_vars=json.dumps(data.env_vars) if data.env_vars else "{}", - - url=data.url or "", - - headers=json.dumps(data.headers) if data.headers else "{}", - - is_enabled=data.is_enabled if data.is_enabled is not None else True, - - status="disconnected", - - created_at=now, - - updated_at=now, - - ) - - db.add(server) - - db.commit() - - db.refresh(server) - - logger.info(f"MCP server created: {server.name} ({server.id})") - - return {"code": 0, "data": server_to_dict(server), "message": "MCP server created"} - -@router.put("/{server_id}", summary="Update MCP server") - -async def update_mcp_server( - - server_id: str, - - data: MCPServerUpdate, - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """Update an existing MCP server""" - - from ...models import MCPServer - - server = db.query(MCPServer).filter(MCPServer.id == server_id).first() - - if not server: - - raise HTTPException(status_code=404, detail="MCP server not found") - - update_fields = data.model_dump(exclude_unset=True) - - # Handle JSON fields - - for json_field in ["args", "env_vars", "headers"]: - - if json_field in update_fields and update_fields[json_field] is not None: - - update_fields[json_field] = json.dumps(update_fields[json_field]) - - for key, value in update_fields.items(): - - if value is not None: - - setattr(server, key, value) - - server.updated_at = datetime.now() - - db.commit() - - db.refresh(server) - - logger.info(f"MCP server updated: {server.name} ({server.id})") - - return {"code": 0, "data": server_to_dict(server), "message": "MCP server updated"} - -@router.delete("/{server_id}", summary="Delete MCP server") - -async def delete_mcp_server( - - server_id: str, - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """Delete an MCP server""" - - from ...models import MCPServer - - server = db.query(MCPServer).filter(MCPServer.id == server_id).first() - - if not server: - - raise HTTPException(status_code=404, detail="MCP server not found") - - name = server.name - - db.delete(server) - - db.commit() - - logger.info(f"MCP server deleted: {name} ({server_id})") - - return {"code": 0, "data": None, "message": "MCP server deleted"} - -@router.put("/{server_id}/toggle", summary="Enable/Disable MCP server") - -async def toggle_mcp_server( - - server_id: str, - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """Toggle MCP server enabled state""" - - from ...models import MCPServer - - server = db.query(MCPServer).filter(MCPServer.id == server_id).first() - - if not server: - - raise HTTPException(status_code=404, detail="MCP server not found") - - server.is_enabled = not server.is_enabled - - server.updated_at = datetime.now() - - db.commit() - - logger.info(f"MCP server toggled: {server.name} -> enabled={server.is_enabled}") - - return { - - "code": 0, - - "data": {"id": server.id, "is_enabled": server.is_enabled}, - - "message": "MCP server toggled", - - } - -@router.post("/{server_id}/test", summary="Test MCP server connection") -async def test_mcp_connection( - server_id: str, - db: Session = Depends(get_db), - admin_user=Depends(get_current_admin), - test_data: Optional[MCPServerConnectionTest] = None, -): - """Test connection to an MCP server""" - from ...models import MCPServer - server = db.query(MCPServer).filter(MCPServer.id == server_id).first() - if not server: - raise HTTPException(status_code=404, detail="MCP server not found") - - # If test_data provides an override URL/transport, apply it temporarily - original_url = server.url - original_transport = server.transport_type - if test_data: - if test_data.transport_type: - server.transport_type = test_data.transport_type - if test_data.url: - server.url = test_data.url - if test_data.command: - server.command = test_data.command - - result = _test_mcp_server(server, db) - - # Restore original values if no override was explicitly persisted - if test_data and test_data.url is None: - server.url = original_url - if test_data and test_data.transport_type is None: - server.transport_type = original_transport - if test_data: - db.commit() - - return {"code": 0, "data": result, "message": "ok"} - -@router.get("/{server_id}/tools", summary="List MCP server tools") - -async def list_mcp_tools( - - server_id: str, - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """List tools exposed by an MCP server (placeholder/mock)""" - - from ...models import MCPServer - - server = db.query(MCPServer).filter(MCPServer.id == server_id).first() - - if not server: - - raise HTTPException(status_code=404, detail="MCP server not found") - - # In a real implementation, this would connect to the MCP server and call tools/list - - # For now, return mock data based on common MCP tools - - mock_tools = [] - - if server.transport_type == "stdio": - - mock_tools = [ - - {"name": "read_file", "description": "Read contents of a file"}, - - {"name": "write_file", "description": "Write contents to a file"}, - - {"name": "execute_command", "description": "Execute a shell command"}, - - {"name": "search_content", "description": "Search for content in files"}, - - ] - - elif server.transport_type == "sse": - - mock_tools = [ - - {"name": "web_search", "description": "Search the web"}, - - {"name": "fetch_url", "description": "Fetch content from URL"}, - - {"name": "api_request", "description": "Make API requests"}, - - ] - - return { - - "code": 0, - - "data": { - - "server_name": server.name, - - "tools": mock_tools, - - "count": len(mock_tools), - - }, - - "message": "ok", - - } - -@router.get("/{server_id}/resources", summary="List MCP server resources") - -async def list_mcp_resources( - - server_id: str, - - db: Session = Depends(get_db), - - admin_user=Depends(get_current_admin), - -): - - """List resources exposed by an MCP server (placeholder/mock)""" - - from ...models import MCPServer - - server = db.query(MCPServer).filter(MCPServer.id == server_id).first() - - if not server: - - raise HTTPException(status_code=404, detail="MCP server not found") - - mock_resources = [ - - {"uri": f"{server.name}://docs/readme", "name": "Documentation", "mime_type": "text/markdown"}, - - {"uri": f"{server.name}://config/settings", "name": "Configuration", "mime_type": "application/json"}, - - ] - - return { - - "code": 0, - - "data": { - - "server_name": server.name, - - "resources": mock_resources, - - "count": len(mock_resources), - - }, - - "message": "ok", - - } -