adelorenzo/portainer-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Simplify authentication to require URL and API key only

Browse Source 
Major configuration and tooling updates:

Authentication Changes:
- Remove username/password authentication support
- Require PORTAINER_URL and PORTAINER_API_KEY (both mandatory)
- Simplify PortainerConfig class and validation logic
- Update all documentation to reflect API key requirement

Multiple Runtime Support:
- Add uvx support for running without installation
- Add uv support with dedicated wrapper script
- Add npx support with Node.js wrapper script
- Maintain backward compatibility with direct Python execution

Documentation Updates:
- Comprehensive README.md with all execution methods
- Detailed USAGE.md with step-by-step instructions
- Updated .env.example with clear required vs optional sections
- Enhanced docstrings in server.py and config.py

Tooling Support:
- package.json for npm/npx support with cross-platform wrapper
- scripts/run-with-uv.py for uv integration
- bin/portainer-core-mcp Node.js wrapper for npx
- test_uvx.py for uvx functionality testing

Configuration Improvements:
- Clear separation of required vs optional environment variables
- Better validation error messages
- Simplified authentication flow
- Enhanced project metadata in pyproject.toml

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Adolfo Delorenzo
2025-07-18 07:48:23 -06:00
parent 84ca8aee99
commit be1e05c382
14 changed files with 1727 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
src/portainer_core/__pycache__/__init__.cpython-312.pyc
View File

Binary file not shown.

BIN
src/portainer_core/__pycache__/config.cpython-312.pyc
View File

Binary file not shown.

BIN
src/portainer_core/__pycache__/server.cpython-312.pyc
View File

Binary file not shown.

347
src/portainer_core/config.py
View File

