qyyfw/qihuaiflow
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1633d5c33e668aaad52d8fb4714c6ffecc4790fa
qihuaiflow/backend/python/app/api/admin/mcp_admin.py
T

3603 lines
27 KiB
Python
Raw Blame History

"""
MCP (Model Context Protocol) Server Management API
Phase 3.3: Manage MCP servers, discover tools/resources, test connections
"""
import asyncio
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:
try:
# Use the real MCP SDK to verify connectivity (same as chat flow).
# A naive HTTP GET/POST can give false negatives for endpoints
# that only accept proper MCP JSON-RPC over POST (e.g., AnySearch).
from ...mcp_client import list_mcp_tools as _lmt
hdrs = json.loads(server.headers) if isinstance(server.headers, str) else (server.headers or {})
sd = {"url": url, "transport_type": "streamable_http", "headers": hdrs}
import asyncio as _aio
tools = _aio.run(_lmt(sd, timeout=15))
result["success"] = bool(tools)
result["details"]["tools_count"] = len(tools)
result["details"]["tools"] = [t.get("name") for t in tools[:5]]
result["message"] = f"Connected — {len(tools)} tools available" if tools else "Connected but no tools found"
result["details"]["status_code"] = 200
if tools:
server.tools_count = len(tools)
server.status = "connected"
except Exception as e:
result["message"] = f"Connection 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 = await asyncio.get_running_loop().run_in_executor(None, _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",
}
Reference in new issue View Git Blame Copy Permalink