██████╗ ██╗ ██╗ ██████╗██╗ █████╗ ██████╗██╗ ██████╗ ██╗ ██╗██████╗
██╔══██╗██║ ██║██╔════╝██║ ██╔══██╗ ██╔════╝██║ ██╔═══██╗██║ ██║██╔══██╗
██║ ██║██║ ██║██║ ██║ ███████║ ██║ ██║ ██║ ██║██║ ██║██║ ██║
██║ ██║██║ ██║██║ ██║ ██╔══██║ ██║ ██║ ██║ ██║██║ ██║██║ ██║
██████╔╝╚██████╔╝╚██████╗███████╗██║ ██║ ╚██████╗███████╗╚██████╔╝╚██████╔╝██████╔╝
╚═════╝ ╚═════╝ ╚═════╝╚══════╝╚═╝ ╚═╝ ╚═════╝╚══════╝ ╚═════╝ ╚═════╝ ╚═════╝
MCP servers có thể trở thành công cụ debugging mạnh mẽ cho các vấn đề hệ thống, database và container. Bằng cách tạo các custom servers chuyên biệt, bạn có thể tự động hóa việc thu thập thông tin, phân tích logs và chẩn đoán lỗi.
#!/usr/bin/env python3
"""
System Debug MCP Server
Provides comprehensive OS debugging capabilities
"""
import asyncio
import json
import psutil
import subprocess
import os
import time
from pathlib import Path
from typing import Dict, List, Any
from mcp import Server, types
from mcp.server.models import InitializationOptions
class SystemDebugServer:
def __init__(self):
self.server = Server("system-debug")
self.setup_handlers()
def setup_handlers(self):
@self.server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
return [
types.Tool(
name="system_health_check",
description="Comprehensive system health analysis",
inputSchema={
"type": "object",
"properties": {
"check_type": {
"type": "string",
"enum": ["full", "cpu", "memory", "disk", "network", "processes"],
"description": "Type of health check to perform"
}
},
"required": ["check_type"]
}
),
types.Tool(
name="process_analyzer",
description="Analyze running processes and resource usage",
inputSchema={
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["list_high_cpu", "list_high_memory", "find_process", "kill_process"],
"description": "Process analysis action"
},
"threshold": {
"type": "number",
"description": "Threshold for CPU/Memory percentage (default: 80)"
},
"process_name": {
"type": "string",
"description": "Process name to find or kill"
},
"pid": {
"type": "integer",
"description": "Process ID to kill"
}
},
"required": ["action"]
}
),
types.Tool(
name="log_analyzer",
description="Analyze system logs for errors and patterns",
inputSchema={
"type": "object",
"properties": {
"log_type": {
"type": "string",
"enum": ["syslog", "dmesg", "auth", "custom"],
"description": "Type of log to analyze"
},
"log_path": {
"type": "string",
"description": "Custom log file path"
},
"search_pattern": {
"type": "string",
"description": "Pattern to search for in logs"
},
"lines": {
"type": "integer",
"description": "Number of recent lines to analyze (default: 100)"
}
},
"required": ["log_type"]
}
),
types.Tool(
name="network_diagnostics",
description="Network connectivity and performance diagnostics",
inputSchema={
"type": "object",
"properties": {
"test_type": {
"type": "string",
"enum": ["connectivity", "ports", "bandwidth", "dns", "routing"],
"description": "Type of network test"
},
"target": {
"type": "string",
"description": "Target host/IP for testing"
},
"port": {
"type": "integer",
"description": "Port number for port testing"
}
},
"required": ["test_type"]
}
)
]
@self.server.call_tool()
async def handle_call_tool(name: str, arguments: Dict[str, Any] | None) -> List[types.TextContent]:
if arguments is None:
arguments = {}
try:
if name == "system_health_check":
return await self._system_health_check(arguments)
elif name == "process_analyzer":
return await self._process_analyzer(arguments)
elif name == "log_analyzer":
return await self._log_analyzer(arguments)
elif name == "network_diagnostics":
return await self._network_diagnostics(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
return [types.TextContent(
type="text",
text=f"Error in {name}: {str(e)}"
)]
async def _system_health_check(self, args: Dict[str, Any]) -> List[types.TextContent]:
check_type = args.get("check_type", "full")
results = {}
if check_type in ["full", "cpu"]:
cpu_info = {
"usage_percent": psutil.cpu_percent(interval=1),
"count": psutil.cpu_count(),
"load_average": os.getloadavg() if hasattr(os, 'getloadavg') else None,
"frequency": psutil.cpu_freq()._asdict() if psutil.cpu_freq() else None
}
results["cpu"] = cpu_info
if check_type in ["full", "memory"]:
memory = psutil.virtual_memory()
swap = psutil.swap_memory()
memory_info = {
"virtual": {
"total": self._bytes_to_gb(memory.total),
"available": self._bytes_to_gb(memory.available),
"used": self._bytes_to_gb(memory.used),
"percent": memory.percent
},
"swap": {
"total": self._bytes_to_gb(swap.total),
"used": self._bytes_to_gb(swap.used),
"percent": swap.percent
}
}
results["memory"] = memory_info
if check_type in ["full", "disk"]:
disk_info = {}
for partition in psutil.disk_partitions():
try:
usage = psutil.disk_usage(partition.mountpoint)
disk_info[partition.mountpoint] = {
"total": self._bytes_to_gb(usage.total),
"used": self._bytes_to_gb(usage.used),
"free": self._bytes_to_gb(usage.free),
"percent": (usage.used / usage.total) * 100
}
except PermissionError:
continue
results["disk"] = disk_info
if check_type in ["full", "network"]:
network_info = {
"interfaces": {},
"connections": len(psutil.net_connections())
}
for interface, stats in psutil.net_io_counters(pernic=True).items():
network_info["interfaces"][interface] = {
"bytes_sent": self._bytes_to_mb(stats.bytes_sent),
"bytes_recv": self._bytes_to_mb(stats.bytes_recv),
"packets_sent": stats.packets_sent,
"packets_recv": stats.packets_recv
}
results["network"] = network_info
if check_type in ["full", "processes"]:
process_count = len(psutil.pids())
results["processes"] = {"total_count": process_count}
# Health assessment
health_status = self._assess_health(results)
results["health_assessment"] = health_status
return [types.TextContent(
type="text",
text=f"System Health Check ({check_type}):\n{json.dumps(results, indent=2)}"
)]
async def _process_analyzer(self, args: Dict[str, Any]) -> List[types.TextContent]:
action = args.get("action")
threshold = args.get("threshold", 80)
process_name = args.get("process_name")
pid = args.get("pid")
if action == "list_high_cpu":
high_cpu_processes = []
for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']):
try:
if proc.info['cpu_percent'] > threshold:
high_cpu_processes.append(proc.info)
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
return [types.TextContent(
type="text",
text=f"High CPU Processes (>{threshold}%):\n{json.dumps(high_cpu_processes, indent=2)}"
)]
elif action == "list_high_memory":
high_mem_processes = []
for proc in psutil.process_iter(['pid', 'name', 'memory_percent']):
try:
if proc.info['memory_percent'] > threshold:
high_mem_processes.append(proc.info)
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
return [types.TextContent(
type="text",
text=f"High Memory Processes (>{threshold}%):\n{json.dumps(high_mem_processes, indent=2)}"
)]
elif action == "find_process":
if not process_name:
return [types.TextContent(
type="text",
text="Error: process_name is required for find_process action"
)]
matching_processes = []
for proc in psutil.process_iter(['pid', 'name', 'cmdline', 'status']):
try:
if process_name.lower() in proc.info['name'].lower():
matching_processes.append(proc.info)
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
return [types.TextContent(
type="text",
text=f"Processes matching '{process_name}':\n{json.dumps(matching_processes, indent=2)}"
)]
elif action == "kill_process":
if not pid:
return [types.TextContent(
type="text",
text="Error: pid is required for kill_process action"
)]
try:
proc = psutil.Process(pid)
proc_info = proc.as_dict(['pid', 'name', 'status'])
proc.terminate()
return [types.TextContent(
type="text",
text=f"Process terminated: {json.dumps(proc_info, indent=2)}"
)]
except psutil.NoSuchProcess:
return [types.TextContent(
type="text",
text=f"Error: Process with PID {pid} not found"
)]
except psutil.AccessDenied:
return [types.TextContent(
type="text",
text=f"Error: Permission denied to kill process {pid}"
)]
def _bytes_to_gb(self, bytes_value):
return round(bytes_value / (1024**3), 2)
def _bytes_to_mb(self, bytes_value):
return round(bytes_value / (1024**2), 2)
def _assess_health(self, results):
issues = []
warnings = []
# CPU health
if "cpu" in results:
cpu_usage = results["cpu"]["usage_percent"]
if cpu_usage > 90:
issues.append(f"Critical CPU usage: {cpu_usage}%")
elif cpu_usage > 80:
warnings.append(f"High CPU usage: {cpu_usage}%")
# Memory health
if "memory" in results:
mem_percent = results["memory"]["virtual"]["percent"]
if mem_percent > 95:
issues.append(f"Critical memory usage: {mem_percent}%")
elif mem_percent > 85:
warnings.append(f"High memory usage: {mem_percent}%")
# Disk health
if "disk" in results:
for mount, info in results["disk"].items():
if info["percent"] > 95:
issues.append(f"Critical disk usage on {mount}: {info['percent']:.1f}%")
elif info["percent"] > 85:
warnings.append(f"High disk usage on {mount}: {info['percent']:.1f}%")
return {
"status": "critical" if issues else "warning" if warnings else "healthy",
"issues": issues,
"warnings": warnings
}
async def run(self):
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="system-debug",
server_version="1.0.0"
)
)
async def main():
server = SystemDebugServer()
await server.run()
if __name__ == "__main__":
asyncio.run(main())#!/usr/bin/env python3
"""
Database Debug MCP Server
Provides database monitoring and troubleshooting capabilities
"""
import asyncio
import json
import sqlite3
import subprocess
import time
from typing import Dict, List, Any, Optional
from mcp import Server, types
from mcp.server.models import InitializationOptions
class DatabaseDebugServer:
def __init__(self):
self.server = Server("database-debug")
self.setup_handlers()
def setup_handlers(self):
@self.server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
return [
types.Tool(
name="db_health_check",
description="Check database health and performance",
inputSchema={
"type": "object",
"properties": {
"db_type": {
"type": "string",
"enum": ["mysql", "postgresql", "sqlite", "mongodb"],
"description": "Database type"
},
"connection_string": {
"type": "string",
"description": "Database connection string"
},
"db_path": {
"type": "string",
"description": "SQLite database file path"
}
},
"required": ["db_type"]
}
),
types.Tool(
name="query_performance",
description="Analyze slow queries and performance issues",
inputSchema={
"type": "object",
"properties": {
"db_type": {
"type": "string",
"enum": ["mysql", "postgresql", "sqlite"],
"description": "Database type"
},
"query": {
"type": "string",
"description": "SQL query to analyze"
},
"connection_string": {
"type": "string",
"description": "Database connection string"
}
},
"required": ["db_type", "query"]
}
),
types.Tool(
name="db_connection_test",
description="Test database connectivity and response time",
inputSchema={
"type": "object",
"properties": {
"db_type": {
"type": "string",
"enum": ["mysql", "postgresql", "sqlite", "mongodb", "redis"],
"description": "Database type"
},
"host": {
"type": "string",
"description": "Database host"
},
"port": {
"type": "integer",
"description": "Database port"
},
"database": {
"type": "string",
"description": "Database name"
},
"username": {
"type": "string",
"description": "Username"
}
},
"required": ["db_type"]
}
),
types.Tool(
name="db_log_analyzer",
description="Analyze database logs for errors and patterns",
inputSchema={
"type": "object",
"properties": {
"db_type": {
"type": "string",
"enum": ["mysql", "postgresql", "mongodb"],
"description": "Database type"
},
"log_path": {
"type": "string",
"description": "Database log file path"
},
"error_pattern": {
"type": "string",
"description": "Error pattern to search for"
},
"lines": {
"type": "integer",
"description": "Number of recent lines to analyze"
}
},
"required": ["db_type"]
}
)
]
@self.server.call_tool()
async def handle_call_tool(name: str, arguments: Dict[str, Any] | None) -> List[types.TextContent]:
if arguments is None:
arguments = {}
try:
if name == "db_health_check":
return await self._db_health_check(arguments)
elif name == "query_performance":
return await self._query_performance(arguments)
elif name == "db_connection_test":
return await self._db_connection_test(arguments)
elif name == "db_log_analyzer":
return await self._db_log_analyzer(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
return [types.TextContent(
type="text",
text=f"Error in {name}: {str(e)}"
)]
async def _db_health_check(self, args: Dict[str, Any]) -> List[types.TextContent]:
db_type = args.get("db_type")
results = {"db_type": db_type, "timestamp": time.time()}
if db_type == "sqlite":
db_path = args.get("db_path", "database.db")
results.update(await self._check_sqlite_health(db_path))
elif db_type == "mysql":
results.update(await self._check_mysql_health(args))
elif db_type == "postgresql":
results.update(await self._check_postgresql_health(args))
elif db_type == "mongodb":
results.update(await self._check_mongodb_health(args))
return [types.TextContent(
type="text",
text=f"Database Health Check:\n{json.dumps(results, indent=2)}"
)]
async def _check_sqlite_health(self, db_path: str) -> Dict[str, Any]:
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Basic info
cursor.execute("PRAGMA database_list")
db_info = cursor.fetchall()
# Table count
cursor.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table'")
table_count = cursor.fetchone()[0]
# Database size
import os
db_size = os.path.getsize(db_path) if os.path.exists(db_path) else 0
# Integrity check
cursor.execute("PRAGMA integrity_check")
integrity = cursor.fetchone()[0]
conn.close()
return {
"status": "healthy" if integrity == "ok" else "corrupted",
"database_info": db_info,
"table_count": table_count,
"size_bytes": db_size,
"size_mb": round(db_size / (1024*1024), 2),
"integrity_check": integrity
}
except Exception as e:
return {"status": "error", "error": str(e)}
async def _check_mysql_health(self, args: Dict[str, Any]) -> Dict[str, Any]:
try:
# Use subprocess to run mysql commands
cmd = ["mysql", "--version"]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
return {
"status": "mysql_available",
"version": result.stdout.strip(),
"note": "Full MySQL health check requires connection credentials"
}
else:
return {"status": "mysql_not_available", "error": result.stderr}
except Exception as e:
return {"status": "error", "error": str(e)}
async def _check_postgresql_health(self, args: Dict[str, Any]) -> Dict[str, Any]:
try:
cmd = ["psql", "--version"]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
return {
"status": "postgresql_available",
"version": result.stdout.strip(),
"note": "Full PostgreSQL health check requires connection credentials"
}
else:
return {"status": "postgresql_not_available", "error": result.stderr}
except Exception as e:
return {"status": "error", "error": str(e)}
async def _check_mongodb_health(self, args: Dict[str, Any]) -> Dict[str, Any]:
try:
cmd = ["mongod", "--version"]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
return {
"status": "mongodb_available",
"version": result.stdout.strip(),
"note": "Full MongoDB health check requires connection credentials"
}
else:
return {"status": "mongodb_not_available", "error": result.stderr}
except Exception as e:
return {"status": "error", "error": str(e)}
async def run(self):
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="database-debug",
server_version="1.0.0"
)
)
async def main():
server = DatabaseDebugServer()
await server.run()
if __name__ == "__main__":
asyncio.run(main())#!/usr/bin/env python3
"""
Container Debug MCP Server
Provides Docker/Podman container debugging capabilities
"""
import asyncio
import json
import subprocess
import time
from typing import Dict, List, Any
from mcp import Server, types
from mcp.server.models import InitializationOptions
class ContainerDebugServer:
def __init__(self):
self.server = Server("container-debug")
self.setup_handlers()
def setup_handlers(self):
@self.server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
return [
types.Tool(
name="container_health_check",
description="Check container health and status",
inputSchema={
"type": "object",
"properties": {
"runtime": {
"type": "string",
"enum": ["docker", "podman"],
"description": "Container runtime"
},
"container_id": {
"type": "string",
"description": "Container ID or name (optional for all containers)"
}
},
"required": ["runtime"]
}
),
types.Tool(
name="container_logs",
description="Analyze container logs for errors",
inputSchema={
"type": "object",
"properties": {
"runtime": {
"type": "string",
"enum": ["docker", "podman"],
"description": "Container runtime"
},
"container_id": {
"type": "string",
"description": "Container ID or name"
},
"lines": {
"type": "integer",
"description": "Number of recent log lines (default: 100)"
},
"follow": {
"type": "boolean",
"description": "Follow log output"
}
},
"required": ["runtime", "container_id"]
}
),
types.Tool(
name="container_resources",
description="Monitor container resource usage",
inputSchema={
"type": "object",
"properties": {
"runtime": {
"type": "string",
"enum": ["docker", "podman"],
"description": "Container runtime"
},
"container_id": {
"type": "string",
"description": "Container ID or name"
}
},
"required": ["runtime", "container_id"]
}
),
types.Tool(
name="container_exec",
description="Execute commands inside container for debugging",
inputSchema={
"type": "object",
"properties": {
"runtime": {
"type": "string",
"enum": ["docker", "podman"],
"description": "Container runtime"
},
"container_id": {
"type": "string",
"description": "Container ID or name"
},
"command": {
"type": "string",
"description": "Command to execute"
}
},
"required": ["runtime", "container_id", "command"]
}
)
]
@self.server.call_tool()
async def handle_call_tool(name: str, arguments: Dict[str, Any] | None) -> List[types.TextContent]:
if arguments is None:
arguments = {}
try:
if name == "container_health_check":
return await self._container_health_check(arguments)
elif name == "container_logs":
return await self._container_logs(arguments)
elif name == "container_resources":
return await self._container_resources(arguments)
elif name == "container_exec":
return await self._container_exec(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
return [types.TextContent(
type="text",
text=f"Error in {name}: {str(e)}"
)]
async def _container_health_check(self, args: Dict[str, Any]) -> List[types.TextContent]:
runtime = args.get("runtime", "docker")
container_id = args.get("container_id")
if container_id:
# Check specific container
cmd = [runtime, "inspect", container_id]
else:
# List all containers
cmd = [runtime, "ps", "-a", "--format", "json"]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if result.returncode == 0:
if container_id:
# Parse inspect output
inspect_data = json.loads(result.stdout)
container_info = inspect_data[0] if inspect_data else {}
health_info = {
"container_id": container_info.get("Id", "")[:12],
"name": container_info.get("Name", "").lstrip("/"),
"status": container_info.get("State", {}).get("Status", "unknown"),
"health": container_info.get("State", {}).get("Health", {}),
"restart_count": container_info.get("RestartCount", 0),
"created": container_info.get("Created", ""),
"image": container_info.get("Config", {}).get("Image", ""),
"ports": container_info.get("NetworkSettings", {}).get("Ports", {})
}
else:
# Parse ps output
containers = []
for line in result.stdout.strip().split('\n'):
if line:
try:
container_data = json.loads(line)
containers.append({
"id": container_data.get("ID", "")[:12],
"name": container_data.get("Names", ""),
"image": container_data.get("Image", ""),
"status": container_data.get("Status", ""),
"ports": container_data.get("Ports", "")
})
except json.JSONDecodeError:
continue
health_info = {
"total_containers": len(containers),
"containers": containers
}
return [types.TextContent(
type="text",
text=f"Container Health Check:\n{json.dumps(health_info, indent=2)}"
)]
else:
return [types.TextContent(
type="text",
text=f"Error running {runtime} command: {result.stderr}"
)]
except subprocess.TimeoutExpired:
return [types.TextContent(
type="text",
text=f"Timeout: {runtime} command took too long"
)]
except Exception as e:
return [types.TextContent(
type="text",
text=f"Error: {str(e)}"
)]
async def _container_logs(self, args: Dict[str, Any]) -> List[types.TextContent]:
runtime = args.get("runtime", "docker")
container_id = args.get("container_id")
lines = args.get("lines", 100)
cmd = [runtime, "logs", "--tail", str(lines), container_id]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if result.returncode == 0:
logs = result.stdout
error_logs = result.stderr
# Analyze logs for common error patterns
error_patterns = [
"error", "ERROR", "Error",
"exception", "Exception", "EXCEPTION",
"failed", "Failed", "FAILED",
"fatal", "Fatal", "FATAL",
"panic", "Panic", "PANIC"
]
errors_found = []
for line in logs.split('\n'):
for pattern in error_patterns:
if pattern in line:
errors_found.append(line.strip())
break
log_analysis = {
"container_id": container_id,
"total_lines": len(logs.split('\n')),
"errors_found": len(errors_found),
"error_lines": errors_found[:10], # Show first 10 errors
"recent_logs": logs.split('\n')[-20:] if logs else [] # Last 20 lines
}
return [types.TextContent(
type="text",
text=f"Container Logs Analysis:\n{json.dumps(log_analysis, indent=2)}"
)]
else:
return [types.TextContent(
type="text",
text=f"Error getting logs: {result.stderr}"
)]
except subprocess.TimeoutExpired:
return [types.TextContent(
type="text",
text="Timeout: Log retrieval took too long"
)]
except Exception as e:
return [types.TextContent(
type="text",
text=f"Error: {str(e)}"
)]
async def run(self):
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="container-debug",
server_version="1.0.0"
)
)
async def main():
server = ContainerDebugServer()
await server.run()
if __name__ == "__main__":
asyncio.run(main()){
"mcpServers": {
"system-debug": {
"command": "python3",
"args": ["/path/to/system_debug_server.py"],
"env": {
"PYTHONPATH": "/path/to/mcp/libs"
}
},
"database-debug": {
"command": "python3",
"args": ["/path/to/database_debug_server.py"],
"env": {
"DB_TIMEOUT": "30"
}
},
"container-debug": {
"command": "python3",
"args": ["/path/to/container_debug_server.py"],
"env": {
"DOCKER_HOST": "unix:///var/run/docker.sock"
}
}
}
}# Khởi động Q với debug servers
q chat --mcp-config ~/.config/q/debug-mcp-config.json
# Trong Q chat:
# "Hãy kiểm tra system health và tìm processes đang sử dụng CPU cao"Q sẽ tự động:
- Chạy system health check
- Phân tích CPU, memory, disk usage
- Liệt kê processes có CPU usage > 80%
- Đưa ra recommendations
# Trong Q chat:
# "Database SQLite của tôi chạy chậm, hãy kiểm tra health và phân tích"Q sẽ:
- Kiểm tra database integrity
- Phân tích database size
- Kiểm tra table count
- Đưa ra optimization suggestions
# Trong Q chat:
# "Container web-app của tôi keep crashing, hãy debug"Q sẽ:
- Kiểm tra container status
- Phân tích logs tìm errors
- Kiểm tra resource usage
- Suggest fixes
# health_monitor.py
import asyncio
import json
from datetime import datetime
async def continuous_monitoring():
"""Continuous system monitoring with MCP"""
while True:
# System health check
system_health = await call_mcp_tool("system_health_check", {
"check_type": "full"
})
# Database health check
db_health = await call_mcp_tool("db_health_check", {
"db_type": "sqlite",
"db_path": "/app/database.db"
})
# Container health check
container_health = await call_mcp_tool("container_health_check", {
"runtime": "docker"
})
# Combine results
health_report = {
"timestamp": datetime.now().isoformat(),
"system": system_health,
"database": db_health,
"containers": container_health
}
# Save report
with open(f"health_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", "w") as f:
json.dump(health_report, f, indent=2)
# Check for critical issues
if has_critical_issues(health_report):
await send_alert(health_report)
# Wait 5 minutes
await asyncio.sleep(300)
async def call_mcp_tool(tool_name, args):
"""Call MCP tool and return results"""
# Implementation depends on your MCP client setup
pass
def has_critical_issues(report):
"""Check if report contains critical issues"""
# Check system health
if report["system"].get("health_assessment", {}).get("status") == "critical":
return True
# Check database health
if report["database"].get("status") == "error":
return True
# Check container health
containers = report["containers"].get("containers", [])
for container in containers:
if "exited" in container.get("status", "").lower():
return True
return False
async def send_alert(report):
"""Send alert for critical issues"""
print(f"🚨 CRITICAL ALERT: {report['timestamp']}")
# Implement your alerting mechanism (email, Slack, etc.)# interactive_debug.py
import asyncio
class InteractiveDebugger:
def __init__(self):
self.session_log = []
async def start_debug_session(self, issue_type):
"""Start interactive debugging session"""
print(f"🔍 Starting debug session for: {issue_type}")
if issue_type == "performance":
await self._debug_performance()
elif issue_type == "database":
await self._debug_database()
elif issue_type == "container":
await self._debug_container()
async def _debug_performance(self):
"""Debug performance issues step by step"""
steps = [
("System Health Check", "system_health_check", {"check_type": "full"}),
("High CPU Processes", "process_analyzer", {"action": "list_high_cpu", "threshold": 70}),
("High Memory Processes", "process_analyzer", {"action": "list_high_memory", "threshold": 70}),
("Network Diagnostics", "network_diagnostics", {"test_type": "connectivity"}),
("Log Analysis", "log_analyzer", {"log_type": "syslog", "lines": 50})
]
for step_name, tool_name, args in steps:
print(f"\n📋 Step: {step_name}")
result = await self._call_mcp_tool(tool_name, args)
self.session_log.append({
"step": step_name,
"tool": tool_name,
"args": args,
"result": result,
"timestamp": datetime.now().isoformat()
})
# Ask user if they want to continue
continue_debug = input("Continue to next step? (y/n): ")
if continue_debug.lower() != 'y':
break
# Generate debug report
await self._generate_debug_report()
async def _generate_debug_report(self):
"""Generate comprehensive debug report"""
report = {
"session_start": self.session_log[0]["timestamp"] if self.session_log else None,
"session_end": datetime.now().isoformat(),
"total_steps": len(self.session_log),
"steps": self.session_log,
"recommendations": self._generate_recommendations()
}
report_file = f"debug_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(report_file, "w") as f:
json.dump(report, f, indent=2)
print(f"📄 Debug report saved: {report_file}")
def _generate_recommendations(self):
"""Generate recommendations based on debug results"""
recommendations = []
# Analyze session log and generate recommendations
for step in self.session_log:
if "high_cpu" in step["tool"]:
recommendations.append("Consider optimizing high CPU processes")
elif "memory" in step["tool"]:
recommendations.append("Monitor memory usage and consider adding swap")
elif "network" in step["tool"]:
recommendations.append("Check network connectivity and firewall rules")
return recommendations
# Usage
async def main():
debugger = InteractiveDebugger()
await debugger.start_debug_session("performance")
if __name__ == "__main__":
asyncio.run(main())# Secure debugging practices
ALLOWED_COMMANDS = [
"ps", "top", "htop", "free", "df", "netstat",
"docker ps", "docker logs", "docker inspect",
"systemctl status", "journalctl"
]
def validate_command(command):
"""Validate command for security"""
if not any(command.startswith(allowed) for allowed in ALLOWED_COMMANDS):
raise ValueError(f"Command not allowed: {command}")
# Check for dangerous patterns
dangerous_patterns = [";", "&&", "||", "|", ">", ">>", "<", "rm ", "del "]
if any(pattern in command for pattern in dangerous_patterns):
raise ValueError(f"Potentially dangerous command: {command}")
return True# Optimize MCP server performance
import functools
import time
def cache_result(ttl_seconds=60):
"""Cache MCP tool results"""
def decorator(func):
cache = {}
@functools.wraps(func)
async def wrapper(*args, **kwargs):
cache_key = str(args) + str(kwargs)
current_time = time.time()
if cache_key in cache:
result, timestamp = cache[cache_key]
if current_time - timestamp < ttl_seconds:
return result
result = await func(*args, **kwargs)
cache[cache_key] = (result, current_time)
return result
return wrapper
return decorator
# Usage
@cache_result(ttl_seconds=30)
async def get_system_health():
"""Cached system health check"""
# Expensive operation
passimport logging
import traceback
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('mcp_debug.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger("mcp-debug")
async def safe_tool_execution(tool_func, *args, **kwargs):
"""Safely execute MCP tool with proper error handling"""
try:
logger.info(f"Executing tool: {tool_func.__name__}")
result = await tool_func(*args, **kwargs)
logger.info(f"Tool completed successfully: {tool_func.__name__}")
return result
except Exception as e:
logger.error(f"Tool execution failed: {tool_func.__name__}")
logger.error(f"Error: {str(e)}")
logger.error(f"Traceback: {traceback.format_exc()}")
return {
"error": True,
"message": str(e),
"tool": tool_func.__name__,
"timestamp": time.time()
}- Permission Denied
# Fix permissions for system monitoring
sudo usermod -a -G docker $USER
sudo chmod +r /var/log/syslog- Missing Dependencies
# Install required packages
pip install psutil docker-py pymongo
sudo apt-get install mysql-client postgresql-client- Container Runtime Issues
# Check Docker/Podman status
systemctl status docker
systemctl status podman
# Test container access
docker ps
podman psMCP servers có thể trở thành công cụ debugging cực kỳ mạnh mẽ cho:
✅ OS Debugging:
- System health monitoring
- Process analysis
- Log analysis
- Network diagnostics
✅ Database Debugging:
- Health checks
- Performance analysis
- Connection testing
- Log analysis
✅ Container Debugging:
- Container health monitoring
- Log analysis
- Resource monitoring
- Command execution
Với các MCP servers này, bạn có thể tự động hóa việc debugging và có được insights sâu sắc về hệ thống một cách nhanh chóng!
Tác giả: ducla-cloud