feat: add three new Portainer MCP servers
- Add portainer-environments server for environment/endpoint management - Add portainer-docker server for Docker and Swarm container operations - Add merged portainer server combining core + teams functionality - Fix JSON schema issues and API compatibility - Add comprehensive documentation for each server - Add .gitignore and .env.example for security 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
7c32d69f2d
commit
e27251b922
@ -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
|
||||
|
||||
|
||||
56
.gitignore
vendored
Normal file
56
.gitignore
vendored
Normal file
@ -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
|
||||
239
README_DOCKER.md
Normal file
239
README_DOCKER.md
Normal file
@ -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
|
||||
196
README_ENVIRONMENTS.md
Normal file
196
README_ENVIRONMENTS.md
Normal file
@ -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)
|
||||
133
README_MERGED.md
Normal file
133
README_MERGED.md
Normal file
@ -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
|
||||
613
merged_mcp_server.py
Executable file
613
merged_mcp_server.py
Executable file
@ -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()
|
||||
1431
portainer_docker_server.py
Executable file
1431
portainer_docker_server.py
Executable file
File diff suppressed because it is too large
Load Diff
861
portainer_environments_server.py
Executable file
861
portainer_environments_server.py
Executable file
@ -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()
|
||||
8
requirements.txt
Normal file
8
requirements.txt
Normal file
@ -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
|
||||
14
run_mcp.py
Executable file
14
run_mcp.py
Executable file
@ -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()
|
||||
241
simple_mcp_server.py
Executable file
241
simple_mcp_server.py
Executable file
@ -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()
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Loading…
Reference in New Issue
Block a user