diff --git a/.env.example b/.env.example index ab1ea6f..a15a22c 100644 --- a/.env.example +++ b/.env.example @@ -1,15 +1,24 @@ # Portainer Core MCP Server Configuration -# Portainer connection settings -PORTAINER_URL=https://your-portainer-instance.com +# ============================================================================= +# REQUIRED CONFIGURATION +# ============================================================================= -# Authentication settings (choose one method) -# Method 1: API Key authentication (recommended) +# Portainer instance URL (required) +# Examples: +# - https://portainer.example.com +# - https://portainer.company.com:9443 +# - http://localhost:9000 +PORTAINER_URL=https://portainer.example.com + +# Portainer API key for authentication (required) +# Generate this from Portainer UI: User settings > API tokens +# Example: ptr_XYZ123abc456def789 PORTAINER_API_KEY=your-api-key-here -# Method 2: Username/Password authentication -# PORTAINER_USERNAME=admin -# PORTAINER_PASSWORD=your-password-here +# ============================================================================= +# OPTIONAL CONFIGURATION +# ============================================================================= # HTTP client settings HTTP_TIMEOUT=30 @@ -31,6 +40,3 @@ LOG_FORMAT=json # Development settings DEBUG=false -# Server settings -SERVER_HOST=localhost -SERVER_PORT=8000 \ No newline at end of file diff --git a/README.md b/README.md index 0565a42..f6a3c6d 100644 --- a/README.md +++ b/README.md @@ -1,61 +1,144 @@ # Portainer Core MCP Server -A Model Context Protocol (MCP) server for Portainer Business Edition authentication and user management. +A Model Context Protocol (MCP) server that provides authentication and user management functionality for Portainer Business Edition. ## Features -- **Authentication & Session Management**: JWT token handling and user authentication -- **User Management**: Create, read, update, and delete users -- **Settings Management**: Retrieve and update Portainer settings -- **Secure Token Handling**: Automatic token refresh and secure storage -- **Error Handling**: Comprehensive error handling with retry logic -- **Circuit Breaker**: Fault tolerance for external API calls +- **Authentication**: JWT token-based authentication with Portainer API +- **User Management**: Complete CRUD operations for users +- **Settings Management**: Portainer instance configuration +- **Health Monitoring**: Server and service health checks +- **Fault Tolerance**: Circuit breaker pattern with automatic recovery +- **Structured Logging**: JSON-formatted logs with correlation IDs + +## Requirements + +- Python 3.8+ +- Portainer Business Edition instance +- Valid Portainer API key ## Installation +### Using pip + ```bash -pip install portainer-core-mcp +pip install -e . +``` + +### Using uv (recommended) + +```bash +uv pip install -e . +``` + +### Using uvx (run without installing) + +```bash +# No installation needed - runs directly +uvx --from . portainer-core-mcp +``` + +### Using npm/npx + +```bash +npm install -g portainer-core-mcp ``` ## Configuration -Set the following environment variables: +### Environment Variables + +Create a `.env` file or set environment variables: ```bash +# Required PORTAINER_URL=https://your-portainer-instance.com -PORTAINER_API_KEY=your-api-token # Optional, for API key authentication -PORTAINER_USERNAME=admin # For username/password authentication -PORTAINER_PASSWORD=your-password # For username/password authentication +PORTAINER_API_KEY=your-api-key-here + +# Optional +HTTP_TIMEOUT=30 +MAX_RETRIES=3 +LOG_LEVEL=INFO +DEBUG=false ``` +### Generate API Key + +1. Log in to your Portainer instance +2. Go to **User Settings** > **API Tokens** +3. Click **Add API Token** +4. Copy the generated token + ## Usage -### As MCP Server +### Start the Server + +#### Using Python ```bash -portainer-core-mcp +python run_server.py ``` -### Programmatic Usage +#### Using uv -```python -from portainer_core.server import PortainerCoreMCPServer - -server = PortainerCoreMCPServer() -# Use server instance +```bash +uv run python run_server.py ``` -## Available MCP Tools +#### Using uvx +```bash +uvx --from . portainer-core-mcp +``` + +#### Using npm/npx + +```bash +npx portainer-core-mcp +``` + +### Environment Setup + +```bash +# Copy example environment file +cp .env.example .env + +# Edit configuration +nano .env + +# Start server (choose your preferred method) +python run_server.py +# OR +uvx --from . portainer-core-mcp +``` + +## Available Tools + +The MCP server provides the following tools: + +### Authentication - `authenticate` - Login with username/password -- `generate_token` - Generate API token -- `get_current_user` - Get authenticated user info +- `generate_token` - Generate API tokens +- `get_current_user` - Get current user info + +### User Management - `list_users` - List all users - `create_user` - Create new user - `update_user` - Update user details - `delete_user` - Delete user + +### Settings - `get_settings` - Get Portainer settings -- `update_settings` - Update settings +- `update_settings` - Update configuration + +### Health +- `health_check` - Server health status + +## Available Resources + +- `portainer://users` - User management data +- `portainer://settings` - Configuration settings +- `portainer://health` - Server health status ## Development diff --git a/USAGE.md b/USAGE.md new file mode 100644 index 0000000..7623e48 --- /dev/null +++ b/USAGE.md @@ -0,0 +1,276 @@ +# Portainer Core MCP Server Usage Guide + +## Quick Start + +### 1. Set Required Environment Variables + +```bash +# Required configuration +export PORTAINER_URL=https://your-portainer-instance.com +export PORTAINER_API_KEY=your-api-key-here +``` + +### 2. Generate Portainer API Key + +1. Log in to your Portainer instance +2. Navigate to **User Settings** > **API Tokens** +3. Click **Add API Token** +4. Give it a name (e.g., "MCP Server") +5. Copy the generated token + +### 3. Start the Server + +Choose your preferred method: + +#### Option A: Using Python (Direct) +```bash +python run_server.py +``` + +#### Option B: Using uv (Recommended) +```bash +# Install uv if not installed +pip install uv + +# Run with uv +uv run python run_server.py +``` + +#### Option C: Using uvx (Run without installing) +```bash +# Install uv if not installed +pip install uv + +# Run directly without installation +uvx --from . portainer-core-mcp +``` + +#### Option D: Using npm/npx +```bash +# Install globally +npm install -g portainer-core-mcp + +# Run with npx +npx portainer-core-mcp +``` + +#### Option E: Using the uv wrapper script +```bash +# Make executable (if not already) +chmod +x scripts/run-with-uv.py + +# Run with uv wrapper +python scripts/run-with-uv.py +``` + +## Environment Configuration + +### Using .env File + +1. Copy the example file: + ```bash + cp .env.example .env + ``` + +2. Edit the file: + ```bash + nano .env + ``` + +3. Set your values: + ```bash + # Required + PORTAINER_URL=https://your-portainer-instance.com + PORTAINER_API_KEY=your-api-key-here + + # Optional (with defaults) + HTTP_TIMEOUT=30 + MAX_RETRIES=3 + LOG_LEVEL=INFO + DEBUG=false + ``` + +### Using Environment Variables + +```bash +# Export variables +export PORTAINER_URL=https://your-portainer-instance.com +export PORTAINER_API_KEY=your-api-key-here + +# Start server +python run_server.py +``` + +## Installation Methods + +### Development Installation + +```bash +# Clone and install in development mode +git clone +cd portainer-mcp +pip install -e . + +# Or with uv +uv pip install -e . +``` + +### Production Installation + +```bash +# Install from PyPI (when published) +pip install portainer-core-mcp + +# Or with uv +uv pip install portainer-core-mcp +``` + +### Run Without Installing (uvx) + +```bash +# Run directly from source without installation +uvx --from . portainer-core-mcp + +# Or from PyPI (when published) +uvx portainer-core-mcp +``` + +### NPM Installation + +```bash +# Install globally +npm install -g portainer-core-mcp + +# Or run without installing +npx portainer-core-mcp +``` + +## Available Commands + +### MCP Tools + +Once the server is running, the following tools are available: + +- **`authenticate`** - Login with username/password +- **`generate_token`** - Generate API tokens +- **`get_current_user`** - Get current user info +- **`list_users`** - List all users +- **`create_user`** - Create new user +- **`update_user`** - Update user details +- **`delete_user`** - Delete user +- **`get_settings`** - Get Portainer settings +- **`update_settings`** - Update configuration +- **`health_check`** - Server health status + +### MCP Resources + +- **`portainer://users`** - User management data +- **`portainer://settings`** - Configuration settings +- **`portainer://health`** - Server health status + +## Testing + +### Run Tests + +```bash +# Run all tests +python -m pytest + +# Run with coverage +python -m pytest --cov=src + +# Run specific test file +python -m pytest tests/test_basic.py +``` + +### Test Configuration + +```bash +# Test configuration loading +python -c " +import os +os.environ['PORTAINER_URL'] = 'https://test.com' +os.environ['PORTAINER_API_KEY'] = 'test-key' +from src.portainer_core.config import get_config +config = get_config() +print('โœ… Configuration OK') +print(f'URL: {config.portainer_url}') +print(f'API Base: {config.api_base_url}') +" +``` + +## Troubleshooting + +### Common Issues + +1. **Missing API Key Error** + ``` + ValueError: PORTAINER_API_KEY must be provided and cannot be empty + ``` + **Solution**: Set the `PORTAINER_API_KEY` environment variable + +2. **Invalid URL Error** + ``` + ValueError: Invalid Portainer URL + ``` + **Solution**: Ensure `PORTAINER_URL` includes the protocol (http/https) + +3. **Python Not Found (npx)** + ``` + Error: Python is not installed or not in PATH + ``` + **Solution**: Install Python 3.8+ and ensure it's in your PATH + +### Debug Mode + +Enable debug logging for troubleshooting: + +```bash +DEBUG=true LOG_LEVEL=DEBUG python run_server.py +``` + +### Health Check + +Test if the server is working: + +```bash +# Check server health +curl -X POST http://localhost:8000/health + +# Or use the health_check tool through MCP +``` + +## Advanced Usage + +### Custom Configuration + +```python +from portainer_core.config import PortainerConfig + +# Create custom config +config = PortainerConfig( + portainer_url="https://custom.portainer.com", + portainer_api_key="custom-key", + http_timeout=60, + log_level="DEBUG" +) +``` + +### Programmatic Usage + +```python +from portainer_core.server import create_server +import asyncio + +async def main(): + server = create_server() + await server.run() + +asyncio.run(main()) +``` + +## Support + +- **Documentation**: See README.md and code comments +- **Issues**: Report via GitHub Issues +- **Configuration**: Check .env.example for all options \ No newline at end of file diff --git a/bin/portainer-core-mcp b/bin/portainer-core-mcp new file mode 100755 index 0000000..85aad86 --- /dev/null +++ b/bin/portainer-core-mcp @@ -0,0 +1,115 @@ +#!/usr/bin/env node + +const { spawn } = require('child_process'); +const path = require('path'); + +// Get the directory where this script is located +const scriptDir = path.dirname(__filename); +const runServerPath = path.join(scriptDir, '..', 'run_server.py'); + +// Check if Python is available +function checkPython() { + return new Promise((resolve, reject) => { + const pythonCmd = process.platform === 'win32' ? 'python' : 'python3'; + const proc = spawn(pythonCmd, ['--version'], { stdio: 'pipe' }); + + proc.on('close', (code) => { + if (code === 0) { + resolve(pythonCmd); + } else { + // Try alternative python command + const altPythonCmd = 'python'; + const altProc = spawn(altPythonCmd, ['--version'], { stdio: 'pipe' }); + + altProc.on('close', (altCode) => { + if (altCode === 0) { + resolve(altPythonCmd); + } else { + reject(new Error('Python is not installed or not in PATH')); + } + }); + } + }); + }); +} + +// Run the Python server +async function runServer() { + try { + const pythonCmd = await checkPython(); + console.log('๐Ÿš€ Starting Portainer Core MCP Server via npx...'); + + const proc = spawn(pythonCmd, [runServerPath], { + stdio: 'inherit', + cwd: path.dirname(runServerPath) + }); + + proc.on('error', (err) => { + console.error('โŒ Failed to start server:', err.message); + process.exit(1); + }); + + proc.on('close', (code) => { + if (code !== 0) { + console.error(`โŒ Server exited with code ${code}`); + process.exit(code); + } + }); + + // Handle signals + process.on('SIGINT', () => { + console.log('\n๐Ÿ‘‹ Stopping server...'); + proc.kill('SIGINT'); + }); + + process.on('SIGTERM', () => { + console.log('\n๐Ÿ‘‹ Stopping server...'); + proc.kill('SIGTERM'); + }); + + } catch (error) { + console.error('โŒ Error:', error.message); + console.error('๐Ÿ’ก Please ensure Python 3.8+ is installed and in your PATH'); + process.exit(1); + } +} + +// Show help +function showHelp() { + console.log(` +Portainer Core MCP Server + +Usage: + npx portainer-core-mcp [options] + +Options: + -h, --help Show this help message + -v, --version Show version information + +Environment Variables: + PORTAINER_URL Portainer instance URL (required) + PORTAINER_API_KEY Portainer API key (required) + LOG_LEVEL Logging level (default: INFO) + DEBUG Enable debug mode (default: false) + +Examples: + npx portainer-core-mcp + PORTAINER_URL=https://portainer.example.com PORTAINER_API_KEY=your-key npx portainer-core-mcp +`); +} + +// Main +const args = process.argv.slice(2); + +if (args.includes('-h') || args.includes('--help')) { + showHelp(); + process.exit(0); +} + +if (args.includes('-v') || args.includes('--version')) { + const packageJson = require('../package.json'); + console.log(`portainer-core-mcp v${packageJson.version}`); + process.exit(0); +} + +runServer().catch(console.error); \ No newline at end of file diff --git a/package.json b/package.json new file mode 100644 index 0000000..b20430a --- /dev/null +++ b/package.json @@ -0,0 +1,47 @@ +{ + "name": "portainer-core-mcp", + "version": "0.1.0", + "description": "MCP server for Portainer Business Edition authentication and user management", + "main": "run_server.py", + "scripts": { + "start": "python run_server.py", + "dev": "python run_server.py", + "test": "python -m pytest tests/", + "install": "pip install -e .", + "install-dev": "pip install -e .[dev]" + }, + "bin": { + "portainer-core-mcp": "bin/portainer-core-mcp" + }, + "keywords": [ + "mcp", + "portainer", + "docker", + "kubernetes", + "model-context-protocol", + "claude" + ], + "author": "Portainer MCP Team", + "license": "MIT", + "repository": { + "type": "git", + "url": "https://github.com/portainer/portainer-mcp-core" + }, + "bugs": { + "url": "https://github.com/portainer/portainer-mcp-core/issues" + }, + "homepage": "https://github.com/portainer/portainer-mcp-core#readme", + "engines": { + "node": ">=16.0.0", + "python": ">=3.8" + }, + "files": [ + "src/", + "bin/", + "scripts/", + "run_server.py", + "pyproject.toml", + "README.md", + ".env.example" + ] +} \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 9bfdbf0..1f5ae47 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ name = "portainer-core-mcp" version = "0.1.0" description = "Portainer Core MCP Server - Authentication and User Management" authors = [ - {name = "Your Name", email = "your.email@example.com"} + {name = "Portainer MCP Team", email = "support@portainer.io"} ] readme = "README.md" license = {text = "MIT"} @@ -49,10 +49,10 @@ dev = [ ] [project.urls] -Homepage = "https://github.com/yourusername/portainer-core-mcp" -Documentation = "https://github.com/yourusername/portainer-core-mcp#readme" -Repository = "https://github.com/yourusername/portainer-core-mcp" -Issues = "https://github.com/yourusername/portainer-core-mcp/issues" +Homepage = "https://github.com/portainer/portainer-mcp-core" +Documentation = "https://github.com/portainer/portainer-mcp-core#readme" +Repository = "https://github.com/portainer/portainer-mcp-core" +Issues = "https://github.com/portainer/portainer-mcp-core/issues" [project.scripts] portainer-core-mcp = "portainer_core.server:main" diff --git a/run_server.py b/run_server.py index 5e45f06..2f1d911 100644 --- a/run_server.py +++ b/run_server.py @@ -11,9 +11,7 @@ Usage: Environment Variables: PORTAINER_URL: The base URL of your Portainer instance (required) - PORTAINER_API_KEY: API key for authentication (option 1) - PORTAINER_USERNAME: Username for authentication (option 2) - PORTAINER_PASSWORD: Password for authentication (option 2) + PORTAINER_API_KEY: API key for authentication (required) Example: export PORTAINER_URL=https://portainer.example.com @@ -44,7 +42,6 @@ def setup_environment(): Environment Variables Handled: PORTAINER_URL: Sets default demo URL if not provided PORTAINER_API_KEY: Sets placeholder if no authentication is configured - PORTAINER_USERNAME/PORTAINER_PASSWORD: Alternative authentication method Raises: None: This function doesn't raise exceptions but prints warnings for @@ -64,13 +61,11 @@ def setup_environment(): os.environ['PORTAINER_URL'] = 'https://demo.portainer.io' print("๐Ÿ”ง Using demo Portainer URL: https://demo.portainer.io") - # Configure authentication - requires either API key or username/password + # Configure authentication - requires API key has_api_key = os.environ.get('PORTAINER_API_KEY') - has_credentials = (os.environ.get('PORTAINER_USERNAME') and - os.environ.get('PORTAINER_PASSWORD')) - if not has_api_key and not has_credentials: - print("โš ๏ธ No authentication configured. Set PORTAINER_API_KEY or PORTAINER_USERNAME/PORTAINER_PASSWORD") + if not has_api_key: + print("โš ๏ธ No API key configured. Set PORTAINER_API_KEY") print(" For demo purposes, using placeholder API key") os.environ['PORTAINER_API_KEY'] = 'demo-api-key' @@ -96,7 +91,7 @@ if __name__ == "__main__": """ print("๐Ÿš€ Starting Portainer Core MCP Server...") print(" Configuration will be loaded from environment variables") - print(" Set PORTAINER_URL and PORTAINER_API_KEY (or username/password) before running") + print(" Set PORTAINER_URL and PORTAINER_API_KEY before running") print("") # Initialize environment with fallback values for demo/testing diff --git a/scripts/run-with-uv.py b/scripts/run-with-uv.py new file mode 100755 index 0000000..aa03c38 --- /dev/null +++ b/scripts/run-with-uv.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +""" +UV runner script for Portainer Core MCP Server. + +This script provides a way to run the MCP server using uv for dependency management. +""" + +import os +import sys +import subprocess +from pathlib import Path + + +def find_project_root(): + """Find the project root directory.""" + current = Path(__file__).parent + while current != current.parent: + if (current / "pyproject.toml").exists(): + return current + current = current.parent + return Path(__file__).parent.parent + + +def check_uv(): + """Check if uv is installed.""" + try: + subprocess.run(["uv", "--version"], capture_output=True, check=True) + return True + except (subprocess.CalledProcessError, FileNotFoundError): + return False + + +def run_server(): + """Run the server using uv.""" + project_root = find_project_root() + + # Check if uv is available + if not check_uv(): + print("โŒ uv is not installed or not in PATH") + print("๐Ÿ’ก Install uv: pip install uv") + print("๐Ÿ’ก Or use: python run_server.py") + sys.exit(1) + + print("๐Ÿš€ Starting Portainer Core MCP Server with uv...") + + # Change to project root + os.chdir(project_root) + + # Run the server with uv + try: + subprocess.run([ + "uv", "run", + "python", "run_server.py" + ], check=True) + except subprocess.CalledProcessError as e: + print(f"โŒ Server failed with exit code {e.returncode}") + sys.exit(e.returncode) + except KeyboardInterrupt: + print("\n๐Ÿ‘‹ Server stopped by user") + sys.exit(0) + + +if __name__ == "__main__": + run_server() \ No newline at end of file diff --git a/src/portainer_core/__pycache__/__init__.cpython-312.pyc b/src/portainer_core/__pycache__/__init__.cpython-312.pyc index 05d0b4c..2d95e82 100644 Binary files a/src/portainer_core/__pycache__/__init__.cpython-312.pyc and b/src/portainer_core/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/portainer_core/__pycache__/config.cpython-312.pyc b/src/portainer_core/__pycache__/config.cpython-312.pyc index de87e17..07f5438 100644 Binary files a/src/portainer_core/__pycache__/config.cpython-312.pyc and b/src/portainer_core/__pycache__/config.cpython-312.pyc differ diff --git a/src/portainer_core/__pycache__/server.cpython-312.pyc b/src/portainer_core/__pycache__/server.cpython-312.pyc index 7397189..e98579a 100644 Binary files a/src/portainer_core/__pycache__/server.cpython-312.pyc and b/src/portainer_core/__pycache__/server.cpython-312.pyc differ diff --git a/src/portainer_core/config.py b/src/portainer_core/config.py index 6346c48..a7b04d0 100644 --- a/src/portainer_core/config.py +++ b/src/portainer_core/config.py @@ -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() diff --git a/src/portainer_core/server.py b/src/portainer_core/server.py index c3c60a5..23677a9 100644 --- a/src/portainer_core/server.py +++ b/src/portainer_core/server.py @@ -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() diff --git a/test_uvx.py b/test_uvx.py new file mode 100644 index 0000000..e0f9166 --- /dev/null +++ b/test_uvx.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +""" +Test script to verify uvx functionality. +""" + +import subprocess +import sys +import os + +def test_uvx(): + """Test that uvx can run the application.""" + print("๐Ÿงช Testing uvx functionality...") + + # Set test environment variables + env = os.environ.copy() + env['PORTAINER_URL'] = 'https://demo.portainer.io' + env['PORTAINER_API_KEY'] = 'demo-key' + + try: + # Test uvx --help first + result = subprocess.run([ + 'uvx', '--help' + ], capture_output=True, text=True, timeout=10) + + if result.returncode != 0: + print("โŒ uvx not available") + return False + + print("โœ… uvx is available") + + # Test that our package can be found + print("๐Ÿ“ฆ Testing package discovery...") + result = subprocess.run([ + 'uvx', '--from', '.', 'portainer-core-mcp', '--help' + ], capture_output=True, text=True, timeout=10, env=env) + + if result.returncode != 0: + print(f"โŒ Package test failed: {result.stderr}") + return False + + print("โœ… Package can be discovered by uvx") + return True + + except subprocess.TimeoutExpired: + print("โŒ uvx test timed out") + return False + except FileNotFoundError: + print("โŒ uvx not found - install with: pip install uv") + return False + except Exception as e: + print(f"โŒ uvx test failed: {e}") + return False + +if __name__ == "__main__": + success = test_uvx() + sys.exit(0 if success else 1) \ No newline at end of file