MCP Docker Server

#!/usr/bin/env python3 """ Docker MCP Server for Claude Code Handles Docker and Docker Compose commands from Claude Code running in a container. Uses MCP Python SDK 1.12.4+ (FastMCP) """ import asyncio import json import logging import os import shlex import subprocess import sys from typing import Any, Dict, List, Optional # MCP Server imports for FastMCP from mcp.server.fastmcp import FastMCP, Context # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class DockerCommandHandler: """Handler for Docker commands with security and validation.""" def __init__(self): self.allowed_commands = { 'docker', 'docker-compose' } def is_allowed_command(self, command: str) -> bool: """Check if a command is allowed.""" try: cmd_parts = shlex.split(command.lower()) if not cmd_parts: return False # Check for docker commands if cmd_parts[0] == 'docker': return True # Check for docker-compose commands if cmd_parts[0] == 'docker-compose': return True # Check for "docker compose" (new syntax) if len(cmd_parts) >= 2 and cmd_parts[0] == 'docker' and cmd_parts[1] == 'compose': return True return False except ValueError: # shlex.split failed, command is malformed return False async def execute_command(self, command: str, working_directory: Optional[str] = None) -> Dict[str, Any]: """Execute a shell command safely.""" try: logger.info(f"Executing command: {command}") # Parse command safely cmd_parts = shlex.split(command) # Set up environment env = os.environ.copy() # Execute command process = await asyncio.create_subprocess_exec( *cmd_parts, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=working_directory, env=env ) stdout, stderr = await process.communicate() return { "success": process.returncode == 0, "stdout": stdout.decode('utf-8', errors='replace'), "stderr": stderr.decode('utf-8', errors='replace'), "return_code": process.returncode } except Exception as e: logger.error(f"Error executing command: {e}") return { "success": False, "stdout": "", "stderr": str(e), "return_code": -1 } # Create command handler instance docker_handler = DockerCommandHandler() # Create the FastMCP server - configure host/port via environment variables host = os.getenv("FASTMCP_HOST", "0.0.0.0") # Bind to all interfaces for container access port = int(os.getenv("FASTMCP_PORT", "3000")) # Default to 3000, can be overridden # Set environment variables for server configuration os.environ["FASTMCP_HOST"] = host os.environ['FASTMCP_PORT'] = str(port) mcp = FastMCP("docker-mcp-server", host=host, port=port) @mcp.tool() async def execute_docker_command(command: str, working_directory: str = None) -> str: """ Execute Docker or Docker Compose commands on the host system. Args: command: The Docker command to execute (e.g., 'docker ps', 'docker-compose up -d') working_directory: Optional working directory for the command Returns: Command execution result with output and error information """ # Validate command is allowed if not docker_handler.is_allowed_command(command): return f"Error: Command '{command}' is not allowed. Only Docker and Docker Compose commands are supported." # Execute command result = await docker_handler.execute_command(command, working_directory) # Format response if result["success"]: response = f"Command executed successfully:\n\n{result['stdout']}" if result["stderr"]: response += f"\n\nWarnings/Info:\n{result['stderr']}" else: response = f"Command failed (exit code: {result['return_code']}):\n\n" if result["stderr"]: response += f"Error: {result['stderr']}\n" if result["stdout"]: response += f"Output: {result['stdout']}\n" return response @mcp.tool() async def docker_system_info() -> str: """ Get Docker system information and status. Returns: Docker version and system information """ # Get Docker version version_result = await docker_handler.execute_command("docker version --format json") # Get Docker system info info_result = await docker_handler.execute_command("docker system df") response = "Docker System Information:\n\n" if version_result["success"]: try: version_data = json.loads(version_result["stdout"]) client_version = version_data.get('Client', {}).get('Version', 'Unknown') server_version = version_data.get('Server', {}).get('Version', 'Unknown') response += f"Client Version: {client_version}\n" response += f"Server Version: {server_version}\n\n" except json.JSONDecodeError: response += "Version: Could not parse version info\n\n" if info_result["success"]: response += "Disk Usage:\n" response += info_result["stdout"] else: response += f"Could not get system info: {info_result['stderr']}" return response @mcp.tool() async def list_containers(all: bool = False) -> str: """ List Docker containers with detailed information. Args: all: Show all containers (including stopped ones) Returns: List of Docker containers in table format """ command = "docker ps --format table" if all: command += " -a" result = await docker_handler.execute_command(command) if result["success"]: response = f"Docker Containers {'(including stopped)' if all else '(running only)'}:\n\n" response += result["stdout"] else: response = f"Error listing containers: {result['stderr']}" return response @mcp.tool() async def list_images(all: bool = False) -> str: """ List Docker images. Args: all: Show all images (including intermediate ones) Returns: List of Docker images in table format """ command = "docker images --format table" if all: command += " -a" result = await docker_handler.execute_command(command) if result["success"]: response = f"Docker Images {'(including intermediate)' if all else ''}:\n\n" response += result["stdout"] else: response = f"Error listing images: {result['stderr']}" return response @mcp.tool() async def docker_compose_status(working_directory: str = None) -> str: """ Get Docker Compose service status. Args: working_directory: Directory containing docker-compose.yml Returns: Status of Docker Compose services """ result = await docker_handler.execute_command("docker-compose ps", working_directory) if result["success"]: response = "Docker Compose Services:\n\n" response += result["stdout"] else: response = f"Error getting compose status: {result['stderr']}" return response async def check_docker_availability(): """Check if Docker is available and running.""" try: process = await asyncio.create_subprocess_exec( "docker", "version", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) await process.communicate() if process.returncode != 0: logger.error("Docker is not available or not running") return False return True except FileNotFoundError: logger.error("Docker is not installed") return False def main(): """Main entry point for the MCP server.""" # Parse command line arguments transport = "stdio" # default if len(sys.argv) > 1: if sys.argv[1] in ["sse", "streamable-http", "stdio"]: transport = sys.argv[1] # Override port if provided as second argument if len(sys.argv) > 2: try: new_port = int(sys.argv[2]) os.environ['FASTMCP_PORT'] = str(new_port) logger.info(f"Port overridden to: {new_port}") except ValueError: logger.error(f"Invalid port: {sys.argv[2]}") sys.exit(1) logger.info(f"Starting Docker MCP Server with transport: {transport}") logger.info(f"Configuration: HOST={os.environ['FASTMCP_HOST']}, PORT={os.environ['FASTMCP_PORT']}") # Check Docker availability if not asyncio.run(check_docker_availability()): sys.exit(1) # Run the server try: if transport == "stdio": logger.info("Running as stdio server") mcp.run() else: logger.info(f"Running as {transport} server on {os.environ['FASTMCP_HOST']}:{os.environ['FASTMCP_PORT']}") mcp.run(transport=transport) except Exception as e: logger.error(f"Failed to start server: {e}") logger.info("Try running with different transport: python main.py stdio") sys.exit(1) if __name__ == "__main__": main()