diff --git a/.env.example b/.env.example index a15a22c..d838afb 100644 --- a/.env.example +++ b/.env.example @@ -9,7 +9,7 @@ # - https://portainer.example.com # - https://portainer.company.com:9443 # - http://localhost:9000 -PORTAINER_URL=https://portainer.example.com +PORTAINER_URL=https://your-portainer-instance.com # Portainer API key for authentication (required) # Generate this from Portainer UI: User settings > API tokens @@ -40,3 +40,6 @@ LOG_FORMAT=json # Development settings DEBUG=false +# MCP mode - disables stdout logging +MCP_MODE=true + diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d15a04f --- /dev/null +++ b/.gitignore @@ -0,0 +1,56 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# Virtual Environment +.venv/ +venv/ +ENV/ +env/ + +# Environment variables +.env +.env.local +.env.*.local + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Logs +*.log + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# MCP specific +.docker_cleaned +*.pyc \ No newline at end of file diff --git a/README_DOCKER.md b/README_DOCKER.md new file mode 100644 index 0000000..9c72b91 --- /dev/null +++ b/README_DOCKER.md @@ -0,0 +1,239 @@ +# Portainer Docker MCP Server + +This MCP server provides comprehensive Docker and Docker Swarm management functionality through Portainer Business Edition. + +## Features + +### Container Management +- **List and Inspect Containers** + - View all containers with status, ports, and resource usage + - Get detailed container information including networks, mounts, and environment + - Support for both running and stopped containers + +- **Container Lifecycle** + - Create containers with custom configurations + - Start, stop, and restart containers + - Remove containers (with force option) + - View container logs with tail and timestamp options + +### Image Management +- **Docker Images** + - List all images with sizes and tags + - Pull images from registries + - Remove unused images + - Support for multi-tag images + +### Volume Management +- **Docker Volumes** + - List volumes with mount points + - Create named volumes + - Remove volumes + - Support for different volume drivers + +### Network Management +- **Docker Networks** + - List networks with details + - Create custom networks (bridge, overlay, etc.) + - Remove networks + - Configure internal networks + +### Docker Swarm Features +- **Service Management** + - List services with replica status + - Create services with scaling options + - Update service configurations + - Remove services + - View service logs + +- **Stack Deployment** + - Deploy stacks from Docker Compose files + - List stacks across environments + - Remove stacks with all resources + - Support for environment variables in stacks + +### System Information +- **Docker Info** + - System-wide Docker information + - Resource usage statistics + - Swarm cluster status + - Version information + +## Installation + +1. Ensure Python dependencies are installed: +```bash +cd /Users/adelorenzo/repos/portainer-mcp +.venv/bin/pip install mcp httpx +``` + +2. Configure in Claude Desktop (`claude_desktop_config.json`): +```json +{ + "mcpServers": { + "portainer-docker": { + "command": "/path/to/.venv/bin/python", + "args": ["/path/to/portainer_docker_server.py"], + "env": { + "PORTAINER_URL": "https://your-portainer-instance.com", + "PORTAINER_API_KEY": "your-api-key", + "MCP_MODE": "true" + } + } + } +} +``` + +3. Restart Claude Desktop + +## Available Tools + +### Container Operations +- `list_containers` - List all containers in an environment +- `inspect_container` - Get detailed container information +- `create_container` - Create a new container +- `start_container` - Start a stopped container +- `stop_container` - Stop a running container +- `restart_container` - Restart a container +- `remove_container` - Remove a container +- `get_container_logs` - View container logs + +### Image Operations +- `list_images` - List Docker images +- `pull_image` - Pull an image from registry +- `remove_image` - Remove an image + +### Volume Operations +- `list_volumes` - List Docker volumes +- `create_volume` - Create a new volume +- `remove_volume` - Remove a volume + +### Network Operations +- `list_networks` - List Docker networks +- `create_network` - Create a new network +- `remove_network` - Remove a network + +### Swarm Service Operations +- `list_services` - List Swarm services +- `create_service` - Create a new service +- `update_service` - Update service configuration +- `remove_service` - Remove a service +- `get_service_logs` - View service logs + +### Stack Operations +- `list_stacks` - List deployed stacks +- `deploy_stack` - Deploy a new stack +- `remove_stack` - Remove a stack + +### System Operations +- `get_docker_info` - Get Docker system information +- `get_docker_version` - Get Docker version details + +## Usage Examples + +### Container Management +``` +# List all containers in environment 3 +Use "list_containers" with environment_id: 3, all: true + +# Create and start an nginx container +Use "create_container" with: +- environment_id: 3 +- image: "nginx:latest" +- name: "my-nginx" +- ports: {"80/tcp": [{"HostPort": "8080"}]} + +# View container logs +Use "get_container_logs" with: +- environment_id: 3 +- container_id: "my-nginx" +- tail: 50 +- timestamps: true +``` + +### Docker Swarm Services +``` +# Create a replicated service +Use "create_service" with: +- environment_id: 4 +- name: "web-service" +- image: "nginx:alpine" +- replicas: 3 +- ports: [{"target": 80, "published": 8080, "protocol": "tcp"}] + +# Scale a service +Use "update_service" with: +- environment_id: 4 +- service_id: "web-service" +- replicas: 5 +``` + +### Stack Deployment +``` +# Deploy a stack +Use "deploy_stack" with: +- environment_id: 4 +- name: "my-app" +- compose_file: | + version: '3.8' + services: + web: + image: nginx:alpine + ports: + - "8080:80" + redis: + image: redis:alpine +``` + +### Docker Volumes +``` +# Create a volume +Use "create_volume" with: +- environment_id: 3 +- name: "app-data" +- driver: "local" + +# List volumes +Use "list_volumes" with environment_id: 3 +``` + +## Container Configuration + +When creating containers, you can specify: +- **Image**: Docker image name with optional tag +- **Name**: Container name (optional) +- **Command**: Override default command +- **Environment Variables**: Array of KEY=VALUE strings +- **Ports**: Port bindings (host:container mapping) +- **Volumes**: Volume mounts +- **Restart Policy**: no, always, unless-stopped, on-failure + +## Swarm vs Standalone + +This server automatically handles both: +- **Docker Standalone**: Container operations, local volumes, bridge networks +- **Docker Swarm**: Services, stacks, overlay networks, multi-node deployments + +Some operations (like services) only work on Swarm environments. + +## Error Handling + +The server provides clear error messages: +- Environment not found +- Container/service not found +- Swarm-only operations on standalone Docker +- Network connectivity issues +- Permission errors + +## Security Notes + +- API key is required for all operations +- Container operations respect Portainer's RBAC +- Sensitive environment variables are filtered from logs +- Force options available for cleanup operations + +## Limitations + +- Log output is limited to prevent overwhelming responses +- Large container lists are truncated (shows first 20) +- Stack deployment requires compose file as string +- Some Docker features may require direct API access \ No newline at end of file diff --git a/README_ENVIRONMENTS.md b/README_ENVIRONMENTS.md new file mode 100644 index 0000000..79e7078 --- /dev/null +++ b/README_ENVIRONMENTS.md @@ -0,0 +1,196 @@ +# Portainer Environments MCP Server + +This MCP server provides comprehensive environment and endpoint management functionality for Portainer Business Edition. + +## Features + +### Environment Management +- **List and View Environments** + - List all environments with status and type + - Get detailed information about specific environments + - View environment statistics and Docker/Kubernetes info + +- **Create Environments** + - Docker environments (local and remote) + - Kubernetes clusters + - Docker Swarm clusters + - Edge Agent environments + - Support for TLS configuration + +- **Update and Delete** + - Update environment settings and URLs + - Change group assignments + - Delete environments safely + +### Environment Organization +- **Environment Groups** + - Create and manage environment groups + - Organize environments by purpose or location + - Update group names and descriptions + - Delete unused groups + +- **Tags Management** + - Create tags for categorizing environments + - List all available tags + - Delete unused tags + - Assign multiple tags to environments + +### Access Control +- **Team Associations** + - Associate environments with teams + - Set read/write access levels + - Bulk update team permissions + +### Edge Computing +- **Edge Agent Deployment** + - Generate Edge keys for remote deployments + - Get deployment scripts automatically + - Support for Edge environment groups + +## Installation + +1. Ensure Python dependencies are installed: +```bash +cd /Users/adelorenzo/repos/portainer-mcp +.venv/bin/pip install mcp httpx +``` + +2. Configure in Claude Desktop (`claude_desktop_config.json`): +```json +{ + "mcpServers": { + "portainer-environments": { + "command": "/path/to/.venv/bin/python", + "args": ["/path/to/portainer_environments_server.py"], + "env": { + "PORTAINER_URL": "https://your-portainer-instance.com", + "PORTAINER_API_KEY": "your-api-key", + "MCP_MODE": "true" + } + } + } +} +``` + +3. Restart Claude Desktop + +## Available Tools + +### Environment Operations +- `list_environments` - List all environments with pagination support +- `get_environment` - Get detailed information about an environment +- `create_docker_environment` - Create a new Docker environment +- `create_kubernetes_environment` - Create a new Kubernetes environment +- `update_environment` - Update environment settings +- `delete_environment` - Delete an environment +- `get_environment_status` - Get real-time status and statistics + +### Access Control +- `associate_environment` - Associate environments with teams and set permissions + +### Organization +- `list_environment_groups` - List all environment groups +- `create_environment_group` - Create a new environment group +- `update_environment_group` - Update group information +- `delete_environment_group` - Delete an environment group + +### Tags +- `list_tags` - List all available tags +- `create_tag` - Create a new tag +- `delete_tag` - Delete a tag + +### Edge Computing +- `generate_edge_key` - Generate Edge agent deployment script + +## Environment Types + +The server supports these environment types: +- **Docker** - Standard Docker environments +- **Docker Swarm** - Swarm cluster management +- **Kubernetes** - Kubernetes cluster management +- **Azure ACI** - Azure Container Instances +- **Edge Agent** - Remote Edge deployments + +## Example Usage + +### Create a Docker environment: +``` +Use "create_docker_environment" with: +- name: "Production Docker" +- url: "tcp://docker-prod.example.com:2375" +- public_url: "https://docker-prod.example.com" +- tls: true +- tags: ["production", "docker"] +``` + +### Create a Kubernetes environment: +``` +Use "create_kubernetes_environment" with: +- name: "K8s Production" +- url: "https://k8s-api.example.com:6443" +- bearer_token: "your-bearer-token" +- tls_skip_verify: false +``` + +### Deploy an Edge agent: +``` +1. Use "generate_edge_key" with name: "Remote Site 1" +2. Copy the generated deployment command +3. Run the command on the remote Docker host +``` + +### Organize environments: +``` +1. Use "create_environment_group" with name: "Production Servers" +2. Use "create_tag" with name: "critical" +3. Use "update_environment" to assign the group and tags +``` + +## API Compatibility + +This server handles both old and new Portainer API endpoints: +- New API (2.19.x+): `/environments` +- Old API (pre-2.19): `/endpoints` + +The server automatically tries both endpoints for compatibility. + +## Security Notes + +- API key is required for all operations +- HTTPS is recommended (SSL verification disabled for development) +- Team associations respect Portainer's RBAC system +- Edge keys should be kept secure + +## Troubleshooting + +### Common Issues + +**Environment shows as "down":** +- Check network connectivity to the Docker/Kubernetes API +- Verify TLS certificates if using secure connections +- Ensure the API endpoint URL is correct + +**Cannot create environment:** +- Verify the URL format (tcp:// for Docker, https:// for Kubernetes) +- Check if the environment name already exists +- Ensure proper authentication credentials + +**Edge agent not connecting:** +- Verify the Edge key is correct +- Check firewall rules for outbound connections +- Ensure Docker is running on the Edge device + +## Docker URL Formats + +- **Local Docker:** `unix:///var/run/docker.sock` +- **Remote Docker (TCP):** `tcp://hostname:2375` +- **Remote Docker (TLS):** `tcp://hostname:2376` +- **Docker Desktop (Mac):** `unix:///$HOME/.docker/run/docker.sock` +- **Docker Desktop (Windows):** `npipe:////./pipe/docker_engine` + +## Kubernetes Authentication + +For Kubernetes environments, you can use: +- **Bearer Token:** Service account token +- **Client Certificate:** Upload certificate files (in UI) +- **Kubeconfig:** Import kubeconfig file (in UI) \ No newline at end of file diff --git a/README_MERGED.md b/README_MERGED.md new file mode 100644 index 0000000..7120026 --- /dev/null +++ b/README_MERGED.md @@ -0,0 +1,133 @@ +# Portainer Unified MCP Server + +This MCP server combines **portainer-core** and **portainer-teams** functionality into a single unified server, providing comprehensive user, team, and RBAC management for Portainer Business Edition. + +## Features + +### User Management (from portainer-core) +- **Authentication & Session Management** + - Test connection to Portainer + - API token validation +- **User CRUD Operations** + - List all users + - Create new users with role assignment + - Update user passwords and roles + - Delete users + +### Teams Management (from portainer-teams) +- **Team Operations** + - List all teams + - Create new teams with optional leaders + - Delete teams +- **Team Membership** + - Add users to teams + - Remove users from teams + - Bulk membership operations + +### RBAC Management +- **Role Management** + - List available roles with descriptions + - View role priorities and permissions +- **Resource Access Control** + - List all resource controls + - Create resource-specific access controls + - Configure public/private access + - Set user and team permissions + +### Settings Management +- **System Configuration** + - Get current Portainer settings + - Update security settings + - Configure user permissions + +## Installation + +1. Install Python dependencies: +```bash +cd /Users/adelorenzo/repos/portainer-mcp +.venv/bin/pip install mcp httpx +``` + +2. Configure in Claude Desktop (`claude_desktop_config.json`): +```json +{ + "mcpServers": { + "portainer": { + "command": "/path/to/.venv/bin/python", + "args": ["/path/to/merged_mcp_server.py"], + "env": { + "PORTAINER_URL": "https://your-portainer-instance.com", + "PORTAINER_API_KEY": "your-api-key", + "MCP_MODE": "true" + } + } + } +} +``` + +3. Restart Claude Desktop + +## Available Tools + +### User Management +- `test_connection` - Test connection to Portainer +- `get_users` - List all users with their roles +- `create_user` - Create a new user (username, password, role) +- `update_user` - Update user password or role +- `delete_user` - Delete a user by ID + +### Teams Management +- `get_teams` - List all teams +- `create_team` - Create a new team with optional leaders +- `add_team_members` - Add users to a team +- `remove_team_members` - Remove users from a team +- `delete_team` - Delete a team by ID + +### RBAC Management +- `get_roles` - List available roles and their descriptions +- `get_resource_controls` - List all resource access controls +- `create_resource_control` - Create access control for a resource + +### Settings +- `get_settings` - Get current Portainer settings +- `update_settings` - Update Portainer security settings + +## Role Types + +The server supports three role types: +- **Administrator** - Full system access +- **StandardUser** - Regular user access +- **ReadOnlyUser** - Read-only access + +## Example Usage + +### Create a user and add to team: +``` +1. Use "create_user" with username: "john", password: "secure123", role: "StandardUser" +2. Use "get_users" to find John's user ID +3. Use "create_team" with name: "DevOps Team" +4. Use "get_teams" to find the team ID +5. Use "add_team_members" with the team ID and John's user ID +``` + +### Set up resource access control: +``` +1. Use "create_resource_control" with: + - resource_id: "container_id_here" + - resource_type: "container" + - teams: [{"team_id": 1, "access_level": "read"}] + - administrators_only: false +``` + +## Compatibility + +- Supports both old (integer) and new (string) role formats +- Works with Portainer Business Edition 2.30.x+ +- Handles API version differences automatically + +## Security Notes + +- API key is required for all operations +- HTTPS is recommended (SSL verification disabled for development) +- Tokens should be rotated regularly +- All operations respect Portainer's RBAC system \ No newline at end of file diff --git a/merged_mcp_server.py b/merged_mcp_server.py new file mode 100755 index 0000000..455c2c8 --- /dev/null +++ b/merged_mcp_server.py @@ -0,0 +1,613 @@ +#!/usr/bin/env python3 +"""Merged MCP server for Portainer Core + Teams functionality.""" + +import os +import sys +import json +import asyncio +from typing import Any, Dict, List, Optional + +# Suppress all logging to stderr +os.environ["MCP_MODE"] = "true" + +import mcp.server.stdio +import mcp.types as types +from mcp.server import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +# Create server +server = Server("portainer-unified") + +# Store for our state +portainer_url = os.getenv("PORTAINER_URL", "https://partner.portainer.live") +api_key = os.getenv("PORTAINER_API_KEY", "") + + +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + # Core tools + types.Tool( + name="test_connection", + description="Test connection to Portainer", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + + # User Management tools + types.Tool( + name="get_users", + description="Get list of Portainer users", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="create_user", + description="Create a new Portainer user", + inputSchema={ + "type": "object", + "properties": { + "username": { + "type": "string", + "description": "Username for the new user" + }, + "password": { + "type": "string", + "description": "Password for the new user" + }, + "role": { + "type": "string", + "description": "User role", + "enum": ["Administrator", "StandardUser", "ReadOnlyUser"], + "default": "StandardUser" + } + }, + "required": ["username", "password"] + } + ), + types.Tool( + name="update_user", + description="Update an existing user", + inputSchema={ + "type": "object", + "properties": { + "user_id": { + "type": "integer", + "description": "ID of the user to update" + }, + "password": { + "type": "string", + "description": "New password (optional)" + }, + "role": { + "type": "string", + "description": "New role (optional)", + "enum": ["Administrator", "StandardUser", "ReadOnlyUser"] + } + }, + "required": ["user_id"] + } + ), + types.Tool( + name="delete_user", + description="Delete a user", + inputSchema={ + "type": "object", + "properties": { + "user_id": { + "type": "integer", + "description": "ID of the user to delete" + } + }, + "required": ["user_id"] + } + ), + + # Teams Management tools + types.Tool( + name="get_teams", + description="Get list of teams", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="create_team", + description="Create a new team", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Name of the team" + }, + "leaders": { + "type": "array", + "items": {"type": "integer"}, + "description": "Array of user IDs to be team leaders (optional)" + } + }, + "required": ["name"] + } + ), + types.Tool( + name="add_team_members", + description="Add members to a team", + inputSchema={ + "type": "object", + "properties": { + "team_id": { + "type": "integer", + "description": "ID of the team" + }, + "user_ids": { + "type": "array", + "items": {"type": "integer"}, + "description": "Array of user IDs to add to the team" + } + }, + "required": ["team_id", "user_ids"] + } + ), + types.Tool( + name="remove_team_members", + description="Remove members from a team", + inputSchema={ + "type": "object", + "properties": { + "team_id": { + "type": "integer", + "description": "ID of the team" + }, + "user_ids": { + "type": "array", + "items": {"type": "integer"}, + "description": "Array of user IDs to remove from the team" + } + }, + "required": ["team_id", "user_ids"] + } + ), + types.Tool( + name="delete_team", + description="Delete a team", + inputSchema={ + "type": "object", + "properties": { + "team_id": { + "type": "integer", + "description": "ID of the team to delete" + } + }, + "required": ["team_id"] + } + ), + + # RBAC tools + types.Tool( + name="get_roles", + description="Get available roles", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="get_resource_controls", + description="Get resource access controls", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="create_resource_control", + description="Create resource access control", + inputSchema={ + "type": "object", + "properties": { + "resource_id": { + "type": "string", + "description": "ID of the resource" + }, + "resource_type": { + "type": "string", + "description": "Type of resource (container, service, volume, etc.)" + }, + "public": { + "type": "boolean", + "description": "Whether the resource is public", + "default": False + }, + "administrators_only": { + "type": "boolean", + "description": "Whether only administrators can access", + "default": False + }, + "users": { + "type": "array", + "items": { + "type": "object", + "properties": { + "user_id": {"type": "integer"}, + "access_level": {"type": "string", "enum": ["read", "write"]} + } + }, + "description": "User access list" + }, + "teams": { + "type": "array", + "items": { + "type": "object", + "properties": { + "team_id": {"type": "integer"}, + "access_level": {"type": "string", "enum": ["read", "write"]} + } + }, + "description": "Team access list" + } + }, + "required": ["resource_id", "resource_type"] + } + ), + + # Settings tools + types.Tool( + name="get_settings", + description="Get Portainer settings", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="update_settings", + description="Update Portainer settings", + inputSchema={ + "type": "object", + "properties": { + "allow_volume_browser": { + "type": "boolean", + "description": "Allow users to browse volumes" + }, + "allow_bind_mounts": { + "type": "boolean", + "description": "Allow regular users to use bind mounts" + }, + "allow_privileged_mode": { + "type": "boolean", + "description": "Allow regular users to use privileged mode" + }, + "allow_stack_management": { + "type": "boolean", + "description": "Allow regular users to manage stacks" + } + }, + "required": [] + } + ) + ] + + +async def make_request(method: str, endpoint: str, json_data: Optional[Dict] = None) -> Dict[str, Any]: + """Make HTTP request to Portainer API.""" + import httpx + + async with httpx.AsyncClient(verify=False) as client: + headers = {"X-API-Key": api_key} if api_key else {} + + if method == "GET": + response = await client.get(f"{portainer_url}/api{endpoint}", headers=headers) + elif method == "POST": + response = await client.post(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "PUT": + response = await client.put(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "DELETE": + response = await client.delete(f"{portainer_url}/api{endpoint}", headers=headers) + else: + raise ValueError(f"Unsupported method: {method}") + + return {"status_code": response.status_code, "data": response.json() if response.text else None, "text": response.text} + + +def convert_role(role): + """Convert between string and integer role representations.""" + if isinstance(role, int): + role_map = {1: "Administrator", 2: "StandardUser", 3: "ReadOnlyUser"} + return role_map.get(role, f"Unknown({role})") + else: + role_map = {"Administrator": 1, "StandardUser": 2, "ReadOnlyUser": 3} + return role_map.get(role, 2) + + +@server.call_tool() +async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]: + """Handle tool calls.""" + + try: + # Core tools + if name == "test_connection": + result = await make_request("GET", "/status") + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"✓ Connected to Portainer at {portainer_url}")] + else: + return [types.TextContent(type="text", text=f"✗ Failed to connect: HTTP {result['status_code']}")] + + # User Management + elif name == "get_users": + result = await make_request("GET", "/users") + if result["status_code"] == 200: + users = result["data"] + output = f"Found {len(users)} users:\n" + for user in users: + role = convert_role(user.get("Role", "Unknown")) + output += f"- ID: {user.get('Id')}, Username: {user.get('Username')}, Role: {role}\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get users: HTTP {result['status_code']}")] + + elif name == "create_user": + username = arguments.get("username") + password = arguments.get("password") + role = arguments.get("role", "StandardUser") + + if not username or not password: + return [types.TextContent(type="text", text="Error: Username and password are required")] + + role_int = convert_role(role) + + # Try with integer role first + result = await make_request("POST", "/users", { + "Username": username, + "Password": password, + "Role": role_int + }) + + if result["status_code"] == 409: + return [types.TextContent(type="text", text=f"User '{username}' already exists")] + elif result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ User '{username}' created successfully with role {role}")] + else: + # Try with string role + result = await make_request("POST", "/users", { + "Username": username, + "Password": password, + "Role": role + }) + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ User '{username}' created successfully with role {role}")] + else: + return [types.TextContent(type="text", text=f"Failed to create user: HTTP {result['status_code']} - {result['text']}")] + + elif name == "update_user": + user_id = arguments.get("user_id") + if not user_id: + return [types.TextContent(type="text", text="Error: user_id is required")] + + update_data = {} + if "password" in arguments: + update_data["Password"] = arguments["password"] + if "role" in arguments: + update_data["Role"] = convert_role(arguments["role"]) + + result = await make_request("PUT", f"/users/{user_id}", update_data) + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"✓ User {user_id} updated successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to update user: HTTP {result['status_code']}")] + + elif name == "delete_user": + user_id = arguments.get("user_id") + if not user_id: + return [types.TextContent(type="text", text="Error: user_id is required")] + + result = await make_request("DELETE", f"/users/{user_id}") + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ User {user_id} deleted successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to delete user: HTTP {result['status_code']}")] + + # Teams Management + elif name == "get_teams": + result = await make_request("GET", "/teams") + if result["status_code"] == 200: + teams = result["data"] + output = f"Found {len(teams)} teams:\n" + for team in teams: + output += f"- ID: {team.get('Id')}, Name: {team.get('Name')}\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get teams: HTTP {result['status_code']}")] + + elif name == "create_team": + team_name = arguments.get("name") + if not team_name: + return [types.TextContent(type="text", text="Error: Team name is required")] + + team_data = { + "Name": team_name, + "Leaders": arguments.get("leaders", []) + } + + result = await make_request("POST", "/teams", team_data) + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Team '{team_name}' created successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to create team: HTTP {result['status_code']} - {result['text']}")] + + elif name == "add_team_members": + team_id = arguments.get("team_id") + user_ids = arguments.get("user_ids", []) + + if not team_id: + return [types.TextContent(type="text", text="Error: team_id is required")] + if not user_ids: + return [types.TextContent(type="text", text="Error: user_ids array is required")] + + result = await make_request("POST", f"/teams/{team_id}/memberships", {"UserIds": user_ids}) + if result["status_code"] in [200, 201, 204]: + return [types.TextContent(type="text", text=f"✓ Added {len(user_ids)} users to team {team_id}")] + else: + return [types.TextContent(type="text", text=f"Failed to add team members: HTTP {result['status_code']}")] + + elif name == "remove_team_members": + team_id = arguments.get("team_id") + user_ids = arguments.get("user_ids", []) + + if not team_id: + return [types.TextContent(type="text", text="Error: team_id is required")] + if not user_ids: + return [types.TextContent(type="text", text="Error: user_ids array is required")] + + result = await make_request("DELETE", f"/teams/{team_id}/memberships", {"UserIds": user_ids}) + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Removed {len(user_ids)} users from team {team_id}")] + else: + return [types.TextContent(type="text", text=f"Failed to remove team members: HTTP {result['status_code']}")] + + elif name == "delete_team": + team_id = arguments.get("team_id") + if not team_id: + return [types.TextContent(type="text", text="Error: team_id is required")] + + result = await make_request("DELETE", f"/teams/{team_id}") + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Team {team_id} deleted successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to delete team: HTTP {result['status_code']}")] + + # RBAC tools + elif name == "get_roles": + result = await make_request("GET", "/roles") + if result["status_code"] == 200: + roles = result["data"] + output = "Available roles:\n" + for role in roles: + output += f"- ID: {role.get('Id')}, Name: {role.get('Name')}, Priority: {role.get('Priority')}\n" + if role.get('Description'): + output += f" Description: {role.get('Description')}\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get roles: HTTP {result['status_code']}")] + + elif name == "get_resource_controls": + result = await make_request("GET", "/resource_controls") + if result["status_code"] == 200: + controls = result["data"] + output = f"Found {len(controls)} resource controls:\n" + for control in controls: + output += f"- ID: {control.get('Id')}, Resource: {control.get('ResourceId')}, " + output += f"Type: {control.get('Type')}, Public: {control.get('Public')}\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get resource controls: HTTP {result['status_code']}")] + + elif name == "create_resource_control": + resource_id = arguments.get("resource_id") + resource_type = arguments.get("resource_type") + + if not resource_id or not resource_type: + return [types.TextContent(type="text", text="Error: resource_id and resource_type are required")] + + control_data = { + "ResourceId": resource_id, + "Type": resource_type, + "Public": arguments.get("public", False), + "AdministratorsOnly": arguments.get("administrators_only", False), + "UserAccesses": arguments.get("users", []), + "TeamAccesses": arguments.get("teams", []) + } + + result = await make_request("POST", "/resource_controls", control_data) + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Resource control created for {resource_id}")] + else: + return [types.TextContent(type="text", text=f"Failed to create resource control: HTTP {result['status_code']}")] + + # Settings tools + elif name == "get_settings": + result = await make_request("GET", "/settings") + if result["status_code"] == 200: + settings = result["data"] + output = "Portainer Settings:\n" + output += f"- Allow Volume Browser: {settings.get('AllowVolumeBrowser', False)}\n" + output += f"- Allow Bind Mounts: {settings.get('AllowBindMountsForRegularUsers', False)}\n" + output += f"- Allow Privileged Mode: {settings.get('AllowPrivilegedModeForRegularUsers', False)}\n" + output += f"- Allow Stack Management: {settings.get('AllowStackManagementForRegularUsers', False)}\n" + output += f"- Authentication Method: {settings.get('AuthenticationMethod', 'Unknown')}\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get settings: HTTP {result['status_code']}")] + + elif name == "update_settings": + # Get current settings first + current = await make_request("GET", "/settings") + if current["status_code"] != 200: + return [types.TextContent(type="text", text=f"Failed to get current settings: HTTP {current['status_code']}")] + + settings_data = current["data"] + + # Update only provided fields + if "allow_volume_browser" in arguments: + settings_data["AllowVolumeBrowser"] = arguments["allow_volume_browser"] + if "allow_bind_mounts" in arguments: + settings_data["AllowBindMountsForRegularUsers"] = arguments["allow_bind_mounts"] + if "allow_privileged_mode" in arguments: + settings_data["AllowPrivilegedModeForRegularUsers"] = arguments["allow_privileged_mode"] + if "allow_stack_management" in arguments: + settings_data["AllowStackManagementForRegularUsers"] = arguments["allow_stack_management"] + + result = await make_request("PUT", "/settings", settings_data) + if result["status_code"] == 200: + return [types.TextContent(type="text", text="✓ Settings updated successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to update settings: HTTP {result['status_code']}")] + + else: + return [types.TextContent(type="text", text=f"Unknown tool: {name}")] + + except Exception as e: + return [types.TextContent(type="text", text=f"Error: {str(e)}")] + + +async def run(): + """Run the MCP server.""" + # Use stdio transport + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="portainer-unified", + server_version="1.0.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions( + prompts_changed=False, + resources_changed=False, + tools_changed=False, + ), + experimental_capabilities={}, + ), + ), + ) + + +def main(): + """Main entry point.""" + asyncio.run(run()) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/portainer_docker_server.py b/portainer_docker_server.py new file mode 100755 index 0000000..dc62eb0 --- /dev/null +++ b/portainer_docker_server.py @@ -0,0 +1,1431 @@ +#!/usr/bin/env python3 +"""MCP server for Portainer Docker and Docker Swarm management.""" + +import os +import sys +import json +import asyncio +from typing import Any, Dict, List, Optional +from enum import Enum + +# Suppress all logging to stderr +os.environ["MCP_MODE"] = "true" + +import mcp.server.stdio +import mcp.types as types +from mcp.server import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +# Create server +server = Server("portainer-docker") + +# Store for our state +portainer_url = os.getenv("PORTAINER_URL", "https://partner.portainer.live") +api_key = os.getenv("PORTAINER_API_KEY", "") + + +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + # Container Management + types.Tool( + name="list_containers", + description="List containers in an environment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "all": { + "type": "boolean", + "description": "Show all containers (default shows only running)", + "default": False + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="inspect_container", + description="Get detailed information about a container", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "container_id": { + "type": "string", + "description": "Container ID or name" + } + }, + "required": ["environment_id", "container_id"] + } + ), + types.Tool( + name="create_container", + description="Create a new container", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "image": { + "type": "string", + "description": "Docker image to use" + }, + "name": { + "type": "string", + "description": "Container name (optional)" + }, + "command": { + "type": "array", + "items": {"type": "string"}, + "description": "Command to run (optional)" + }, + "env": { + "type": "array", + "items": {"type": "string"}, + "description": "Environment variables in KEY=VALUE format (optional)" + }, + "ports": { + "type": "object", + "description": "Port bindings (e.g., {'80/tcp': [{'HostPort': '8080'}]})" + }, + "volumes": { + "type": "object", + "description": "Volume bindings (e.g., {'/host/path': {'bind': '/container/path', 'mode': 'rw'}})" + }, + "restart_policy": { + "type": "string", + "enum": ["no", "always", "unless-stopped", "on-failure"], + "default": "no" + } + }, + "required": ["environment_id", "image"] + } + ), + types.Tool( + name="start_container", + description="Start a stopped container", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "container_id": { + "type": "string", + "description": "Container ID or name" + } + }, + "required": ["environment_id", "container_id"] + } + ), + types.Tool( + name="stop_container", + description="Stop a running container", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "container_id": { + "type": "string", + "description": "Container ID or name" + } + }, + "required": ["environment_id", "container_id"] + } + ), + types.Tool( + name="restart_container", + description="Restart a container", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "container_id": { + "type": "string", + "description": "Container ID or name" + } + }, + "required": ["environment_id", "container_id"] + } + ), + types.Tool( + name="remove_container", + description="Remove a container", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "container_id": { + "type": "string", + "description": "Container ID or name" + }, + "force": { + "type": "boolean", + "description": "Force removal of running container", + "default": False + } + }, + "required": ["environment_id", "container_id"] + } + ), + types.Tool( + name="get_container_logs", + description="Get logs from a container", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "container_id": { + "type": "string", + "description": "Container ID or name" + }, + "tail": { + "type": "integer", + "description": "Number of lines to show from the end", + "default": 100 + }, + "timestamps": { + "type": "boolean", + "description": "Show timestamps", + "default": False + } + }, + "required": ["environment_id", "container_id"] + } + ), + # Image Management + types.Tool( + name="list_images", + description="List Docker images in an environment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="pull_image", + description="Pull a Docker image", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "image": { + "type": "string", + "description": "Image name with optional tag (e.g., nginx:latest)" + } + }, + "required": ["environment_id", "image"] + } + ), + types.Tool( + name="remove_image", + description="Remove a Docker image", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "image_id": { + "type": "string", + "description": "Image ID or name:tag" + }, + "force": { + "type": "boolean", + "description": "Force removal", + "default": False + } + }, + "required": ["environment_id", "image_id"] + } + ), + # Volume Management + types.Tool( + name="list_volumes", + description="List Docker volumes", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_volume", + description="Create a Docker volume", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "name": { + "type": "string", + "description": "Volume name" + }, + "driver": { + "type": "string", + "description": "Volume driver", + "default": "local" + } + }, + "required": ["environment_id", "name"] + } + ), + types.Tool( + name="remove_volume", + description="Remove a Docker volume", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "volume_name": { + "type": "string", + "description": "Volume name" + } + }, + "required": ["environment_id", "volume_name"] + } + ), + # Network Management + types.Tool( + name="list_networks", + description="List Docker networks", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_network", + description="Create a Docker network", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "name": { + "type": "string", + "description": "Network name" + }, + "driver": { + "type": "string", + "description": "Network driver", + "default": "bridge" + }, + "internal": { + "type": "boolean", + "description": "Restrict external access", + "default": False + } + }, + "required": ["environment_id", "name"] + } + ), + types.Tool( + name="remove_network", + description="Remove a Docker network", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "network_id": { + "type": "string", + "description": "Network ID or name" + } + }, + "required": ["environment_id", "network_id"] + } + ), + # Docker Swarm Services + types.Tool( + name="list_services", + description="List Docker Swarm services", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Swarm environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_service", + description="Create a Docker Swarm service", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Swarm environment" + }, + "name": { + "type": "string", + "description": "Service name" + }, + "image": { + "type": "string", + "description": "Docker image" + }, + "replicas": { + "type": "integer", + "description": "Number of replicas", + "default": 1 + }, + "command": { + "type": "array", + "items": {"type": "string"}, + "description": "Command to run (optional)" + }, + "env": { + "type": "array", + "items": {"type": "string"}, + "description": "Environment variables in KEY=VALUE format" + }, + "ports": { + "type": "array", + "items": { + "type": "object", + "properties": { + "target": {"type": "integer"}, + "published": {"type": "integer"}, + "protocol": {"type": "string", "enum": ["tcp", "udp"]} + } + }, + "description": "Published ports" + } + }, + "required": ["environment_id", "name", "image"] + } + ), + types.Tool( + name="update_service", + description="Update a Docker Swarm service", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Swarm environment" + }, + "service_id": { + "type": "string", + "description": "Service ID or name" + }, + "image": { + "type": "string", + "description": "New image (optional)" + }, + "replicas": { + "type": "integer", + "description": "New replica count (optional)" + }, + "env": { + "type": "array", + "items": {"type": "string"}, + "description": "New environment variables (optional)" + } + }, + "required": ["environment_id", "service_id"] + } + ), + types.Tool( + name="remove_service", + description="Remove a Docker Swarm service", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Swarm environment" + }, + "service_id": { + "type": "string", + "description": "Service ID or name" + } + }, + "required": ["environment_id", "service_id"] + } + ), + types.Tool( + name="get_service_logs", + description="Get logs from a Docker Swarm service", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Swarm environment" + }, + "service_id": { + "type": "string", + "description": "Service ID or name" + }, + "tail": { + "type": "integer", + "description": "Number of lines from the end", + "default": 100 + } + }, + "required": ["environment_id", "service_id"] + } + ), + # Stack Management + types.Tool( + name="list_stacks", + description="List Docker stacks", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="deploy_stack", + description="Deploy a Docker stack from compose file", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "name": { + "type": "string", + "description": "Stack name" + }, + "compose_file": { + "type": "string", + "description": "Docker Compose file content (YAML)" + }, + "env": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "value": {"type": "string"} + } + }, + "description": "Environment variables for the stack" + } + }, + "required": ["environment_id", "name", "compose_file"] + } + ), + types.Tool( + name="remove_stack", + description="Remove a Docker stack", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "stack_id": { + "type": "integer", + "description": "Stack ID" + } + }, + "required": ["environment_id", "stack_id"] + } + ), + # System Information + types.Tool( + name="get_docker_info", + description="Get Docker system information", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="get_docker_version", + description="Get Docker version information", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + } + }, + "required": ["environment_id"] + } + ) + ] + + +async def make_request(method: str, endpoint: str, json_data: Optional[Dict] = None, + params: Optional[Dict] = None, text_response: bool = False) -> Dict[str, Any]: + """Make HTTP request to Portainer API.""" + import httpx + + async with httpx.AsyncClient(verify=False, timeout=30.0) as client: + headers = {"X-API-Key": api_key} if api_key else {} + + # Add JSON content type for POST/PUT requests + if method in ["POST", "PUT"] and json_data is not None: + headers["Content-Type"] = "application/json" + + if method == "GET": + response = await client.get(f"{portainer_url}/api{endpoint}", headers=headers, params=params) + elif method == "POST": + response = await client.post(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "PUT": + response = await client.put(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "DELETE": + response = await client.delete(f"{portainer_url}/api{endpoint}", headers=headers) + else: + raise ValueError(f"Unsupported method: {method}") + + # Handle text responses (like logs) + if text_response: + return {"status_code": response.status_code, "data": None, "text": response.text} + + # Parse JSON response safely + try: + data = response.json() if response.text and response.headers.get("content-type", "").startswith("application/json") else None + except Exception: + data = None + + return {"status_code": response.status_code, "data": data, "text": response.text} + + +def format_container_state(state: str) -> str: + """Format container state for display.""" + state_map = { + "running": "🟢 Running", + "paused": "⏸️ Paused", + "exited": "🔴 Stopped", + "dead": "💀 Dead", + "created": "🆕 Created", + "restarting": "🔄 Restarting", + "removing": "🗑️ Removing" + } + return state_map.get(state.lower(), state) + + +def format_bytes(bytes_val: int) -> str: + """Format bytes to human readable format.""" + for unit in ['B', 'KB', 'MB', 'GB', 'TB']: + if bytes_val < 1024.0: + return f"{bytes_val:.1f} {unit}" + bytes_val /= 1024.0 + return f"{bytes_val:.1f} PB" + + +@server.call_tool() +async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]: + """Handle tool calls.""" + import httpx + + try: + env_id = arguments.get("environment_id") if arguments else None + + # Container Management + if name == "list_containers": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + all_containers = arguments.get("all", False) + params = {"all": "true"} if all_containers else {"all": "false"} + + result = await make_request("GET", f"/endpoints/{env_id}/docker/containers/json", params=params) + + if result["status_code"] == 200 and result["data"] is not None: + containers = result["data"] + if not containers: + return [types.TextContent(type="text", text="No containers found")] + + output = f"Found {len(containers)} container(s):\n" + for container in containers[:20]: # Limit to 20 containers + # Container names come with leading slash + names = ", ".join([n.lstrip("/") for n in container.get("Names", [])]) + state = container.get("State", "unknown") + status = container.get("Status", "") + image = container.get("Image", "unknown") + container_id = container.get("Id", "")[:12] # Short ID + + output += f"\n- {names} ({container_id})" + output += f"\n Image: {image}" + output += f"\n Status: {format_container_state(state)} - {status}" + + # Show ports if any + ports = container.get("Ports", []) + if ports: + port_info = [] + for port in ports: + if port.get("PublicPort"): + port_info.append(f"{port.get('PublicPort')}→{port.get('PrivatePort')}/{port.get('Type')}") + else: + port_info.append(f"{port.get('PrivatePort')}/{port.get('Type')}") + output += f"\n Ports: {', '.join(port_info)}" + + if len(containers) > 20: + output += f"\n\n... and {len(containers) - 20} more containers" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list containers: HTTP {result['status_code']}")] + + elif name == "inspect_container": + container_id = arguments.get("container_id") + if not env_id or not container_id: + return [types.TextContent(type="text", text="Error: environment_id and container_id are required")] + + result = await make_request("GET", f"/endpoints/{env_id}/docker/containers/{container_id}/json") + + if result["status_code"] == 200 and result["data"]: + info = result["data"] + output = f"Container Details:\n" + output += f"- ID: {info.get('Id', '')[:12]}\n" + output += f"- Name: {info.get('Name', '').lstrip('/')}\n" + output += f"- Image: {info.get('Config', {}).get('Image', '')}\n" + output += f"- Status: {format_container_state(info.get('State', {}).get('Status', ''))}\n" + output += f"- Created: {info.get('Created', '')}\n" + + # State details + state = info.get('State', {}) + if state.get('Running'): + output += f"- PID: {state.get('Pid')}\n" + output += f"- Started: {state.get('StartedAt')}\n" + + # Network info + networks = info.get('NetworkSettings', {}).get('Networks', {}) + if networks: + output += "\nNetworks:\n" + for net_name, net_info in networks.items(): + output += f"- {net_name}: {net_info.get('IPAddress', 'N/A')}\n" + + # Mounts + mounts = info.get('Mounts', []) + if mounts: + output += "\nMounts:\n" + for mount in mounts: + output += f"- {mount.get('Source')} → {mount.get('Destination')} ({mount.get('Mode', 'rw')})\n" + + # Environment variables + env_vars = info.get('Config', {}).get('Env', []) + if env_vars: + output += "\nEnvironment Variables:\n" + for env in env_vars[:10]: # Limit to 10 + if '=' in env and not any(secret in env.lower() for secret in ['password', 'secret', 'key', 'token']): + output += f"- {env}\n" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to inspect container: HTTP {result['status_code']}")] + + elif name == "create_container": + image = arguments.get("image") + if not env_id or not image: + return [types.TextContent(type="text", text="Error: environment_id and image are required")] + + # Build container configuration + config = { + "Image": image, + "Hostname": arguments.get("name", ""), + "AttachStdin": False, + "AttachStdout": True, + "AttachStderr": True, + "Tty": False, + "OpenStdin": False + } + + if arguments.get("command"): + config["Cmd"] = arguments["command"] + + if arguments.get("env"): + config["Env"] = arguments["env"] + + # Host configuration + host_config = { + "RestartPolicy": {"Name": arguments.get("restart_policy", "no")} + } + + if arguments.get("ports"): + config["ExposedPorts"] = {port: {} for port in arguments["ports"].keys()} + host_config["PortBindings"] = arguments["ports"] + + if arguments.get("volumes"): + host_config["Binds"] = [] + for host_path, container_config in arguments["volumes"].items(): + bind = f"{host_path}:{container_config['bind']}" + if container_config.get('mode'): + bind += f":{container_config['mode']}" + host_config["Binds"].append(bind) + + create_data = { + **config, + "HostConfig": host_config + } + + # Create container + params = {"name": arguments.get("name")} if arguments.get("name") else {} + result = await make_request("POST", f"/endpoints/{env_id}/docker/containers/create", + json_data=create_data, params=params) + + if result["status_code"] in [200, 201] and result["data"]: + container_id = result["data"].get("Id", "")[:12] + output = f"✓ Container created: {container_id}" + + # Auto-start the container + start_result = await make_request("POST", f"/endpoints/{env_id}/docker/containers/{container_id}/start") + if start_result["status_code"] in [200, 204]: + output += "\n✓ Container started successfully" + else: + output += f"\n⚠️ Container created but failed to start: HTTP {start_result['status_code']}" + + return [types.TextContent(type="text", text=output)] + else: + error_msg = f"Failed to create container: HTTP {result['status_code']}" + if result.get("text"): + error_msg += f"\n{result['text']}" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "start_container": + container_id = arguments.get("container_id") + if not env_id or not container_id: + return [types.TextContent(type="text", text="Error: environment_id and container_id are required")] + + result = await make_request("POST", f"/endpoints/{env_id}/docker/containers/{container_id}/start") + + if result["status_code"] in [200, 204, 304]: # 304 = already started + return [types.TextContent(type="text", text=f"✓ Container {container_id} started")] + else: + return [types.TextContent(type="text", text=f"Failed to start container: HTTP {result['status_code']}")] + + elif name == "stop_container": + container_id = arguments.get("container_id") + if not env_id or not container_id: + return [types.TextContent(type="text", text="Error: environment_id and container_id are required")] + + result = await make_request("POST", f"/endpoints/{env_id}/docker/containers/{container_id}/stop") + + if result["status_code"] in [200, 204, 304]: # 304 = already stopped + return [types.TextContent(type="text", text=f"✓ Container {container_id} stopped")] + else: + return [types.TextContent(type="text", text=f"Failed to stop container: HTTP {result['status_code']}")] + + elif name == "restart_container": + container_id = arguments.get("container_id") + if not env_id or not container_id: + return [types.TextContent(type="text", text="Error: environment_id and container_id are required")] + + result = await make_request("POST", f"/endpoints/{env_id}/docker/containers/{container_id}/restart") + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Container {container_id} restarted")] + else: + return [types.TextContent(type="text", text=f"Failed to restart container: HTTP {result['status_code']}")] + + elif name == "remove_container": + container_id = arguments.get("container_id") + if not env_id or not container_id: + return [types.TextContent(type="text", text="Error: environment_id and container_id are required")] + + params = {"force": "true"} if arguments.get("force") else {} + result = await make_request("DELETE", f"/endpoints/{env_id}/docker/containers/{container_id}", params=params) + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Container {container_id} removed")] + else: + return [types.TextContent(type="text", text=f"Failed to remove container: HTTP {result['status_code']}")] + + elif name == "get_container_logs": + container_id = arguments.get("container_id") + if not env_id or not container_id: + return [types.TextContent(type="text", text="Error: environment_id and container_id are required")] + + params = { + "stdout": "true", + "stderr": "true", + "tail": str(arguments.get("tail", 100)) + } + if arguments.get("timestamps"): + params["timestamps"] = "true" + + result = await make_request("GET", f"/endpoints/{env_id}/docker/containers/{container_id}/logs", + params=params, text_response=True) + + if result["status_code"] == 200: + logs = result.get("text", "") + if logs: + # Docker logs have special encoding, try to clean them up + import re + # Remove ANSI escape codes and Docker log headers + logs = re.sub(r'\x1b\[[0-9;]*m', '', logs) + logs = re.sub(r'[\x00-\x08\x0e-\x1f]', '', logs) + + output = f"Container logs (last {arguments.get('tail', 100)} lines):\n" + output += "-" * 50 + "\n" + output += logs + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text="No logs available")] + else: + return [types.TextContent(type="text", text=f"Failed to get logs: HTTP {result['status_code']}")] + + # Image Management + elif name == "list_images": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/docker/images/json") + + if result["status_code"] == 200 and result["data"]: + images = result["data"] + output = f"Found {len(images)} image(s):\n" + + for image in images[:20]: # Limit to 20 + tags = image.get("RepoTags", []) + if tags and tags[0] != ":": + image_name = tags[0] + else: + image_name = image.get("Id", "")[:12] + + size = format_bytes(image.get("Size", 0)) + created = image.get("Created", 0) + + output += f"\n- {image_name}" + output += f"\n ID: {image.get('Id', '')[:12]}" + output += f"\n Size: {size}" + + # Additional tags + if len(tags) > 1: + output += f"\n Other tags: {', '.join(tags[1:])}" + + if len(images) > 20: + output += f"\n\n... and {len(images) - 20} more images" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list images: HTTP {result['status_code']}")] + + elif name == "pull_image": + image = arguments.get("image") + if not env_id or not image: + return [types.TextContent(type="text", text="Error: environment_id and image are required")] + + # Parse image name and tag + if ":" in image: + image_name, tag = image.rsplit(":", 1) + else: + image_name = image + tag = "latest" + + params = {"fromImage": image_name, "tag": tag} + result = await make_request("POST", f"/endpoints/{env_id}/docker/images/create", params=params) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"✓ Image {image} pulled successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to pull image: HTTP {result['status_code']}")] + + elif name == "remove_image": + image_id = arguments.get("image_id") + if not env_id or not image_id: + return [types.TextContent(type="text", text="Error: environment_id and image_id are required")] + + params = {"force": "true"} if arguments.get("force") else {} + result = await make_request("DELETE", f"/endpoints/{env_id}/docker/images/{image_id}", params=params) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"✓ Image {image_id} removed")] + else: + return [types.TextContent(type="text", text=f"Failed to remove image: HTTP {result['status_code']}")] + + # Volume Management + elif name == "list_volumes": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/docker/volumes") + + if result["status_code"] == 200 and result["data"]: + volumes = result["data"].get("Volumes", []) + output = f"Found {len(volumes)} volume(s):\n" + + for volume in volumes[:20]: + output += f"\n- {volume.get('Name')}" + output += f"\n Driver: {volume.get('Driver')}" + output += f"\n Mountpoint: {volume.get('Mountpoint')}" + if volume.get('Options'): + output += f"\n Options: {volume.get('Options')}" + + if len(volumes) > 20: + output += f"\n\n... and {len(volumes) - 20} more volumes" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list volumes: HTTP {result['status_code']}")] + + elif name == "create_volume": + volume_name = arguments.get("name") + if not env_id or not volume_name: + return [types.TextContent(type="text", text="Error: environment_id and name are required")] + + volume_data = { + "Name": volume_name, + "Driver": arguments.get("driver", "local") + } + + result = await make_request("POST", f"/endpoints/{env_id}/docker/volumes/create", json_data=volume_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Volume '{volume_name}' created")] + else: + return [types.TextContent(type="text", text=f"Failed to create volume: HTTP {result['status_code']}")] + + elif name == "remove_volume": + volume_name = arguments.get("volume_name") + if not env_id or not volume_name: + return [types.TextContent(type="text", text="Error: environment_id and volume_name are required")] + + result = await make_request("DELETE", f"/endpoints/{env_id}/docker/volumes/{volume_name}") + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Volume '{volume_name}' removed")] + else: + return [types.TextContent(type="text", text=f"Failed to remove volume: HTTP {result['status_code']}")] + + # Network Management + elif name == "list_networks": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/docker/networks") + + if result["status_code"] == 200 and result["data"]: + networks = result["data"] + output = f"Found {len(networks)} network(s):\n" + + for network in networks: + output += f"\n- {network.get('Name')} ({network.get('Id', '')[:12]})" + output += f"\n Driver: {network.get('Driver')}" + output += f"\n Scope: {network.get('Scope')}" + if network.get('Internal'): + output += f"\n Internal: Yes" + if network.get('Containers'): + output += f"\n Containers: {len(network['Containers'])}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list networks: HTTP {result['status_code']}")] + + elif name == "create_network": + network_name = arguments.get("name") + if not env_id or not network_name: + return [types.TextContent(type="text", text="Error: environment_id and name are required")] + + network_data = { + "Name": network_name, + "Driver": arguments.get("driver", "bridge"), + "Internal": arguments.get("internal", False) + } + + result = await make_request("POST", f"/endpoints/{env_id}/docker/networks/create", json_data=network_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Network '{network_name}' created")] + else: + return [types.TextContent(type="text", text=f"Failed to create network: HTTP {result['status_code']}")] + + elif name == "remove_network": + network_id = arguments.get("network_id") + if not env_id or not network_id: + return [types.TextContent(type="text", text="Error: environment_id and network_id are required")] + + result = await make_request("DELETE", f"/endpoints/{env_id}/docker/networks/{network_id}") + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Network '{network_id}' removed")] + else: + return [types.TextContent(type="text", text=f"Failed to remove network: HTTP {result['status_code']}")] + + # Docker Swarm Services + elif name == "list_services": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/docker/services") + + if result["status_code"] == 200 and result["data"]: + services = result["data"] + output = f"Found {len(services)} service(s):\n" + + for service in services[:20]: + spec = service.get("Spec", {}) + mode = spec.get("Mode", {}) + + # Get replica info + if "Replicated" in mode: + replicas = mode["Replicated"].get("Replicas", 0) + mode_str = f"replicated ({replicas} replicas)" + else: + mode_str = "global" + + output += f"\n- {spec.get('Name')} ({service.get('ID', '')[:12]})" + output += f"\n Image: {spec.get('TaskTemplate', {}).get('ContainerSpec', {}).get('Image', 'unknown')}" + output += f"\n Mode: {mode_str}" + + # Get current state + if service.get("UpdateStatus"): + status = service["UpdateStatus"].get("State", "") + if status: + output += f"\n Update Status: {status}" + + if len(services) > 20: + output += f"\n\n... and {len(services) - 20} more services" + + return [types.TextContent(type="text", text=output)] + else: + error_msg = f"Failed to list services: HTTP {result['status_code']}" + if result['status_code'] == 503: + error_msg += "\nNote: This environment might not be a Docker Swarm cluster" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "create_service": + service_name = arguments.get("name") + image = arguments.get("image") + if not env_id or not service_name or not image: + return [types.TextContent(type="text", text="Error: environment_id, name, and image are required")] + + service_spec = { + "Name": service_name, + "TaskTemplate": { + "ContainerSpec": { + "Image": image + } + }, + "Mode": { + "Replicated": { + "Replicas": arguments.get("replicas", 1) + } + } + } + + # Add command if provided + if arguments.get("command"): + service_spec["TaskTemplate"]["ContainerSpec"]["Command"] = arguments["command"] + + # Add environment variables + if arguments.get("env"): + service_spec["TaskTemplate"]["ContainerSpec"]["Env"] = arguments["env"] + + # Add ports + if arguments.get("ports"): + service_spec["EndpointSpec"] = { + "Ports": arguments["ports"] + } + + result = await make_request("POST", f"/endpoints/{env_id}/docker/services/create", json_data=service_spec) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Service '{service_name}' created")] + else: + error_msg = f"Failed to create service: HTTP {result['status_code']}" + if result.get("text"): + error_msg += f"\n{result['text']}" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "update_service": + service_id = arguments.get("service_id") + if not env_id or not service_id: + return [types.TextContent(type="text", text="Error: environment_id and service_id are required")] + + # Get current service spec + result = await make_request("GET", f"/endpoints/{env_id}/docker/services/{service_id}") + if result["status_code"] != 200: + return [types.TextContent(type="text", text=f"Failed to get service: HTTP {result['status_code']}")] + + service = result["data"] + spec = service.get("Spec", {}) + version = service.get("Version", {}).get("Index") + + # Update fields if provided + if arguments.get("image"): + spec["TaskTemplate"]["ContainerSpec"]["Image"] = arguments["image"] + + if arguments.get("replicas") is not None: + if "Replicated" in spec.get("Mode", {}): + spec["Mode"]["Replicated"]["Replicas"] = arguments["replicas"] + + if arguments.get("env"): + spec["TaskTemplate"]["ContainerSpec"]["Env"] = arguments["env"] + + # Update service + result = await make_request("POST", f"/endpoints/{env_id}/docker/services/{service_id}/update", + json_data=spec, params={"version": version}) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"✓ Service '{service_id}' updated")] + else: + return [types.TextContent(type="text", text=f"Failed to update service: HTTP {result['status_code']}")] + + elif name == "remove_service": + service_id = arguments.get("service_id") + if not env_id or not service_id: + return [types.TextContent(type="text", text="Error: environment_id and service_id are required")] + + result = await make_request("DELETE", f"/endpoints/{env_id}/docker/services/{service_id}") + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Service '{service_id}' removed")] + else: + return [types.TextContent(type="text", text=f"Failed to remove service: HTTP {result['status_code']}")] + + elif name == "get_service_logs": + service_id = arguments.get("service_id") + if not env_id or not service_id: + return [types.TextContent(type="text", text="Error: environment_id and service_id are required")] + + params = { + "stdout": "true", + "stderr": "true", + "tail": str(arguments.get("tail", 100)) + } + + result = await make_request("GET", f"/endpoints/{env_id}/docker/services/{service_id}/logs", + params=params, text_response=True) + + if result["status_code"] == 200: + logs = result.get("text", "") + if logs: + output = f"Service logs (last {arguments.get('tail', 100)} lines):\n" + output += "-" * 50 + "\n" + output += logs + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text="No logs available")] + else: + return [types.TextContent(type="text", text=f"Failed to get logs: HTTP {result['status_code']}")] + + # Stack Management + elif name == "list_stacks": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", "/stacks") + + if result["status_code"] == 200 and result["data"]: + stacks = result["data"] + # Filter stacks for this environment + env_stacks = [s for s in stacks if s.get("EndpointId") == env_id] + + output = f"Found {len(env_stacks)} stack(s) in this environment:\n" + + for stack in env_stacks: + output += f"\n- {stack.get('Name')} (ID: {stack.get('Id')})" + output += f"\n Type: {stack.get('Type', 'unknown')}" + output += f"\n Status: {stack.get('Status', 'unknown')}" + if stack.get("GitConfig"): + output += f"\n Git Repository: Yes" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list stacks: HTTP {result['status_code']}")] + + elif name == "deploy_stack": + stack_name = arguments.get("name") + compose_file = arguments.get("compose_file") + if not env_id or not stack_name or not compose_file: + return [types.TextContent(type="text", text="Error: environment_id, name, and compose_file are required")] + + stack_data = { + "Name": stack_name, + "StackFileContent": compose_file, + "Env": arguments.get("env", []) + } + + params = { + "type": 2, # Compose stack + "method": "string", # Stack file provided as string + "endpointId": env_id + } + + result = await make_request("POST", "/stacks", json_data=stack_data, params=params) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Stack '{stack_name}' deployed successfully")] + else: + error_msg = f"Failed to deploy stack: HTTP {result['status_code']}" + if result.get("text"): + error_msg += f"\n{result['text']}" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "remove_stack": + stack_id = arguments.get("stack_id") + if not env_id or not stack_id: + return [types.TextContent(type="text", text="Error: environment_id and stack_id are required")] + + result = await make_request("DELETE", f"/stacks/{stack_id}") + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Stack removed successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to remove stack: HTTP {result['status_code']}")] + + # System Information + elif name == "get_docker_info": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/docker/info") + + if result["status_code"] == 200 and result["data"]: + info = result["data"] + output = "Docker System Information:\n" + output += f"- Docker Version: {info.get('ServerVersion', 'unknown')}\n" + output += f"- API Version: {info.get('ApiVersion', 'unknown')}\n" + output += f"- Operating System: {info.get('OperatingSystem', 'unknown')}\n" + output += f"- Architecture: {info.get('Architecture', 'unknown')}\n" + output += f"- Kernel Version: {info.get('KernelVersion', 'unknown')}\n" + output += f"- Total Memory: {format_bytes(info.get('MemTotal', 0))}\n" + output += f"- CPUs: {info.get('NCPU', 0)}\n" + output += f"- Containers: {info.get('Containers', 0)} ({info.get('ContainersRunning', 0)} running)\n" + output += f"- Images: {info.get('Images', 0)}\n" + + # Swarm info + swarm = info.get('Swarm', {}) + if swarm.get('LocalNodeState') == 'active': + output += f"\nSwarm Status:\n" + output += f"- Node ID: {swarm.get('NodeID', '')[:12]}\n" + output += f"- Is Manager: {swarm.get('ControlAvailable', False)}\n" + if swarm.get('RemoteManagers'): + output += f"- Managers: {len(swarm['RemoteManagers'])}\n" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get Docker info: HTTP {result['status_code']}")] + + elif name == "get_docker_version": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/docker/version") + + if result["status_code"] == 200 and result["data"]: + version = result["data"] + output = "Docker Version Information:\n" + output += f"- Version: {version.get('Version', 'unknown')}\n" + output += f"- API Version: {version.get('ApiVersion', 'unknown')}\n" + output += f"- Git Commit: {version.get('GitCommit', 'unknown')}\n" + output += f"- Go Version: {version.get('GoVersion', 'unknown')}\n" + output += f"- OS/Arch: {version.get('Os', 'unknown')}/{version.get('Arch', 'unknown')}\n" + output += f"- Build Time: {version.get('BuildTime', 'unknown')}\n" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get Docker version: HTTP {result['status_code']}")] + + else: + return [types.TextContent(type="text", text=f"Unknown tool: {name}")] + + except httpx.TimeoutException: + return [types.TextContent(type="text", text="Error: Request timed out. The operation may take longer than expected.")] + except httpx.ConnectError: + return [types.TextContent(type="text", text="Error: Could not connect to Portainer server. Please check the URL and network connection.")] + except Exception as e: + import traceback + error_details = f"Error: {str(e)}\nType: {type(e).__name__}" + return [types.TextContent(type="text", text=error_details)] + + +async def run(): + """Run the MCP server.""" + # Use stdio transport + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="portainer-docker", + server_version="1.0.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions( + prompts_changed=False, + resources_changed=False, + tools_changed=False, + ), + experimental_capabilities={}, + ), + ), + ) + + +def main(): + """Main entry point.""" + asyncio.run(run()) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/portainer_environments_server.py b/portainer_environments_server.py new file mode 100755 index 0000000..d63c93a --- /dev/null +++ b/portainer_environments_server.py @@ -0,0 +1,861 @@ +#!/usr/bin/env python3 +"""MCP server for Portainer Environments management.""" + +import os +import sys +import json +import asyncio +from typing import Any, Dict, List, Optional +from enum import Enum + +# Suppress all logging to stderr +os.environ["MCP_MODE"] = "true" + +import mcp.server.stdio +import mcp.types as types +from mcp.server import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +# Create server +server = Server("portainer-environments") + +# Store for our state +portainer_url = os.getenv("PORTAINER_URL", "https://partner.portainer.live") +api_key = os.getenv("PORTAINER_API_KEY", "") + + +# Environment types enum +class EnvironmentType(Enum): + DOCKER = 1 + DOCKER_SWARM = 2 + KUBERNETES = 3 + ACI = 4 + EDGE_AGENT = 5 + + +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + # Basic environment operations + types.Tool( + name="list_environments", + description="List all environments/endpoints", + inputSchema={ + "type": "object", + "properties": { + "limit": { + "type": "integer", + "description": "Number of environments to return (default: all)", + "default": 100 + }, + "start": { + "type": "integer", + "description": "Starting index for pagination", + "default": 0 + } + }, + "required": [] + } + ), + types.Tool( + name="get_environment", + description="Get details of a specific environment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_docker_environment", + description="Create a new Docker environment", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Name of the environment" + }, + "url": { + "type": "string", + "description": "Docker API URL (e.g., tcp://localhost:2375 or unix:///var/run/docker.sock)" + }, + "public_url": { + "type": "string", + "description": "Public URL for accessing the environment (optional)" + }, + "group_id": { + "type": "integer", + "description": "Environment group ID (optional)" + }, + "tags": { + "type": "array", + "items": {"type": "string"}, + "description": "Array of tags for the environment (optional)" + }, + "tls": { + "type": "boolean", + "description": "Enable TLS", + "default": False + }, + "tls_skip_verify": { + "type": "boolean", + "description": "Skip TLS certificate verification", + "default": False + } + }, + "required": ["name", "url"] + } + ), + types.Tool( + name="create_kubernetes_environment", + description="Create a new Kubernetes environment", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Name of the environment" + }, + "url": { + "type": "string", + "description": "Kubernetes API server URL" + }, + "bearer_token": { + "type": "string", + "description": "Bearer token for authentication" + }, + "group_id": { + "type": "integer", + "description": "Environment group ID (optional)" + }, + "tags": { + "type": "array", + "items": {"type": "string"}, + "description": "Array of tags for the environment (optional)" + }, + "tls_skip_verify": { + "type": "boolean", + "description": "Skip TLS certificate verification", + "default": False + } + }, + "required": ["name", "url"] + } + ), + types.Tool( + name="update_environment", + description="Update an existing environment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment to update" + }, + "name": { + "type": "string", + "description": "New name for the environment (optional)" + }, + "url": { + "type": "string", + "description": "New URL for the environment (optional)" + }, + "public_url": { + "type": "string", + "description": "New public URL (optional)" + }, + "group_id": { + "type": "integer", + "description": "New group ID (optional)" + }, + "tags": { + "type": "array", + "items": {"type": "string"}, + "description": "New tags array (optional)" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="delete_environment", + description="Delete an environment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment to delete" + } + }, + "required": ["environment_id"] + } + ), + # Environment status and management + types.Tool( + name="get_environment_status", + description="Get status and statistics of an environment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="associate_environment", + description="Associate/disassociate environment with teams", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the environment" + }, + "teams": { + "type": "array", + "items": { + "type": "object", + "properties": { + "team_id": {"type": "integer"}, + "access_level": {"type": "string", "enum": ["read", "write"]} + } + }, + "description": "Array of team associations" + } + }, + "required": ["environment_id", "teams"] + } + ), + # Environment groups + types.Tool( + name="list_environment_groups", + description="List all environment groups", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="create_environment_group", + description="Create a new environment group", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Name of the environment group" + }, + "description": { + "type": "string", + "description": "Description of the group (optional)" + } + }, + "required": ["name"] + } + ), + types.Tool( + name="update_environment_group", + description="Update an environment group", + inputSchema={ + "type": "object", + "properties": { + "group_id": { + "type": "integer", + "description": "ID of the group to update" + }, + "name": { + "type": "string", + "description": "New name for the group (optional)" + }, + "description": { + "type": "string", + "description": "New description (optional)" + } + }, + "required": ["group_id"] + } + ), + types.Tool( + name="delete_environment_group", + description="Delete an environment group", + inputSchema={ + "type": "object", + "properties": { + "group_id": { + "type": "integer", + "description": "ID of the group to delete" + } + }, + "required": ["group_id"] + } + ), + # Tags management + types.Tool( + name="list_tags", + description="List all available tags", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="create_tag", + description="Create a new tag", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Name of the tag" + } + }, + "required": ["name"] + } + ), + types.Tool( + name="delete_tag", + description="Delete a tag", + inputSchema={ + "type": "object", + "properties": { + "tag_id": { + "type": "integer", + "description": "ID of the tag to delete" + } + }, + "required": ["tag_id"] + } + ), + # Edge environments + types.Tool( + name="generate_edge_key", + description="Generate Edge agent deployment script", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Name for the Edge environment" + }, + "group_id": { + "type": "integer", + "description": "Environment group ID (optional)" + } + }, + "required": ["name"] + } + ) + ] + + +async def make_request(method: str, endpoint: str, json_data: Optional[Dict] = None) -> Dict[str, Any]: + """Make HTTP request to Portainer API.""" + import httpx + + async with httpx.AsyncClient(verify=False, timeout=30.0) as client: + headers = {"X-API-Key": api_key} if api_key else {} + + if method == "GET": + response = await client.get(f"{portainer_url}/api{endpoint}", headers=headers) + elif method == "POST": + response = await client.post(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "PUT": + response = await client.put(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "DELETE": + response = await client.delete(f"{portainer_url}/api{endpoint}", headers=headers) + else: + raise ValueError(f"Unsupported method: {method}") + + # Parse JSON response safely + try: + data = response.json() if response.text and response.headers.get("content-type", "").startswith("application/json") else None + except Exception: + data = None + + return {"status_code": response.status_code, "data": data, "text": response.text} + + +def format_environment_type(env_type: int) -> str: + """Convert environment type ID to readable string.""" + type_map = { + 1: "Docker", + 2: "Docker Swarm", + 3: "Kubernetes", + 4: "Azure ACI", + 5: "Edge Agent", + 6: "Local Docker", + 7: "Local Kubernetes" + } + return type_map.get(env_type, f"Unknown({env_type})") + + +def format_environment_status(status: int) -> str: + """Convert environment status to readable string.""" + status_map = { + 1: "up", + 2: "down" + } + return status_map.get(status, f"unknown({status})") + + +@server.call_tool() +async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]: + """Handle tool calls.""" + import httpx + + try: + # Basic environment operations + if name == "list_environments": + limit = arguments.get("limit", 100) if arguments else 100 + start = arguments.get("start", 0) if arguments else 0 + + # Try both /environments (new) and /endpoints (old) endpoints + result = await make_request("GET", f"/environments?limit={limit}&start={start}") + if result["status_code"] == 404: + # Legacy endpoint doesn't support pagination params in URL + result = await make_request("GET", "/endpoints") + + if result["status_code"] == 200 and result["data"] is not None: + environments = result["data"] + + # Handle pagination manually for legacy endpoint + if isinstance(environments, list): + total = len(environments) + environments = environments[start:start + limit] + output = f"Found {total} environments (showing {len(environments)}):\n" + else: + output = f"Found environments:\n" + + for env in environments[:10]: # Limit output to first 10 to avoid huge responses + env_type = format_environment_type(env.get("Type", 0)) + status = format_environment_status(env.get("Status", 0)) + output += f"\n- ID: {env.get('Id')}, Name: {env.get('Name')}" + output += f"\n Type: {env_type}, Status: {status}" + output += f"\n URL: {env.get('URL', 'N/A')}" + if env.get("GroupId"): + output += f"\n Group ID: {env.get('GroupId')}" + if env.get("TagIds") or env.get("Tags"): + tags = env.get("Tags", []) + if tags: + tag_names = [t.get('Name', '') for t in tags if isinstance(t, dict)] + if tag_names: + output += f"\n Tags: {', '.join(tag_names)}" + + if len(environments) > 10: + output += f"\n\n... and {len(environments) - 10} more environments" + + return [types.TextContent(type="text", text=output)] + else: + error_msg = f"Failed to list environments: HTTP {result['status_code']}" + if result.get("text"): + error_msg += f" - {result['text'][:200]}" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "get_environment": + env_id = arguments.get("environment_id") + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + # Try both endpoints + result = await make_request("GET", f"/environments/{env_id}") + if result["status_code"] == 404: + result = await make_request("GET", f"/endpoints/{env_id}") + + if result["status_code"] == 200: + env = result["data"] + output = f"Environment Details:\n" + output += f"- ID: {env.get('Id')}\n" + output += f"- Name: {env.get('Name')}\n" + output += f"- Type: {format_environment_type(env.get('Type', 0))}\n" + output += f"- Status: {format_environment_status(env.get('Status', 0))}\n" + output += f"- URL: {env.get('URL', 'N/A')}\n" + output += f"- Public URL: {env.get('PublicURL', 'N/A')}\n" + if env.get("GroupId"): + output += f"- Group ID: {env.get('GroupId')}\n" + if env.get("Tags"): + output += f"- Tags: {', '.join([t.get('Name', '') for t in env.get('Tags', [])])}\n" + if env.get("TLSConfig"): + output += f"- TLS Enabled: {env['TLSConfig'].get('TLS', False)}\n" + output += f"- TLS Skip Verify: {env['TLSConfig'].get('TLSSkipVerify', False)}\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get environment: HTTP {result['status_code']}")] + + elif name == "create_docker_environment": + req_name = arguments.get("name") + url = arguments.get("url") + + if not req_name or not url: + return [types.TextContent(type="text", text="Error: name and url are required")] + + env_data = { + "Name": req_name, + "Type": EnvironmentType.DOCKER.value, + "URL": url, + "EndpointCreationType": 1 # Local environment + } + + if arguments.get("public_url"): + env_data["PublicURL"] = arguments["public_url"] + if arguments.get("group_id"): + env_data["GroupId"] = arguments["group_id"] + if arguments.get("tags"): + env_data["TagIds"] = arguments["tags"] + + # TLS configuration + if arguments.get("tls") or arguments.get("tls_skip_verify"): + env_data["TLSConfig"] = { + "TLS": arguments.get("tls", False), + "TLSSkipVerify": arguments.get("tls_skip_verify", False) + } + + # Try both endpoints + result = await make_request("POST", "/environments", env_data) + if result["status_code"] == 404: + result = await make_request("POST", "/endpoints", env_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Docker environment '{req_name}' created successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to create Docker environment: HTTP {result['status_code']} - {result['text']}")] + + elif name == "create_kubernetes_environment": + req_name = arguments.get("name") + url = arguments.get("url") + + if not req_name or not url: + return [types.TextContent(type="text", text="Error: name and url are required")] + + env_data = { + "Name": req_name, + "Type": EnvironmentType.KUBERNETES.value, + "URL": url, + "EndpointCreationType": 3 # Kubernetes environment + } + + if arguments.get("bearer_token"): + env_data["Token"] = arguments["bearer_token"] + if arguments.get("group_id"): + env_data["GroupId"] = arguments["group_id"] + if arguments.get("tags"): + env_data["TagIds"] = arguments["tags"] + if arguments.get("tls_skip_verify"): + env_data["TLSConfig"] = { + "TLS": True, + "TLSSkipVerify": arguments["tls_skip_verify"] + } + + # Try both endpoints + result = await make_request("POST", "/environments", env_data) + if result["status_code"] == 404: + result = await make_request("POST", "/endpoints", env_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Kubernetes environment '{req_name}' created successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to create Kubernetes environment: HTTP {result['status_code']} - {result['text']}")] + + elif name == "update_environment": + env_id = arguments.get("environment_id") + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + # Get current environment first + result = await make_request("GET", f"/environments/{env_id}") + if result["status_code"] == 404: + result = await make_request("GET", f"/endpoints/{env_id}") + + if result["status_code"] != 200: + return [types.TextContent(type="text", text=f"Failed to get current environment: HTTP {result['status_code']}")] + + env_data = result["data"] + + # Update only provided fields + if "name" in arguments: + env_data["Name"] = arguments["name"] + if "url" in arguments: + env_data["URL"] = arguments["url"] + if "public_url" in arguments: + env_data["PublicURL"] = arguments["public_url"] + if "group_id" in arguments: + env_data["GroupId"] = arguments["group_id"] + if "tags" in arguments: + env_data["TagIds"] = arguments["tags"] + + # Try both endpoints + result = await make_request("PUT", f"/environments/{env_id}", env_data) + if result["status_code"] == 404: + result = await make_request("PUT", f"/endpoints/{env_id}", env_data) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"✓ Environment {env_id} updated successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to update environment: HTTP {result['status_code']}")] + + elif name == "delete_environment": + env_id = arguments.get("environment_id") + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + # Try both endpoints + result = await make_request("DELETE", f"/environments/{env_id}") + if result["status_code"] == 404: + result = await make_request("DELETE", f"/endpoints/{env_id}") + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Environment {env_id} deleted successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to delete environment: HTTP {result['status_code']}")] + + # Environment status and management + elif name == "get_environment_status": + env_id = arguments.get("environment_id") + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + # Try to get docker info through Portainer proxy + result = await make_request("GET", f"/environments/{env_id}/docker/info") + if result["status_code"] == 404: + result = await make_request("GET", f"/endpoints/{env_id}/docker/info") + + if result["status_code"] == 200: + info = result["data"] + output = f"Environment Status:\n" + output += f"- Status: up\n" + output += f"- Docker Version: {info.get('ServerVersion', 'N/A')}\n" + output += f"- Containers: {info.get('Containers', 0)}\n" + output += f"- Running: {info.get('ContainersRunning', 0)}\n" + output += f"- Stopped: {info.get('ContainersStopped', 0)}\n" + output += f"- Images: {info.get('Images', 0)}\n" + output += f"- CPU Count: {info.get('NCPU', 0)}\n" + output += f"- Memory: {info.get('MemTotal', 0) / (1024**3):.2f} GB\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text="Environment is down or inaccessible")] + + elif name == "associate_environment": + env_id = arguments.get("environment_id") + teams = arguments.get("teams", []) + + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + assoc_data = { + "TeamAccessPolicies": {} + } + + for team in teams: + team_id = team.get("team_id") + access_level = team.get("access_level", "read") + if team_id: + assoc_data["TeamAccessPolicies"][str(team_id)] = {"AccessLevel": access_level} + + # Try both endpoints + result = await make_request("PUT", f"/environments/{env_id}/association", assoc_data) + if result["status_code"] == 404: + result = await make_request("PUT", f"/endpoints/{env_id}/association", assoc_data) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"✓ Environment {env_id} team associations updated")] + else: + return [types.TextContent(type="text", text=f"Failed to update associations: HTTP {result['status_code']}")] + + # Environment groups + elif name == "list_environment_groups": + result = await make_request("GET", "/endpoint_groups") + + if result["status_code"] == 200: + groups = result["data"] + output = f"Found {len(groups)} environment groups:\n" + for group in groups: + output += f"- ID: {group.get('Id')}, Name: {group.get('Name')}\n" + if group.get('Description'): + output += f" Description: {group.get('Description')}\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list groups: HTTP {result['status_code']}")] + + elif name == "create_environment_group": + group_name = arguments.get("name") + if not group_name: + return [types.TextContent(type="text", text="Error: name is required")] + + group_data = { + "Name": group_name, + "Description": arguments.get("description", "") + } + + result = await make_request("POST", "/endpoint_groups", group_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Environment group '{group_name}' created successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to create group: HTTP {result['status_code']}")] + + elif name == "update_environment_group": + group_id = arguments.get("group_id") + if not group_id: + return [types.TextContent(type="text", text="Error: group_id is required")] + + update_data = {} + if "name" in arguments: + update_data["Name"] = arguments["name"] + if "description" in arguments: + update_data["Description"] = arguments["description"] + + result = await make_request("PUT", f"/endpoint_groups/{group_id}", update_data) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"✓ Environment group {group_id} updated successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to update group: HTTP {result['status_code']}")] + + elif name == "delete_environment_group": + group_id = arguments.get("group_id") + if not group_id: + return [types.TextContent(type="text", text="Error: group_id is required")] + + result = await make_request("DELETE", f"/endpoint_groups/{group_id}") + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Environment group {group_id} deleted successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to delete group: HTTP {result['status_code']}")] + + # Tags management + elif name == "list_tags": + result = await make_request("GET", "/tags") + + if result["status_code"] == 200: + tags = result["data"] + output = f"Found {len(tags)} tags:\n" + for tag in tags: + output += f"- ID: {tag.get('ID')}, Name: {tag.get('Name')}\n" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list tags: HTTP {result['status_code']}")] + + elif name == "create_tag": + tag_name = arguments.get("name") + if not tag_name: + return [types.TextContent(type="text", text="Error: name is required")] + + tag_data = {"Name": tag_name} + result = await make_request("POST", "/tags", tag_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"✓ Tag '{tag_name}' created successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to create tag: HTTP {result['status_code']}")] + + elif name == "delete_tag": + tag_id = arguments.get("tag_id") + if not tag_id: + return [types.TextContent(type="text", text="Error: tag_id is required")] + + result = await make_request("DELETE", f"/tags/{tag_id}") + + if result["status_code"] in [200, 204]: + return [types.TextContent(type="text", text=f"✓ Tag {tag_id} deleted successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to delete tag: HTTP {result['status_code']}")] + + # Edge environments + elif name == "generate_edge_key": + edge_name = arguments.get("name") + if not edge_name: + return [types.TextContent(type="text", text="Error: name is required")] + + edge_data = { + "Name": edge_name, + "Type": EnvironmentType.EDGE_AGENT.value, + "EndpointCreationType": 4 # Edge agent + } + + if arguments.get("group_id"): + edge_data["GroupId"] = arguments["group_id"] + + # Try both endpoints + result = await make_request("POST", "/environments", edge_data) + if result["status_code"] == 404: + result = await make_request("POST", "/endpoints", edge_data) + + if result["status_code"] in [200, 201]: + env_id = result["data"].get("Id") + edge_key = result["data"].get("EdgeKey", "") + output = f"✓ Edge environment '{edge_name}' created\n" + output += f"- Environment ID: {env_id}\n" + output += f"- Edge Key: {edge_key}\n" + output += f"\nDeployment command:\n" + output += f"docker run -d --name portainer_edge_agent --restart always \\\n" + output += f" -v /var/run/docker.sock:/var/run/docker.sock \\\n" + output += f" -v /var/lib/docker/volumes:/var/lib/docker/volumes \\\n" + output += f" -v /:/host \\\n" + output += f" -v portainer_agent_data:/data \\\n" + output += f" --env EDGE=1 \\\n" + output += f" --env EDGE_ID={env_id} \\\n" + output += f" --env EDGE_KEY={edge_key} \\\n" + output += f" --env EDGE_INSECURE_POLL=1 \\\n" + output += f" portainer/agent:latest" + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to create Edge environment: HTTP {result['status_code']}")] + + else: + return [types.TextContent(type="text", text=f"Unknown tool: {name}")] + + except httpx.TimeoutException: + return [types.TextContent(type="text", text="Error: Request timed out. The Portainer server may be slow or unresponsive.")] + except httpx.ConnectError: + return [types.TextContent(type="text", text="Error: Could not connect to Portainer server. Please check the URL and network connection.")] + except Exception as e: + import traceback + error_details = f"Error: {str(e)}\nType: {type(e).__name__}" + if hasattr(e, "__traceback__"): + error_details += f"\nTraceback: {traceback.format_exc()}" + return [types.TextContent(type="text", text=error_details)] + + +async def run(): + """Run the MCP server.""" + # Use stdio transport + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="portainer-environments", + server_version="1.0.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions( + prompts_changed=False, + resources_changed=False, + tools_changed=False, + ), + experimental_capabilities={}, + ), + ), + ) + + +def main(): + """Main entry point.""" + asyncio.run(run()) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..fb40685 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +mcp>=1.0.0 +httpx>=0.25.0 +pydantic>=2.0.0 +pydantic-settings>=2.0.0 +structlog>=23.0.0 +PyJWT>=2.8.0 +python-dotenv>=1.0.0 +tenacity>=8.0.0 \ No newline at end of file diff --git a/run_mcp.py b/run_mcp.py new file mode 100755 index 0000000..c472453 --- /dev/null +++ b/run_mcp.py @@ -0,0 +1,14 @@ +#!/usr/bin/env python3 +"""Direct runner for Portainer MCP server without dev dependencies.""" + +import sys +import os + +# Add src to path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) + +# Import and run the server +from portainer_core.server import main_sync + +if __name__ == "__main__": + main_sync() \ No newline at end of file diff --git a/simple_mcp_server.py b/simple_mcp_server.py new file mode 100755 index 0000000..f2c3114 --- /dev/null +++ b/simple_mcp_server.py @@ -0,0 +1,241 @@ +#!/usr/bin/env python3 +"""Simple MCP server for Portainer that actually works.""" + +import os +import sys +import json +import asyncio +from typing import Any, Dict, List + +# Suppress all logging to stderr +os.environ["MCP_MODE"] = "true" + +import mcp.server.stdio +import mcp.types as types +from mcp.server import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +# Add src to path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) + +# Create server +server = Server("portainer-core") + +# Store for our state +portainer_url = os.getenv("PORTAINER_URL", "https://partner.portainer.live") +api_key = os.getenv("PORTAINER_API_KEY", "") + + +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + types.Tool( + name="test_connection", + description="Test connection to Portainer", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="get_users", + description="Get list of Portainer users", + inputSchema={ + "type": "object", + "properties": {}, + "required": [] + } + ), + types.Tool( + name="create_user", + description="Create a new Portainer user", + inputSchema={ + "type": "object", + "properties": { + "username": { + "type": "string", + "description": "Username for the new user" + }, + "password": { + "type": "string", + "description": "Password for the new user" + }, + "role": { + "type": "string", + "description": "User role (Administrator, StandardUser, ReadOnlyUser)", + "enum": ["Administrator", "StandardUser", "ReadOnlyUser"], + "default": "StandardUser" + } + }, + "required": ["username", "password"] + } + ) + ] + + +@server.call_tool() +async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]: + """Handle tool calls.""" + import httpx + + if name == "test_connection": + try: + async with httpx.AsyncClient(verify=False) as client: + response = await client.get( + f"{portainer_url}/api/status", + headers={"X-API-Key": api_key} if api_key else {} + ) + if response.status_code == 200: + return [types.TextContent( + type="text", + text=f"✓ Connected to Portainer at {portainer_url}" + )] + else: + return [types.TextContent( + type="text", + text=f"✗ Failed to connect: HTTP {response.status_code}" + )] + except Exception as e: + return [types.TextContent( + type="text", + text=f"✗ Connection error: {str(e)}" + )] + + elif name == "get_users": + try: + async with httpx.AsyncClient(verify=False) as client: + response = await client.get( + f"{portainer_url}/api/users", + headers={"X-API-Key": api_key} + ) + if response.status_code == 200: + users = response.json() + result = f"Found {len(users)} users:\n" + for user in users: + # Handle both old (int) and new (string) role formats + role = user.get("Role", "Unknown") + if isinstance(role, int): + role_map = {1: "Administrator", 2: "StandardUser", 3: "ReadOnlyUser"} + role = role_map.get(role, f"Unknown({role})") + result += f"- {user.get('Username', 'Unknown')} ({role})\n" + return [types.TextContent(type="text", text=result)] + else: + return [types.TextContent( + type="text", + text=f"Failed to get users: HTTP {response.status_code}" + )] + except Exception as e: + return [types.TextContent( + type="text", + text=f"Error getting users: {str(e)}" + )] + + elif name == "create_user": + if not arguments: + return [types.TextContent( + type="text", + text="Error: Missing required arguments" + )] + + username = arguments.get("username") + password = arguments.get("password") + role = arguments.get("role", "StandardUser") + + if not username or not password: + return [types.TextContent( + type="text", + text="Error: Username and password are required" + )] + + # Convert role string to integer for older Portainer versions + role_map = {"Administrator": 1, "StandardUser": 2, "ReadOnlyUser": 3} + role_int = role_map.get(role, 2) + + try: + async with httpx.AsyncClient(verify=False) as client: + # Try with integer role first (older versions) + response = await client.post( + f"{portainer_url}/api/users", + headers={"X-API-Key": api_key}, + json={ + "Username": username, + "Password": password, + "Role": role_int + } + ) + + if response.status_code == 409: + return [types.TextContent( + type="text", + text=f"User '{username}' already exists" + )] + elif response.status_code in [200, 201]: + return [types.TextContent( + type="text", + text=f"✓ User '{username}' created successfully with role {role}" + )] + else: + # Try with string role (newer versions) + response = await client.post( + f"{portainer_url}/api/users", + headers={"X-API-Key": api_key}, + json={ + "Username": username, + "Password": password, + "Role": role + } + ) + if response.status_code in [200, 201]: + return [types.TextContent( + type="text", + text=f"✓ User '{username}' created successfully with role {role}" + )] + else: + return [types.TextContent( + type="text", + text=f"Failed to create user: HTTP {response.status_code} - {response.text}" + )] + except Exception as e: + return [types.TextContent( + type="text", + text=f"Error creating user: {str(e)}" + )] + + else: + return [types.TextContent( + type="text", + text=f"Unknown tool: {name}" + )] + + +async def run(): + """Run the MCP server.""" + # Use stdio transport + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="portainer-core", + server_version="0.2.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions( + prompts_changed=False, + resources_changed=False, + tools_changed=False, + ), + experimental_capabilities={}, + ), + ), + ) + + +def main(): + """Main entry point.""" + asyncio.run(run()) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/portainer_core/__pycache__/__init__.cpython-312.pyc b/src/portainer_core/__pycache__/__init__.cpython-312.pyc deleted file mode 100644 index 2d95e82..0000000 Binary files a/src/portainer_core/__pycache__/__init__.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/__pycache__/config.cpython-312.pyc b/src/portainer_core/__pycache__/config.cpython-312.pyc deleted file mode 100644 index 07f5438..0000000 Binary files a/src/portainer_core/__pycache__/config.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/__pycache__/server.cpython-312.pyc b/src/portainer_core/__pycache__/server.cpython-312.pyc deleted file mode 100644 index e98579a..0000000 Binary files a/src/portainer_core/__pycache__/server.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/models/__pycache__/__init__.cpython-312.pyc b/src/portainer_core/models/__pycache__/__init__.cpython-312.pyc deleted file mode 100644 index 4966039..0000000 Binary files a/src/portainer_core/models/__pycache__/__init__.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/models/__pycache__/auth.cpython-312.pyc b/src/portainer_core/models/__pycache__/auth.cpython-312.pyc deleted file mode 100644 index 314ccad..0000000 Binary files a/src/portainer_core/models/__pycache__/auth.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/models/__pycache__/settings.cpython-312.pyc b/src/portainer_core/models/__pycache__/settings.cpython-312.pyc deleted file mode 100644 index 47345d1..0000000 Binary files a/src/portainer_core/models/__pycache__/settings.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/models/__pycache__/users.cpython-312.pyc b/src/portainer_core/models/__pycache__/users.cpython-312.pyc deleted file mode 100644 index 0734c66..0000000 Binary files a/src/portainer_core/models/__pycache__/users.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/services/__pycache__/__init__.cpython-312.pyc b/src/portainer_core/services/__pycache__/__init__.cpython-312.pyc deleted file mode 100644 index 3e5d91a..0000000 Binary files a/src/portainer_core/services/__pycache__/__init__.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/services/__pycache__/auth.cpython-312.pyc b/src/portainer_core/services/__pycache__/auth.cpython-312.pyc deleted file mode 100644 index 12ae685..0000000 Binary files a/src/portainer_core/services/__pycache__/auth.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/services/__pycache__/base.cpython-312.pyc b/src/portainer_core/services/__pycache__/base.cpython-312.pyc deleted file mode 100644 index 3858a72..0000000 Binary files a/src/portainer_core/services/__pycache__/base.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/services/__pycache__/settings.cpython-312.pyc b/src/portainer_core/services/__pycache__/settings.cpython-312.pyc deleted file mode 100644 index fd72d24..0000000 Binary files a/src/portainer_core/services/__pycache__/settings.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/services/__pycache__/users.cpython-312.pyc b/src/portainer_core/services/__pycache__/users.cpython-312.pyc deleted file mode 100644 index 1345960..0000000 Binary files a/src/portainer_core/services/__pycache__/users.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/utils/__pycache__/__init__.cpython-312.pyc b/src/portainer_core/utils/__pycache__/__init__.cpython-312.pyc deleted file mode 100644 index 4eb41dc..0000000 Binary files a/src/portainer_core/utils/__pycache__/__init__.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/utils/__pycache__/errors.cpython-312.pyc b/src/portainer_core/utils/__pycache__/errors.cpython-312.pyc deleted file mode 100644 index ea68d33..0000000 Binary files a/src/portainer_core/utils/__pycache__/errors.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/utils/__pycache__/logging.cpython-312.pyc b/src/portainer_core/utils/__pycache__/logging.cpython-312.pyc deleted file mode 100644 index 61961f8..0000000 Binary files a/src/portainer_core/utils/__pycache__/logging.cpython-312.pyc and /dev/null differ diff --git a/src/portainer_core/utils/__pycache__/tokens.cpython-312.pyc b/src/portainer_core/utils/__pycache__/tokens.cpython-312.pyc deleted file mode 100644 index c92e5cc..0000000 Binary files a/src/portainer_core/utils/__pycache__/tokens.cpython-312.pyc and /dev/null differ diff --git a/tests/__pycache__/test_basic.cpython-312-pytest-7.4.3.pyc b/tests/__pycache__/test_basic.cpython-312-pytest-7.4.3.pyc deleted file mode 100644 index bdac776..0000000 Binary files a/tests/__pycache__/test_basic.cpython-312-pytest-7.4.3.pyc and /dev/null differ