@@ -1,7 +1,41 @@
"""
Configuration management for Portainer Core MCP Server.
This module handles environment variable validation and configuration settings.
This module provides comprehensive configuration management for the Portainer Core MCP Server,
including environment variable validation, authentication setup, and runtime settings.
The configuration system supports:
- Environment variable loading with validation
- Authentication via API key (required)
- HTTP client configuration with retry logic
- Circuit breaker settings for fault tolerance
- Token management and caching
- Logging configuration
- Development and debug settings
Key Components:
- PortainerConfig: Main configuration class with validation
- ServerConfig: Server-specific settings (host, port, name)
- Global configuration instances with lazy initialization
- Factory functions for configuration creation
- Validation methods for authentication and settings
Usage:
```python
from portainer_core.config import get_global_config
config = get_global_config()
print(config.portainer_url)
print(config.api_base_url)
```
Environment Variables:
- PORTAINER_URL: Base URL of Portainer instance (required)
- PORTAINER_API_KEY: API key for authentication (required)
- HTTP_TIMEOUT: HTTP request timeout in seconds (default: 30)
- MAX_RETRIES: Maximum retry attempts (default: 3)
- LOG_LEVEL: Logging level (default: INFO)
- DEBUG: Enable debug mode (default: false)
"""
import os
@@ -13,7 +47,44 @@ from pydantic_settings import BaseSettings
class PortainerConfig(BaseSettings):
"""Configuration settings for Portainer Core MCP Server."""
"""
Configuration settings for Portainer Core MCP Server.
This class defines all configuration parameters for the MCP server,
including connection settings, authentication, HTTP client behavior,
circuit breaker configuration, and logging settings.
The configuration is loaded from environment variables with validation
and provides computed properties for common operations.
Configuration Categories:
- Connection: Portainer URL and API endpoints
- Authentication: API key (required)
- HTTP Client: Timeouts, retries, and connection settings
- Circuit Breaker: Fault tolerance and recovery settings
- Token Management: Caching and refresh policies
- Logging: Level, format, and debug settings
Validation:
- URL format validation with scheme checking
- API key presence validation (required field)
- Log level and format validation
- Numeric range validation for timeouts and thresholds
Properties:
- api_base_url: Computed API base URL
Example:
```python
# Environment variables:
# PORTAINER_URL=https://portainer.example.com
# PORTAINER_API_KEY=your-api-key
config = PortainerConfig()
print(config.api_base_url) # https://portainer.example.com/api
print(config.portainer_api_key) # your-api-key
```
"""
# Portainer connection settings
portainer_url: str = Field(
@@ -21,22 +92,12 @@ class PortainerConfig(BaseSettings):
description="Base URL of the Portainer instance"
)
# Authentication settings (either API key or username/password)
portainer_api_key: Optional[str] = Field(
default=None,
# Authentication settings (API key required)
portainer_api_key: str = Field(
...,
description="Portainer API key for authentication"
)
portainer_username: Optional[str] = Field(
default=None,
description="Portainer username for authentication"
)
portainer_password: Optional[str] = Field(
default=None,
description="Portainer password for authentication"
)
# HTTP client settings
http_timeout: int = Field(
default=30,
@@ -101,7 +162,32 @@ class PortainerConfig(BaseSettings):
@field_validator("portainer_url")
@classmethod
def validate_portainer_url(cls, v):
"""Validate Portainer URL format."""
"""
Validate Portainer URL format and structure.
Ensures the provided URL is properly formatted, uses a valid scheme,
and normalizes the URL by removing trailing slashes.
Args:
v: URL string to validate
Returns:
str: Validated and normalized URL (trailing slash removed)
Raises:
ValueError: If URL is empty, malformed, or uses invalid scheme
Validation Rules:
- URL must not be empty
- Must have valid scheme (http or https)
- Must have network location (hostname)
- Trailing slashes are removed for consistency
Examples:
- https://portainer.example.com/ -> https://portainer.example.com
- http://localhost:9000 -> http://localhost:9000
- invalid-url -> raises ValueError
"""
if not v:
raise ValueError("Portainer URL is required")
@@ -121,7 +207,33 @@ class PortainerConfig(BaseSettings):
@field_validator("log_level")
@classmethod
def validate_log_level(cls, v):
"""Validate logging level."""
"""
Validate logging level against allowed values.
Ensures the logging level is one of the standard Python logging levels
and normalizes it to uppercase.
Args:
v: Log level string to validate
Returns:
str: Validated log level in uppercase
Raises:
ValueError: If log level is not in allowed values
Valid Levels:
- DEBUG: Detailed diagnostic information
- INFO: General information about program execution
- WARNING: Warning messages for potential issues
- ERROR: Error messages for serious problems
- CRITICAL: Critical error messages for fatal problems
Examples:
- "info" -> "INFO"
- "Debug" -> "DEBUG"
- "invalid" -> raises ValueError
"""
valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
if v.upper() not in valid_levels:
raise ValueError(f"Log level must be one of {valid_levels}")
@@ -130,52 +242,103 @@ class PortainerConfig(BaseSettings):
@field_validator("log_format")
@classmethod
def validate_log_format(cls, v):
"""Validate logging format."""
"""
Validate logging format against allowed values.
Ensures the logging format is one of the supported formats
and normalizes it to lowercase.
Args:
v: Log format string to validate
Returns:
str: Validated log format in lowercase
Raises:
ValueError: If log format is not in allowed values
Valid Formats:
- json: Structured JSON logging for machine parsing
- text: Human-readable text logging for development
Examples:
- "JSON" -> "json"
- "Text" -> "text"
- "invalid" -> raises ValueError
"""
valid_formats = ["json", "text"]
if v.lower() not in valid_formats:
raise ValueError(f"Log format must be one of {valid_formats}")
return v.lower()
def validate_auth_config(self) -> None:
"""Validate authentication configuration."""
has_api_key = self.portainer_api_key is not None
has_credentials = (
self.portainer_username is not None
and self.portainer_password is not None
)
"""
Validate authentication configuration.
if not has_api_key and not has_credentials:
Ensures that the required API key is configured properly.
Authentication Method:
- API Key: Direct API key authentication (required)
Validation Rules:
- API key must be provided and non-empty
- API key should not contain only whitespace
Raises:
ValueError: If API key is missing or invalid
Note:
This method is called automatically during configuration validation
to ensure authentication is properly configured.
"""
if not self.portainer_api_key or not self.portainer_api_key.strip():
raise ValueError(
"Either PORTAINER_API_KEY or both PORTAINER_USERNAME and "
"PORTAINER_PASSWORD must be provided"
"PORTAINER_API_KEY must be provided and cannot be empty"
)
if has_api_key and has_credentials:
# Prefer API key if both are provided
self.portainer_username = None
self.portainer_password = None
@property
def api_base_url(self) -> str:
"""Get the base API URL."""
"""
Get the base API URL for Portainer REST API.
Constructs the full API base URL by appending '/api' to the
configured Portainer URL.
Returns:
str: Complete API base URL for making requests
Example:
If portainer_url is 'https://portainer.example.com',
returns 'https://portainer.example.com/api'
"""
return f"{self.portainer_url}/api"
@property
def use_api_key_auth(self) -> bool:
"""Check if API key authentication should be used."""
return self.portainer_api_key is not None
@property
def use_credentials_auth(self) -> bool:
"""Check if username/password authentication should be used."""
return (
self.portainer_username is not None
and self.portainer_password is not None
)
class ServerConfig(BaseModel):
"""Server-specific configuration settings."""
"""
Server-specific configuration settings.
This class defines configuration parameters specific to the MCP server
instance, including network settings and server identification.
Configuration Parameters:
- host: Server host address (default: localhost)
- port: Server port number (default: 8000)
- server_name: Server identifier for MCP protocol (default: portainer-core-mcp)
- version: Server version string (default: 0.1.0)
Usage:
```python
config = ServerConfig()
print(f"Server: {config.server_name} v{config.version}")
print(f"Listening on: {config.host}:{config.port}")
```
Note:
This configuration is separate from PortainerConfig to allow
independent management of server vs. Portainer connection settings.
"""
host: str = Field(
default="localhost",
@@ -199,14 +362,46 @@ class ServerConfig(BaseModel):
def get_config() -> PortainerConfig:
"""Get validated configuration instance."""
"""
Get validated configuration instance.
Creates a new PortainerConfig instance from environment variables
and validates the authentication configuration.
Returns:
PortainerConfig: Validated configuration instance
Raises:
ValueError: If configuration is invalid or authentication is missing
ValidationError: If environment variables have invalid values
Process:
1. Create PortainerConfig instance from environment variables
2. Validate authentication configuration
3. Return validated configuration
Note:
This function creates a new instance each time it's called.
For singleton behavior, use get_global_config() instead.
"""
config = PortainerConfig()
config.validate_auth_config()
return config
def get_server_config() -> ServerConfig:
"""Get server configuration instance."""
"""
Get server configuration instance.
Creates a new ServerConfig instance with default values.
Returns:
ServerConfig: Server configuration instance
Note:
This function creates a new instance each time it's called.
For singleton behavior, use get_global_server_config() instead.
"""
return ServerConfig()
@@ -215,14 +410,66 @@ _config = None
_server_config = None
def get_global_config() -> PortainerConfig:
"""Get global configuration instance with lazy initialization."""
"""
Get global configuration instance with lazy initialization.
Implements singleton pattern for configuration management. The configuration
is loaded and validated only once, then cached for subsequent calls.
Returns:
PortainerConfig: Global configuration instance
Thread Safety:
This function is not thread-safe. In multi-threaded environments,
ensure it's called during initialization before spawning threads.
Lazy Initialization:
The configuration is only loaded when first accessed, which allows
for environment variable changes during testing.
Usage:
```python
# First call loads and validates configuration
config = get_global_config()
# Subsequent calls return cached instance
same_config = get_global_config()
assert config is same_config # True
```
"""
global _config
if _config is None:
_config = get_config()
return _config
def get_global_server_config() -> ServerConfig:
"""Get global server configuration instance with lazy initialization."""
"""
Get global server configuration instance with lazy initialization.
Implements singleton pattern for server configuration management.
The configuration is created only once, then cached for subsequent calls.
Returns:
ServerConfig: Global server configuration instance
Thread Safety:
This function is not thread-safe. In multi-threaded environments,
ensure it's called during initialization before spawning threads.
Lazy Initialization:
The configuration is only created when first accessed, which reduces
startup overhead when server configuration isn't needed.
Usage:
```python
# First call creates configuration
config = get_global_server_config()
# Subsequent calls return cached instance
same_config = get_global_server_config()
assert config is same_config # True
```
"""
global _server_config
if _server_config is None:
_server_config = get_server_config()

763
src/portainer_core/server.py
View File

@@ -34,9 +34,64 @@ logger = get_logger(__name__)
class PortainerCoreMCPServer:
"""Main MCP server class for Portainer Core functionality."""
"""
Main MCP server class for Portainer Core functionality.
This class implements the Model Context Protocol (MCP) server that provides
authentication and user management functionality for Portainer Business Edition.
It handles MCP protocol communication, resource management, and tool execution.
The server provides the following capabilities:
- User authentication with JWT token management
- User management operations (CRUD)
- Settings configuration management
- Health monitoring and status reporting
- Resource and tool discovery through MCP protocol
Attributes:
server: The underlying MCP server instance
auth_service: Authentication service for login/token operations
user_service: User management service for CRUD operations
settings_service: Settings management service for configuration
Architecture:
The server follows a service-oriented architecture with:
- Service Layer: Business logic separation (auth, users, settings)
- Error Handling: Comprehensive error mapping and logging
- Circuit Breaker: Fault tolerance with automatic recovery
- Correlation IDs: Request tracing and debugging support
Usage:
```python
server = PortainerCoreMCPServer()
await server.run()
```
"""
def __init__(self):
"""
Initialize the Portainer Core MCP Server.
Sets up the MCP server instance with configuration from environment variables,
initializes the handler setup, and prepares service instances for lazy loading.
The initialization process:
1. Loads global configuration from environment variables
2. Creates the underlying MCP server instance
3. Sets up MCP protocol handlers for resources and tools
4. Prepares service instances for lazy initialization
5. Logs initialization success with configuration details
Raises:
PortainerConfigurationError: If required configuration is missing
PortainerError: If server initialization fails
Side Effects:
- Creates MCP server instance
- Registers protocol handlers
- Initializes logging context
- Prepares service instances (not yet initialized)
"""
config = get_global_config()
server_config = get_global_server_config()
@@ -56,11 +111,57 @@ class PortainerCoreMCPServer:
)
def setup_handlers(self) -> None:
"""Set up MCP server handlers."""
"""
Set up MCP server handlers for resources and tools.
Registers handlers for the MCP protocol operations:
- Resource handlers: For listing and reading available resources
- Tool handlers: For executing available tools and operations
The handlers are registered using decorators on the server instance:
- @server.list_resources(): Lists available resources (users, settings, health)
- @server.read_resource(): Reads specific resource content
- @server.list_tools(): Lists available tools with their schemas
- @server.call_tool(): Executes specific tools with arguments
Resource Types:
- portainer://users: User management resource
- portainer://settings: Settings configuration resource
- portainer://health: Server health status resource
Tool Types:
- Authentication: authenticate, generate_token, get_current_user
- User Management: list_users, create_user, update_user, delete_user
- Settings: get_settings, update_settings
- Health: health_check
Complexity: O(1) - Handler registration is constant time
Side Effects:
- Registers MCP protocol handlers on server instance
- Creates handler closures with access to server state
"""
@self.server.list_resources()
async def handle_list_resources() -> List[Resource]:
"""List available resources."""
"""
List available MCP resources.
Returns a list of all available resources that can be accessed through
the MCP protocol. Each resource represents a collection of data that
can be read or queried.
Returns:
List[Resource]: List of available resources with metadata:
- portainer://users: User management data
- portainer://settings: Configuration settings
- portainer://health: Server health status
Complexity: O(1) - Returns static resource list
Note:
This is a discovery endpoint - no authentication required
"""
return [
Resource(
uri="portainer://users",
@@ -84,7 +185,38 @@ class PortainerCoreMCPServer:
@self.server.read_resource()
async def handle_read_resource(uri: str) -> str:
"""Read a specific resource."""
"""
Read a specific MCP resource by URI.
Retrieves the content of a specific resource identified by its URI.
Different resource types return different data formats and may have
different authentication requirements.
Args:
uri: Resource URI to read (e.g., 'portainer://users')
Returns:
str: Resource content as string (usually JSON formatted)
Raises:
PortainerError: If resource URI is unknown or invalid
PortainerAuthenticationError: If authentication is required but missing
PortainerNetworkError: If communication with Portainer fails
Supported URIs:
- portainer://health: Server health status (no auth required)
- portainer://users: User list (requires authentication)
- portainer://settings: Settings data (requires authentication)
Complexity: O(n) where n is the size of the resource data
Flow:
1. Log resource access attempt
2. Route to appropriate resource handler
3. Ensure required services are initialized
4. Fetch and return resource data
5. Handle errors with appropriate error types
"""
with LogContext():
logger.info("Reading resource", uri=uri)
@@ -104,7 +236,37 @@ class PortainerCoreMCPServer:
@self.server.list_tools()
async def handle_list_tools() -> List[Tool]:
"""List available tools."""
"""
List available MCP tools.
Returns a comprehensive list of all available tools that can be executed
through the MCP protocol. Each tool includes its schema definition,
parameter requirements, and description.
Returns:
List[Tool]: List of available tools with complete schemas:
- Authentication tools: authenticate, generate_token, get_current_user
- User management tools: list_users, create_user, update_user, delete_user
- Settings tools: get_settings, update_settings
- Health tools: health_check
Tool Categories:
1. Authentication (3 tools): Login, token generation, current user
2. User Management (4 tools): Full CRUD operations for users
3. Settings (2 tools): Get and update configuration
4. Health (1 tool): Server health monitoring
Schema Features:
- JSON Schema validation for all parameters
- Required vs optional field definitions
- Type validation and constraints
- Human-readable descriptions
Complexity: O(1) - Returns static tool list
Note:
This is a discovery endpoint - no authentication required
"""
return [
Tool(
name="authenticate",
@@ -258,7 +420,46 @@ class PortainerCoreMCPServer:
@self.server.call_tool()
async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls."""
"""
Handle MCP tool execution requests.
Executes a specific tool with provided arguments and returns the result.
Each tool call is tracked with correlation IDs for debugging and logging.
Args:
name: Tool name to execute (e.g., 'authenticate', 'list_users')
arguments: Tool arguments as dictionary (validated against tool schema)
Returns:
List[TextContent]: Tool execution results wrapped in TextContent
Raises:
PortainerError: If tool name is unknown
PortainerValidationError: If arguments are invalid
PortainerAuthenticationError: If authentication is required but missing
PortainerNetworkError: If communication with Portainer fails
Tool Execution Flow:
1. Generate correlation ID for request tracing
2. Log tool execution attempt with arguments
3. Route to appropriate tool handler method
4. Ensure required services are initialized
5. Execute tool with argument validation
6. Return formatted result or error message
Error Handling:
- All exceptions are caught and logged
- Error messages are returned as TextContent
- Correlation IDs are preserved for debugging
- Service failures are gracefully handled
Complexity: O(f) where f is the complexity of the specific tool function
Security:
- Arguments are validated against tool schemas
- Authentication is enforced per tool requirements
- Sensitive data is not logged in arguments
"""
correlation_id = set_correlation_id()
with LogContext(correlation_id):
@@ -297,7 +498,39 @@ class PortainerCoreMCPServer:
return [TextContent(type="text", text=error_message)]
async def _get_health_status(self) -> str:
"""Get server health status."""
"""
Get comprehensive server health status.
Performs health checks on all services and returns a detailed status report.
This method is used by both the health resource and health_check tool.
Returns:
str: JSON-formatted health status containing:
- Overall server status (healthy/degraded)
- Individual service statuses
- Server configuration details
- Portainer connection status
Health Status Levels:
- healthy: All services operational
- degraded: Some services failing
- unhealthy: Critical services failing
- not_initialized: Services not yet initialized
Complexity: O(s) where s is the number of services to check
Flow:
1. Ensure all services are initialized
2. Check health of each service individually
3. Aggregate results into overall status
4. Include configuration and connection details
5. Return formatted status report
Error Handling:
- Service failures are caught and reported
- Partial failures don't prevent status reporting
- Connection errors are logged but don't crash health check
"""
try:
config = get_global_config()
server_config = get_global_server_config()
@@ -340,7 +573,41 @@ class PortainerCoreMCPServer:
return f"Health check failed: {str(e)}"
async def _ensure_services_initialized(self) -> None:
"""Ensure all services are initialized."""
"""
Ensure all required services are properly initialized.
Implements lazy initialization pattern for services - they are only
initialized when first needed. This reduces startup time and resource
usage when services aren't required.
Services Initialized:
- AuthService: JWT authentication and token management
- UserService: User CRUD operations and management
- SettingsService: Configuration management
Initialization Process:
1. Check if service instance exists (None check)
2. Create service instance if needed
3. Call service.initialize() for setup
4. Service handles its own configuration and HTTP client setup
Complexity: O(1) per service - constant time initialization
Thread Safety:
- Safe for concurrent access (async/await pattern)
- Services handle their own initialization state
- No shared mutable state during initialization
Error Handling:
- Service initialization failures are propagated
- Partial initialization leaves other services unaffected
- Subsequent calls will retry failed initializations
Side Effects:
- Creates service instances
- Establishes HTTP client connections
- Validates service configurations
"""
if self.auth_service is None:
self.auth_service = AuthService()
await self.auth_service.initialize()
@@ -354,7 +621,31 @@ class PortainerCoreMCPServer:
await self.settings_service.initialize()
async def _get_users_resource(self) -> str:
"""Get users resource."""
"""
Get users resource data.
Retrieves the complete list of users from Portainer for the users resource.
This method is called when the portainer://users resource is accessed.
Returns:
str: JSON-formatted user list or error message
Raises:
PortainerAuthenticationError: If authentication is required but missing
PortainerNetworkError: If communication with Portainer fails
Complexity: O(n) where n is the number of users in the system
Flow:
1. Ensure UserService is initialized
2. Call user_service.list_users() to fetch data
3. Convert result to string format
4. Handle errors gracefully with logging
Authentication:
- Requires valid authentication token
- Respects user permissions and RBAC
"""
try:
await self._ensure_services_initialized()
users = await self.user_service.list_users()
@@ -364,7 +655,31 @@ class PortainerCoreMCPServer:
return f"Failed to get users: {str(e)}"
async def _get_settings_resource(self) -> str:
"""Get settings resource."""
"""
Get settings resource data.
Retrieves the current Portainer settings for the settings resource.
This method is called when the portainer://settings resource is accessed.
Returns:
str: JSON-formatted settings data or error message
Raises:
PortainerAuthenticationError: If authentication is required but missing
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - Settings data is typically small and constant
Flow:
1. Ensure SettingsService is initialized
2. Call settings_service.get_settings() to fetch data
3. Convert result to string format
4. Handle errors gracefully with logging
Authentication:
- Requires valid authentication token
- Admin privileges may be required for some settings
"""
try:
await self._ensure_services_initialized()
settings = await self.settings_service.get_settings()
@@ -374,11 +689,57 @@ class PortainerCoreMCPServer:
return f"Failed to get settings: {str(e)}"
async def _handle_health_check(self) -> str:
"""Handle health check tool call."""
"""
Handle health check tool execution.
Wrapper method that delegates to _get_health_status() for consistency
between the health resource and health_check tool.
Returns:
str: JSON-formatted health status report
Complexity: O(s) where s is the number of services to check
Note:
This is a convenience method to maintain consistency between
resource access and tool execution for health monitoring.
"""
return await self._get_health_status()
async def _handle_authenticate(self, arguments: Dict[str, Any]) -> str:
"""Handle authentication tool call."""
"""
Handle user authentication tool execution.
Authenticates a user with username and password, returning a JWT token
for subsequent API calls.
Args:
arguments: Dictionary containing:
- username (str): Username for authentication
- password (str): Password for authentication
Returns:
str: JSON-formatted authentication result with JWT token
Raises:
PortainerValidationError: If username/password are missing
PortainerAuthenticationError: If credentials are invalid
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - Authentication is a constant time operation
Flow:
1. Ensure AuthService is initialized
2. Extract username and password from arguments
3. Call auth_service.login() with credentials
4. Log successful authentication (username only)
5. Return formatted authentication result
Security:
- Passwords are not logged
- Failed attempts are logged for security monitoring
- JWT tokens are returned securely
"""
try:
await self._ensure_services_initialized()
username = arguments.get("username")
@@ -392,7 +753,39 @@ class PortainerCoreMCPServer:
return f"Authentication failed: {str(e)}"
async def _handle_generate_token(self, arguments: Dict[str, Any]) -> str:
"""Handle token generation tool call."""
"""
Handle API token generation tool execution.
Generates a new API token for a specified user that can be used for
programmatic access to the Portainer API.
Args:
arguments: Dictionary containing:
- user_id (int): User ID to generate token for
- description (str, optional): Token description (default: 'MCP Server Token')
Returns:
str: JSON-formatted token generation result with API token
Raises:
PortainerValidationError: If user_id is missing or invalid
PortainerAuthenticationError: If not authorized to generate tokens
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - Token generation is a constant time operation
Flow:
1. Ensure AuthService is initialized
2. Extract user_id and optional description from arguments
3. Call auth_service.generate_api_token() with parameters
4. Log successful token generation (user_id only)
5. Return formatted token result
Security:
- Tokens are generated with appropriate permissions
- Token creation is logged for audit purposes
- Only authorized users can generate tokens
"""
try:
await self._ensure_services_initialized()
user_id = arguments.get("user_id")
@@ -406,7 +799,33 @@ class PortainerCoreMCPServer:
return f"Token generation failed: {str(e)}"
async def _handle_get_current_user(self, arguments: Dict[str, Any]) -> str:
"""Handle get current user tool call."""
"""
Handle get current user tool execution.
Retrieves information about the currently authenticated user based on
the authentication token in the request context.
Args:
arguments: Dictionary (empty - no arguments required)
Returns:
str: JSON-formatted current user information
Raises:
PortainerAuthenticationError: If authentication token is missing/invalid
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - User lookup is a constant time operation
Flow:
1. Ensure AuthService is initialized
2. Call auth_service.get_current_user() with current token
3. Return formatted user information
Authentication:
- Requires valid authentication token
- Returns user data based on token context
"""
try:
await self._ensure_services_initialized()
result = await self.auth_service.get_current_user()
@@ -416,7 +835,35 @@ class PortainerCoreMCPServer:
return f"Get current user failed: {str(e)}"
async def _handle_list_users(self, arguments: Dict[str, Any]) -> str:
"""Handle list users tool call."""
"""
Handle list users tool execution.
Retrieves a list of all users in the Portainer instance. The results
are filtered based on the current user's permissions.
Args:
arguments: Dictionary (empty - no arguments required)
Returns:
str: JSON-formatted list of users
Raises:
PortainerAuthenticationError: If authentication is required but missing
PortainerAuthorizationError: If user lacks permission to list users
PortainerNetworkError: If communication with Portainer fails
Complexity: O(n) where n is the number of users in the system
Flow:
1. Ensure UserService is initialized
2. Call user_service.list_users() to fetch all users
3. Return formatted user list
Authorization:
- Requires authentication
- May require admin privileges depending on configuration
- Results filtered by user permissions
"""
try:
await self._ensure_services_initialized()
result = await self.user_service.list_users()
@@ -426,7 +873,41 @@ class PortainerCoreMCPServer:
return f"List users failed: {str(e)}"
async def _handle_create_user(self, arguments: Dict[str, Any]) -> str:
"""Handle create user tool call."""
"""
Handle create user tool execution.
Creates a new user in the Portainer instance with the specified
username, password, and role.
Args:
arguments: Dictionary containing:
- username (str): Username for the new user
- password (str): Password for the new user
- role (int): Role ID (1=Admin, 2=User)
Returns:
str: JSON-formatted created user information
Raises:
PortainerValidationError: If required fields are missing or invalid
PortainerAuthenticationError: If authentication is required but missing
PortainerAuthorizationError: If user lacks permission to create users
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - User creation is a constant time operation
Flow:
1. Ensure UserService is initialized
2. Extract username, password, and role from arguments
3. Call user_service.create_user() with parameters
4. Log successful user creation (username only)
5. Return formatted user information
Security:
- Passwords are validated but not logged
- User creation is logged for audit purposes
- Role assignments are validated
"""
try:
await self._ensure_services_initialized()
username = arguments.get("username")
@@ -441,7 +922,42 @@ class PortainerCoreMCPServer:
return f"User creation failed: {str(e)}"
async def _handle_update_user(self, arguments: Dict[str, Any]) -> str:
"""Handle update user tool call."""
"""
Handle update user tool execution.
Updates an existing user's information including username, password,
and role. Only provided fields are updated (partial updates supported).
Args:
arguments: Dictionary containing:
- user_id (int): User ID to update (required)
- username (str, optional): New username
- password (str, optional): New password
- role (int, optional): New role ID
Returns:
str: JSON-formatted updated user information
Raises:
PortainerValidationError: If user_id is missing or invalid
PortainerAuthenticationError: If authentication is required but missing
PortainerAuthorizationError: If user lacks permission to update users
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - User update is a constant time operation
Flow:
1. Ensure UserService is initialized
2. Extract user_id and optional fields from arguments
3. Call user_service.update_user() with parameters
4. Log successful user update (user_id only)
5. Return formatted user information
Security:
- Passwords are validated but not logged
- User updates are logged for audit purposes
- Role changes are validated and logged
"""
try:
await self._ensure_services_initialized()
user_id = arguments.get("user_id")
@@ -457,7 +973,39 @@ class PortainerCoreMCPServer:
return f"User update failed: {str(e)}"
async def _handle_delete_user(self, arguments: Dict[str, Any]) -> str:
"""Handle delete user tool call."""
"""
Handle delete user tool execution.
Deletes a user from the Portainer instance. This operation is
irreversible and removes all user data and permissions.
Args:
arguments: Dictionary containing:
- user_id (int): User ID to delete
Returns:
str: Success message confirming user deletion
Raises:
PortainerValidationError: If user_id is missing or invalid
PortainerAuthenticationError: If authentication is required but missing
PortainerAuthorizationError: If user lacks permission to delete users
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - User deletion is a constant time operation
Flow:
1. Ensure UserService is initialized
2. Extract user_id from arguments
3. Call user_service.delete_user() with user_id
4. Log successful user deletion
5. Return success confirmation message
Security:
- User deletion is logged for audit purposes
- Prevents self-deletion in service layer
- Requires appropriate permissions
"""
try:
await self._ensure_services_initialized()
user_id = arguments.get("user_id")
@@ -470,7 +1018,34 @@ class PortainerCoreMCPServer:
return f"User deletion failed: {str(e)}"
async def _handle_get_settings(self, arguments: Dict[str, Any]) -> str:
"""Handle get settings tool call."""
"""
Handle get settings tool execution.
Retrieves the current Portainer instance settings including
authentication, templates, and other configuration options.
Args:
arguments: Dictionary (empty - no arguments required)
Returns:
str: JSON-formatted settings data
Raises:
PortainerAuthenticationError: If authentication is required but missing
PortainerAuthorizationError: If user lacks permission to view settings
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - Settings retrieval is a constant time operation
Flow:
1. Ensure SettingsService is initialized
2. Call settings_service.get_settings() to fetch current settings
3. Return formatted settings data
Authorization:
- Requires authentication
- May require admin privileges for sensitive settings
"""
try:
await self._ensure_services_initialized()
result = await self.settings_service.get_settings()
@@ -480,7 +1055,39 @@ class PortainerCoreMCPServer:
return f"Get settings failed: {str(e)}"
async def _handle_update_settings(self, arguments: Dict[str, Any]) -> str:
"""Handle update settings tool call."""
"""
Handle update settings tool execution.
Updates Portainer instance settings with new configuration values.
Only provided settings are updated (partial updates supported).
Args:
arguments: Dictionary containing:
- settings (dict): Settings to update as key-value pairs
Returns:
str: JSON-formatted updated settings data
Raises:
PortainerValidationError: If settings format is invalid
PortainerAuthenticationError: If authentication is required but missing
PortainerAuthorizationError: If user lacks permission to update settings
PortainerNetworkError: If communication with Portainer fails
Complexity: O(1) - Settings update is a constant time operation
Flow:
1. Ensure SettingsService is initialized
2. Extract settings dictionary from arguments
3. Call settings_service.update_settings() with new values
4. Log successful settings update
5. Return formatted updated settings
Security:
- Settings updates are logged for audit purposes
- Sensitive settings may require additional validation
- Only authorized users can modify settings
"""
try:
await self._ensure_services_initialized()
settings = arguments.get("settings")
@@ -493,7 +1100,39 @@ class PortainerCoreMCPServer:
return f"Settings update failed: {str(e)}"
async def run(self) -> None:
"""Run the MCP server."""
"""
Run the MCP server with full lifecycle management.
Starts the MCP server and handles the complete lifecycle including
service initialization, MCP protocol communication, and graceful shutdown.
The server runs until interrupted by user or system signal.
Lifecycle Flow:
1. Initialize all required services
2. Set up MCP protocol communication streams
3. Start server with initialization options
4. Handle requests until shutdown
5. Clean up resources and close connections
Raises:
PortainerConfigurationError: If configuration is invalid
PortainerError: If server fails to start
Complexity: O(∞) - Server runs indefinitely until shutdown
Error Handling:
- Service initialization failures prevent startup
- Communication errors are logged and handled
- Graceful shutdown on exceptions
- Resource cleanup is guaranteed
Side Effects:
- Initializes all services
- Establishes MCP protocol streams
- Starts background request processing
- Maintains active connections
"""
logger.info("Starting Portainer Core MCP Server")
try:
@@ -519,7 +1158,31 @@ class PortainerCoreMCPServer:
await self._cleanup_services()
async def _cleanup_services(self) -> None:
"""Clean up service resources."""
"""
Clean up service resources during shutdown.
Properly shuts down all services and releases their resources including
HTTP connections, authentication tokens, and any background tasks.
Cleanup Process:
1. Call cleanup() on AuthService if initialized
2. Call cleanup() on UserService if initialized
3. Call cleanup() on SettingsService if initialized
4. Each service handles its own resource cleanup
Complexity: O(s) where s is the number of initialized services
Error Handling:
- Service cleanup errors are logged but don't prevent other cleanups
- Cleanup is attempted for all services regardless of individual failures
- Resources are released even if cleanup methods fail
Side Effects:
- Closes HTTP client connections
- Invalidates authentication tokens
- Stops background tasks
- Releases system resources
"""
if self.auth_service:
await self.auth_service.cleanup()
if self.user_service:
@@ -529,12 +1192,64 @@ class PortainerCoreMCPServer:
def create_server() -> PortainerCoreMCPServer:
"""Create and return a configured MCP server instance."""
"""
Create and return a configured MCP server instance.
Factory function that creates a new PortainerCoreMCPServer instance with
default configuration from environment variables.
Returns:
PortainerCoreMCPServer: Configured server instance ready to run
Raises:
PortainerConfigurationError: If required environment variables are missing
Complexity: O(1) - Server creation is constant time
Usage:
```python
server = create_server()
await server.run()
```
Note:
This is the preferred way to create server instances as it ensures
proper configuration and initialization order.
"""
return PortainerCoreMCPServer()
async def main() -> None:
"""Main entry point for the MCP server."""
"""
Main entry point for the MCP server.
Async main function that handles server creation, execution, and shutdown.
This function is called by the run_server.py script or when the module
is run directly.
Lifecycle:
1. Create server instance using factory function
2. Run server until completion or interruption
3. Handle graceful shutdown on KeyboardInterrupt
4. Log errors and re-raise for proper exit codes
Raises:
PortainerError: If server fails to start or run
KeyboardInterrupt: Re-raised after logging for proper shutdown
Complexity: O(∞) - Runs until shutdown
Error Handling:
- KeyboardInterrupt: Logged as user-initiated shutdown
- Other exceptions: Logged as server failures and re-raised
- Ensures proper exit codes for process management
Usage:
```python
import asyncio
asyncio.run(main())
```
"""
try:
server = create_server()
await server.run()