Docker MCP Server

#!/usr/bin/env python # -*- coding: utf-8 -*- import subprocess from fastmcp import FastMCP from termcolor import colored import sys import os import shlex # Add shlex for proper command splitting import json import logging # Set up logging to stderr for MCP compatibility logging.basicConfig( level=logging.INFO, format="[DockerMCP] %(levelname)s: %(message)s", stream=sys.stderr ) # CONSTANTS SERVER_NAME = "DockerManager" def print_colored(message, color="green", prefix="[DockerMCP]"): """Print colored messages to stderr for better compatibility with MCP protocol.""" # Use stderr only to avoid interfering with stdout JSON communication print(colored(f"{prefix} {message}", color), file=sys.stderr) # Create an MCP server instance mcp = FastMCP(SERVER_NAME, disable_stdout_logging=True) # When the server starts, check if Docker is available try: docker_check = subprocess.run( ["docker", "--version"], capture_output=True, text=True ) print_colored(f"Docker version: {docker_check.stdout.strip()}", "cyan") except subprocess.SubprocessError: print_colored("Docker is not available. Please make sure Docker is installed and running.", "red") logging.error("Docker is not available. Please make sure Docker is installed and running.") sys.exit(1) except Exception as e: print_colored(f"Error checking Docker: {str(e)}", "red") logging.error(f"Error checking Docker: {str(e)}") sys.exit(1) @mcp.tool() def create_container(image: str, container_name: str, dependencies: str = "") -> str: """ Create and start a Docker container with optional dependencies. Parameters: • image: The Docker image to use (e.g., "ubuntu:latest", "node:16", "python:3.9-slim"). • container_name: A unique name for the container. • dependencies: Space-separated list of packages to install (e.g., "numpy pandas matplotlib" or "express lodash"). This tool uses 'docker run' in detached mode with a command that keeps the container running. If dependencies are specified, they will be installed after the container starts. Automatically detects appropriate package manager (pip, npm, apt, apk) based on the image. """ print_colored(f"Creating container with name '{container_name}' from image '{image}'...") try: # Start the container in detached mode with an infinite sleep to keep it running. result = subprocess.run( ["docker", "run", "-d", "--name", container_name, image, "sleep", "infinity"], capture_output=True, text=True, check=True, timeout=30 ) container_id = result.stdout.strip() print_colored(f"Container created successfully. ID: {container_id}") # Install dependencies if specified if dependencies: print_colored(f"Installing dependencies: {dependencies}", "cyan") # Determine package manager based on image if "node" in image.lower() or "javascript" in image.lower(): # For Node.js images, use npm print_colored("Detected Node.js image, using npm", "cyan") # First check if npm is available try: npm_check = subprocess.run( ["docker", "exec", container_name, "npm", "--version"], capture_output=True, text=True, check=True, timeout=10 ) print_colored(f"npm version: {npm_check.stdout.strip()}", "cyan") # For npm, we'll install packages globally to avoid needing a package.json install_cmd = f"npm install -g {dependencies}" except subprocess.CalledProcessError: error_msg = "npm not found in container, defaulting to other package managers" print_colored(error_msg, "yellow") # Fall through to other package manager detection elif "python" in image.lower(): # For Python images, use pip print_colored("Detected Python image, using pip", "cyan") install_cmd = f"pip install {dependencies}" elif any(distro in image.lower() for distro in ["ubuntu", "debian"]): # For Debian/Ubuntu images print_colored("Detected Debian/Ubuntu image, using apt-get", "cyan") install_cmd = f"apt-get update && apt-get install -y {dependencies}" elif "alpine" in image.lower(): # For Alpine images print_colored("Detected Alpine image, using apk", "cyan") install_cmd = f"apk add --no-cache {dependencies}" else: # Default to pip for unknown images with a warning print_colored("Image type not recognized, attempting to detect available package managers", "yellow") # Try to detect available package managers package_managers = [] # Check for npm try: npm_check = subprocess.run( ["docker", "exec", container_name, "npm", "--version"], capture_output=True, text=True, check=False, timeout=5 ) if npm_check.returncode == 0: package_managers.append("npm") except Exception: pass # Check for pip try: pip_check = subprocess.run( ["docker", "exec", container_name, "pip", "--version"], capture_output=True, text=True, check=False, timeout=5 ) if pip_check.returncode == 0: package_managers.append("pip") except Exception: pass # Choose package manager based on detection if "npm" in package_managers: print_colored("npm found, using it for installation", "cyan") install_cmd = f"npm install -g {dependencies}" elif "pip" in package_managers: print_colored("pip found, using it for installation", "cyan") install_cmd = f"pip install {dependencies}" else: print_colored("No known package managers detected, defaulting to pip install", "yellow") install_cmd = f"pip install {dependencies}" try: # Execute the installation command print_colored(f"Running: {install_cmd}", "cyan") install_result = subprocess.run( ["docker", "exec", container_name, "sh", "-c", install_cmd], capture_output=True, text=True, check=True, timeout=180 # Allow more time for installations ) print_colored(f"Dependencies installed successfully", "green") return f"Container created with ID: {container_id}\nDependencies installed: {dependencies}" except subprocess.CalledProcessError as e: error_msg = f"Error installing dependencies: {e.stderr}" print_colored(error_msg, "red") return f"Container created with ID: {container_id}\n{error_msg}" except subprocess.TimeoutExpired: error_msg = "Timeout while installing dependencies. Operation took too long." print_colored(error_msg, "red") return f"Container created with ID: {container_id}\n{error_msg}" except Exception as e: error_msg = f"Unexpected error installing dependencies: {str(e)}" print_colored(error_msg, "red") return f"Container created with ID: {container_id}\n{error_msg}" return f"Container created with ID: {container_id}" except subprocess.CalledProcessError as e: error_msg = f"Error creating container: {e.stderr}" print_colored(error_msg, "red") return error_msg except subprocess.TimeoutExpired: error_msg = "Timeout while creating container. Operation took too long." print_colored(error_msg, "red") return error_msg except Exception as e: error_msg = f"Unexpected error: {str(e)}" print_colored(error_msg, "red") return error_msg @mcp.tool() def execute_code(container_name: str, command: str) -> str: """ Execute a command inside a running Docker container. Parameters: • container_name: The name of the target container. • command: The command to execute inside the container. This tool uses 'docker exec' to run the command and returns the output. """ print_colored(f"Executing command in container '{container_name}': {command}") try: # Check if this is a python -c command if command.startswith("python -c"): # For python -c commands, keep the entire string after -c as a single argument prefix, python_code = command.split("-c", 1) cmd_parts = ["python", "-c", python_code.strip()] print_colored(f"Detected Python code execution: {cmd_parts}", "cyan") else: # For other commands, use proper shell-like splitting cmd_parts = shlex.split(command) # Execute the command result = subprocess.run( ["docker", "exec", container_name] + cmd_parts, capture_output=True, text=True, check=True, timeout=30 ) print_colored(f"Command executed successfully") return f"Command output: {result.stdout}" except subprocess.CalledProcessError as e: error_msg = f"Error executing command: {e.stderr}" print_colored(error_msg, "red") return error_msg except subprocess.TimeoutExpired: error_msg = "Timeout while executing command. Operation took too long." print_colored(error_msg, "red") return error_msg except Exception as e: error_msg = f"Unexpected error: {str(e)}" print_colored(error_msg, "red") return error_msg @mcp.tool() def execute_python_script(container_name: str, script_content: str, script_args: str = "") -> str: """ Execute a Python script inside a running Docker container. Parameters: • container_name: The name of the target container. • script_content: The full Python script content to execute. • script_args: Optional arguments to pass to the script (default: ""). This tool writes the script to a file in the container and executes it. """ print_colored(f"Executing Python script in container '{container_name}'") script_path = "/tmp/script.py" try: # Write the script content to a file in the container print_colored("Writing Python script to container...", "cyan") # Escape single quotes in the script content escaped_content = script_content.replace("'", "'\\''") write_result = subprocess.run( ["docker", "exec", container_name, "bash", "-c", f"echo '{escaped_content}' > {script_path} && chmod +x {script_path}"], capture_output=True, text=True, check=True, timeout=30 ) # Execute the script print_colored(f"Running Python script: {script_path} {script_args}", "cyan") cmd_parts = ["python", script_path] if script_args: cmd_parts.extend(shlex.split(script_args)) result = subprocess.run( ["docker", "exec", container_name] + cmd_parts, capture_output=True, text=True, check=True, timeout=60 # Allow more time for script execution ) print_colored("Python script executed successfully") return f"Command output: {result.stdout}" except subprocess.CalledProcessError as e: error_msg = f"Error executing Python script: {e.stderr}" print_colored(error_msg, "red") return error_msg except subprocess.TimeoutExpired: error_msg = "Timeout while executing Python script. Operation took too long." print_colored(error_msg, "red") return error_msg except Exception as e: error_msg = f"Unexpected error: {str(e)}" print_colored(error_msg, "red") return error_msg @mcp.tool() def cleanup_container(container_name: str) -> str: """ Stop and remove a Docker container. Parameters: • container_name: The name of the container to stop and remove. This tool uses 'docker stop' followed by 'docker rm' to clean up the container. """ print_colored(f"Cleaning up container '{container_name}'...") try: # Stop the container with a shorter timeout (5 seconds instead of default 10) print_colored(f"Stopping container '{container_name}'...", "yellow") stop_result = subprocess.run( ["docker", "stop", "--time", "5", container_name], capture_output=True, text=True, check=True, timeout=10 ) # Remove the container print_colored(f"Removing container '{container_name}'...", "yellow") rm_result = subprocess.run( ["docker", "rm", container_name], capture_output=True, text=True, check=True, timeout=10 ) print_colored(f"Container '{container_name}' has been successfully stopped and removed.") return f"Container '{container_name}' has been stopped and removed." except subprocess.CalledProcessError as e: error_msg = f"Error cleaning the container: {e.stderr}" print_colored(error_msg, "red") # If stop fails, try to force kill the container try: print_colored(f"Attempting to force kill container '{container_name}'...", "yellow") kill_result = subprocess.run( ["docker", "kill", container_name], capture_output=True, text=True, check=False, timeout=5 ) rm_result = subprocess.run( ["docker", "rm", "-f", container_name], capture_output=True, text=True, check=False, timeout=5 ) if rm_result.returncode == 0: print_colored(f"Container '{container_name}' has been forcibly removed.") return f"Container '{container_name}' has been forcibly removed after timeout." except Exception: pass return error_msg except subprocess.TimeoutExpired: error_msg = "Timeout while cleaning up container. Attempting to force remove..." print_colored(error_msg, "yellow") # If timeout occurs, try to force remove the container try: kill_result = subprocess.run( ["docker", "kill", container_name], capture_output=True, text=True, check=False, timeout=5 ) rm_result = subprocess.run( ["docker", "rm", "-f", container_name], capture_output=True, text=True, check=False, timeout=5 ) if rm_result.returncode == 0: print_colored(f"Container '{container_name}' has been forcibly removed after timeout.") return f"Container '{container_name}' has been forcibly removed after timeout." else: return "Failed to clean up container after timeout." except Exception as e: return f"Failed to force remove container: {str(e)}" except Exception as e: error_msg = f"Unexpected error: {str(e)}" print_colored(error_msg, "red") return error_msg @mcp.tool() def add_dependencies(container_name: str, dependencies: str) -> str: """ Install additional dependencies in an existing Docker container. Parameters: • container_name: The name of the target container. • dependencies: Space-separated list of packages to install (e.g., "numpy pandas matplotlib" or "express lodash"). This tool automatically detects the appropriate package manager (pip, npm, apt, apk) available in the container and uses it to install the specified dependencies. """ print_colored(f"Adding dependencies to container '{container_name}': {dependencies}") try: # Check if container exists and is running container_check = subprocess.run( ["docker", "container", "inspect", "-f", "{{.State.Running}}", container_name], capture_output=True, text=True, check=True, timeout=10 ) if container_check.stdout.strip() != "true": error_msg = f"Container '{container_name}' is not running or does not exist" print_colored(error_msg, "red") return error_msg # Try to detect available package managers package_managers = [] # Check for npm try: npm_check = subprocess.run( ["docker", "exec", container_name, "npm", "--version"], capture_output=True, text=True, check=False, timeout=5 ) if npm_check.returncode == 0: package_managers.append("npm") print_colored(f"Found npm: {npm_check.stdout.strip()}", "cyan") except Exception: pass # Check for pip try: pip_check = subprocess.run( ["docker", "exec", container_name, "pip", "--version"], capture_output=True, text=True, check=False, timeout=5 ) if pip_check.returncode == 0: package_managers.append("pip") print_colored(f"Found pip: {pip_check.stdout.strip()}", "cyan") except Exception: pass # Check for apt-get try: apt_check = subprocess.run( ["docker", "exec", container_name, "apt-get", "--version"], capture_output=True, text=True, check=False, timeout=5 ) if apt_check.returncode == 0: package_managers.append("apt-get") print_colored("Found apt-get", "cyan") except Exception: pass # Check for apk try: apk_check = subprocess.run( ["docker", "exec", container_name, "apk", "--version"], capture_output=True, text=True, check=False, timeout=5 ) if apk_check.returncode == 0: package_managers.append("apk") print_colored("Found apk", "cyan") except Exception: pass # Choose package manager based on detection if not package_managers: error_msg = "No supported package managers found in the container" print_colored(error_msg, "red") return error_msg print_colored(f"Available package managers: {', '.join(package_managers)}", "cyan") # Choose the appropriate package manager if "npm" in package_managers: print_colored("Using npm for installation", "cyan") install_cmd = f"npm install -g {dependencies}" elif "pip" in package_managers: print_colored("Using pip for installation", "cyan") install_cmd = f"pip install {dependencies}" elif "apt-get" in package_managers: print_colored("Using apt-get for installation", "cyan") install_cmd = f"apt-get update && apt-get install -y {dependencies}" elif "apk" in package_managers: print_colored("Using apk for installation", "cyan") install_cmd = f"apk add --no-cache {dependencies}" else: error_msg = "No supported package managers found in the container" print_colored(error_msg, "red") return error_msg # Execute the installation command print_colored(f"Running: {install_cmd}", "cyan") install_result = subprocess.run( ["docker", "exec", container_name, "sh", "-c", install_cmd], capture_output=True, text=True, check=True, timeout=180 # Allow more time for installations ) print_colored("Dependencies installed successfully", "green") return f"Dependencies installed in container '{container_name}': {dependencies}" except subprocess.CalledProcessError as e: error_msg = f"Error installing dependencies: {e.stderr}" print_colored(error_msg, "red") return error_msg except subprocess.TimeoutExpired: error_msg = "Timeout while installing dependencies. Operation took too long." print_colored(error_msg, "red") return error_msg except Exception as e: error_msg = f"Unexpected error: {str(e)}" print_colored(error_msg, "red") return error_msg @mcp.tool() def list_containers(show_all: bool = True) -> str: """ List all Docker containers with their details. Parameters: • show_all: Whether to show all containers including stopped ones (default: True). If False, only shows running containers. Returns information about containers including ID, name, status, and image. """ print_colored(f"Listing {'all' if show_all else 'running'} containers...") try: cmd = ["docker", "ps"] if show_all: cmd.append("-a") # Show all containers, not just running ones # Add format to get consistent, parseable output with specific fields cmd.extend([ "--format", "table {{.ID}}\t{{.Names}}\t{{.Status}}\t{{.Image}}\t{{.RunningFor}}" ]) result = subprocess.run( cmd, capture_output=True, text=True, check=True, timeout=15 ) container_list = result.stdout.strip() if container_list and not container_list.startswith("CONTAINER ID"): # This means we have a table header but no data print_colored("No containers found", "yellow") return f"No {'existing' if show_all else 'running'} containers found." print_colored(f"Found containers:\n{container_list}", "cyan") # Add extra information about how to use this data usage_info = "\nYou can use these container names with other tools like add_dependencies, execute_code, etc." if show_all: usage_info += "\nNote: Only containers with 'Up' in their Status are currently running and can be interacted with." return f"{container_list}{usage_info}" except subprocess.CalledProcessError as e: error_msg = f"Error listing containers: {e.stderr}" print_colored(error_msg, "red") return error_msg except subprocess.TimeoutExpired: error_msg = "Timeout while listing containers. Operation took too long." print_colored(error_msg, "red") return error_msg except Exception as e: error_msg = f"Unexpected error: {str(e)}" print_colored(error_msg, "red") return error_msg # Main entry point for MCP server if __name__ == "__main__": print_colored("Starting MCP Docker Manager server...", "green") try: # This starts the MCP server mcp.run() except Exception as e: error_message = f"Error starting MCP server: {str(e)}" logging.error(error_message) print(error_message, file=sys.stderr) sys.exit(1)