diff --git a/README_MERGED.md b/README_CORE.md similarity index 91% rename from README_MERGED.md rename to README_CORE.md index 7120026..e24af19 100644 --- a/README_MERGED.md +++ b/README_CORE.md @@ -1,6 +1,6 @@ -# Portainer Unified MCP Server +# Portainer Core MCP Server -This MCP server combines **portainer-core** and **portainer-teams** functionality into a single unified server, providing comprehensive user, team, and RBAC management for Portainer Business Edition. +This MCP server provides core Portainer functionality by combining user management and teams/RBAC features into a single unified server. It offers comprehensive user, team, and role-based access control management for Portainer Business Edition. ## Features @@ -54,7 +54,7 @@ cd /Users/adelorenzo/repos/portainer-mcp "mcpServers": { "portainer": { "command": "/path/to/.venv/bin/python", - "args": ["/path/to/merged_mcp_server.py"], + "args": ["/path/to/portainer_core_server.py"], "env": { "PORTAINER_URL": "https://your-portainer-instance.com", "PORTAINER_API_KEY": "your-api-key", diff --git a/create_nginx_stack.py b/create_nginx_stack.py deleted file mode 100644 index 7d79f27..0000000 --- a/create_nginx_stack.py +++ /dev/null @@ -1,131 +0,0 @@ -#!/usr/bin/env python3 -"""Create nginx02 stack from Git repository""" - -import aiohttp -import asyncio -import json -import sys - -# Configuration -PORTAINER_URL = "https://partner.portainer.live" -PORTAINER_API_KEY = "ptr_uMqreULEo44qvuszgG8oZWdjkDx3K9HBXSmjd+F/vDE=" - -# Stack configuration -STACK_NAME = "nginx02" -ENVIRONMENT_ID = 6 # docker03 -REPOSITORY_URL = "https://git.oe74.net/adelorenzo/portainer-yaml" -REPOSITORY_REF = "main" # or master -COMPOSE_PATH = "nginx-cmpose.yaml" -GIT_USERNAME = "adelorenzo" -GIT_PASSWORD = "dimi2014" - -def create_stack_from_git(): - """Create a stack from Git repository""" - - # Build request data - data = { - "Name": STACK_NAME, - "EndpointId": ENVIRONMENT_ID, - "GitConfig": { - "URL": REPOSITORY_URL, - "ReferenceName": REPOSITORY_REF, - "ComposeFilePathInRepository": COMPOSE_PATH, - "Authentication": { - "Username": GIT_USERNAME, - "Password": GIT_PASSWORD - } - } - } - - # Headers - headers = { - "X-API-Key": PORTAINER_API_KEY, - "Content-Type": "application/json" - } - - # API endpoint - url = f"{PORTAINER_URL}/api/stacks" - - print(f"Creating stack '{STACK_NAME}' from Git repository...") - print(f"Repository: {REPOSITORY_URL}") - print(f"Compose file: {COMPOSE_PATH}") - print(f"Environment ID: {ENVIRONMENT_ID}") - - try: - response = requests.post(url, json=data, headers=headers) - - if response.status_code == 200 or response.status_code == 201: - result = response.json() - print(f"\nโœ… Stack created successfully!") - print(f"Stack ID: {result['Id']}") - print(f"Stack Name: {result['Name']}") - return result - else: - print(f"\nโŒ Error creating stack: {response.status_code}") - print(f"Response: {response.text}") - - # Try to parse error message - try: - error_data = response.json() - if "message" in error_data: - print(f"Error message: {error_data['message']}") - elif "details" in error_data: - print(f"Error details: {error_data['details']}") - except: - pass - - except Exception as e: - print(f"\nโŒ Exception occurred: {str(e)}") - return None - -def list_existing_stacks(): - """List existing stacks to check for references""" - - headers = { - "X-API-Key": PORTAINER_API_KEY - } - - url = f"{PORTAINER_URL}/api/stacks" - - try: - response = requests.get(url, headers=headers) - if response.status_code == 200: - stacks = response.json() - print("\n๐Ÿ“š Existing stacks:") - for stack in stacks: - if stack.get("EndpointId") == ENVIRONMENT_ID: - print(f" - {stack['Name']} (ID: {stack['Id']})") - if stack.get("GitConfig"): - print(f" Git: {stack['GitConfig']['URL']}") - print(f" Path: {stack['GitConfig'].get('ComposeFilePathInRepository', 'N/A')}") - return stacks - else: - print(f"Error listing stacks: {response.status_code}") - return [] - except Exception as e: - print(f"Exception listing stacks: {str(e)}") - return [] - -if __name__ == "__main__": - # First, list existing stacks - print("Checking existing stacks on environment docker03...") - existing_stacks = list_existing_stacks() - - # Check if stack already exists - stack_exists = any(s['Name'] == STACK_NAME and s.get('EndpointId') == ENVIRONMENT_ID for s in existing_stacks) - - if stack_exists: - print(f"\nโš ๏ธ Stack '{STACK_NAME}' already exists on this environment!") - response = input("Do you want to continue anyway? (y/n): ") - if response.lower() != 'y': - print("Aborting...") - sys.exit(0) - - # Create the stack - print("\n" + "="*50) - result = create_stack_from_git() - - if result: - print("\n๐ŸŽ‰ Stack deployment completed!") - else: - print("\n๐Ÿ˜ž Stack deployment failed!") \ No newline at end of file diff --git a/merged_mcp_server.py b/portainer_core_server.py similarity index 95% rename from merged_mcp_server.py rename to portainer_core_server.py index 455c2c8..be608c5 100755 --- a/merged_mcp_server.py +++ b/portainer_core_server.py @@ -1,5 +1,32 @@ #!/usr/bin/env python3 -"""Merged MCP server for Portainer Core + Teams functionality.""" +""" +Portainer Core MCP Server + +This module provides the core MCP server for Portainer Business Edition, +combining essential user management and teams/RBAC functionality into a +single unified server. + +The server implements comprehensive management capabilities including: +- User authentication and session management +- User CRUD operations with role assignments +- Teams creation and membership management +- Role-based access control (RBAC) configuration +- Resource access controls for fine-grained permissions +- System settings management + +This core server serves as the foundation for Portainer management, +providing the essential identity and access management features that +other specialized servers (Docker, Kubernetes, Edge, etc.) depend on. + +Complexity: O(1) for all operations (simple HTTP requests) +Dependencies: httpx for async HTTP, mcp for server protocol +Call Flow: MCP client -> handle_call_tool() -> make_request() -> Portainer API + +Environment Variables: + PORTAINER_URL: Base URL of Portainer instance + PORTAINER_API_KEY: API key for authentication + MCP_MODE: Set to "true" to suppress logging +""" import os import sys @@ -16,7 +43,7 @@ from mcp.server import NotificationOptions, Server from mcp.server.models import InitializationOptions # Create server -server = Server("portainer-unified") +server = Server("portainer-core") # Store for our state portainer_url = os.getenv("PORTAINER_URL", "https://partner.portainer.live") @@ -590,7 +617,7 @@ async def run(): read_stream, write_stream, InitializationOptions( - server_name="portainer-unified", + server_name="portainer-core", server_version="1.0.0", capabilities=server.get_capabilities( notification_options=NotificationOptions( diff --git a/portainer_docker_server.py b/portainer_docker_server.py index dc62eb0..fd386c3 100755 --- a/portainer_docker_server.py +++ b/portainer_docker_server.py @@ -1203,6 +1203,7 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ return [types.TextContent(type="text", text="Error: environment_id and service_id are required")] # Get current service spec + # Docker Swarm requires the current version for optimistic locking result = await make_request("GET", f"/endpoints/{env_id}/docker/services/{service_id}") if result["status_code"] != 200: return [types.TextContent(type="text", text=f"Failed to get service: HTTP {result['status_code']}")] @@ -1279,6 +1280,7 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ if result["status_code"] == 200 and result["data"]: stacks = result["data"] # Filter stacks for this environment + # Stacks API returns all stacks, need to filter by environment env_stacks = [s for s in stacks if s.get("EndpointId") == env_id] output = f"Found {len(env_stacks)} stack(s) in this environment:\n" @@ -1354,7 +1356,7 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ output += f"- Containers: {info.get('Containers', 0)} ({info.get('ContainersRunning', 0)} running)\n" output += f"- Images: {info.get('Images', 0)}\n" - # Swarm info + # Swarm info if this is a Swarm cluster swarm = info.get('Swarm', {}) if swarm.get('LocalNodeState') == 'active': output += f"\nSwarm Status:\n" diff --git a/portainer_edge_server.py b/portainer_edge_server.py index d5809d2..2a6636a 100644 --- a/portainer_edge_server.py +++ b/portainer_edge_server.py @@ -4,6 +4,36 @@ Portainer Edge MCP Server Provides edge computing functionality through Portainer's API. Manages edge environments, edge stacks, edge groups, and edge jobs. + +This module implements comprehensive edge computing management including: +- Edge environment registration and monitoring +- Edge agent deployment and configuration +- Edge groups for organizing edge devices +- Edge stacks for distributed application deployment +- Edge jobs for remote command execution +- Edge environment status tracking and diagnostics + +The server supports various edge computing scenarios: +- IoT device management at scale +- Distributed application deployment to edge locations +- Remote configuration and updates +- Centralized monitoring of edge infrastructure +- Batch operations across edge device groups + +Complexity: O(n) for list operations where n is number of edge environments +Dependencies: aiohttp for async HTTP, mcp for server protocol +Call Flow: MCP client -> handle_call_tool() -> make_request() -> Portainer API + +Environment Variables: + PORTAINER_URL: Base URL of Portainer instance (required) + PORTAINER_API_KEY: API key for authentication (required) + MCP_MODE: Set to "true" to suppress logging (default: true) + +Edge Concepts: + - Edge Environment: Remote Docker/K8s instance with edge agent + - Edge Group: Logical grouping of edge environments + - Edge Stack: Application deployed to multiple edge environments + - Edge Job: Script/command executed on edge environments """ import os @@ -47,7 +77,36 @@ async def make_request( data: Optional[Any] = None, headers: Optional[dict] = None ) -> dict: - """Make an authenticated request to Portainer API.""" + """ + Make an authenticated request to Portainer API. + + Centralized HTTP request handler for all Portainer Edge API interactions. + Handles authentication, error responses, and timeout management. + + Args: + method: HTTP method (GET, POST, PUT, DELETE) + endpoint: API endpoint path (e.g., /api/edge_groups) + json_data: Optional JSON payload for request body + params: Optional query parameters + data: Optional form data or raw body content + headers: Optional additional headers to merge with defaults + + Returns: + Dict containing response data or error information + Error responses have 'error' key with descriptive message + + Complexity: O(1) - Single HTTP request + + Call Flow: + - Called by: All tool handler functions + - Calls: aiohttp for async HTTP operations + + Error Handling: + - HTTP 4xx/5xx errors return structured error dict + - Timeout errors (30s) return timeout error + - Network errors return connection error + - Parses Portainer error details from response + """ url = f"{PORTAINER_URL}{endpoint}" default_headers = { @@ -95,7 +154,29 @@ async def make_request( return {"error": f"Request failed: {str(e)}"} def format_edge_status(status: int) -> str: - """Format edge environment status with emoji.""" + """ + Format edge environment status with emoji. + + Converts numeric edge status codes to user-friendly strings + with visual indicators for quick status recognition. + + Args: + status: Edge environment status code from API + + Returns: + Formatted status string with emoji indicator + + Complexity: O(1) - Simple lookup + + Call Flow: + - Called by: Edge environment listing functions + - Calls: None (pure function) + + Status Codes: + - 1: Connected (๐Ÿ”ด green) - Agent is online + - 2: Disconnected (๐Ÿ”ด red) - Agent is offline + - Other: Unknown (โ”) - Undefined status + """ if status == 1: return "๐ŸŸข Connected" elif status == 2: diff --git a/portainer_environments_server.py b/portainer_environments_server.py index d63c93a..8c16e60 100755 --- a/portainer_environments_server.py +++ b/portainer_environments_server.py @@ -1,5 +1,32 @@ #!/usr/bin/env python3 -"""MCP server for Portainer Environments management.""" +""" +MCP server for Portainer Environments management. + +This module provides comprehensive environment and endpoint management functionality +for Portainer Business Edition. It handles Docker, Kubernetes, and Edge environments +with support for groups, tags, and team associations. + +The server implements tools for: +- Environment CRUD operations (Docker, Kubernetes, Edge) +- Environment status monitoring and health checks +- Environment groups for logical organization +- Tag management for categorization +- Team access control associations +- Edge agent deployment script generation + +Complexity: O(1) for all operations (simple HTTP requests) +Dependencies: httpx for async HTTP, mcp for server protocol, enum for type constants +Call Flow: MCP client -> handle_call_tool() -> make_request() -> Portainer API + +Environment Variables: + PORTAINER_URL: Base URL of Portainer instance + PORTAINER_API_KEY: API key for authentication + MCP_MODE: Set to "true" to suppress logging + +API Compatibility: + Supports both new (/environments) and legacy (/endpoints) API endpoints + for backward compatibility with older Portainer versions. +""" import os import sys @@ -26,6 +53,19 @@ api_key = os.getenv("PORTAINER_API_KEY", "") # Environment types enum class EnvironmentType(Enum): + """ + Enumeration of supported Portainer environment types. + + Maps environment type names to their numeric API values. + Used when creating new environments to specify the type. + + Values: + DOCKER: Standard Docker environment + DOCKER_SWARM: Docker Swarm cluster + KUBERNETES: Kubernetes cluster + ACI: Azure Container Instances + EDGE_AGENT: Edge computing environment + """ DOCKER = 1 DOCKER_SWARM = 2 KUBERNETES = 3 @@ -35,7 +75,29 @@ class EnvironmentType(Enum): @server.list_tools() async def handle_list_tools() -> list[types.Tool]: - """List available tools.""" + """ + List available tools. + + Returns a comprehensive list of environment management tools. + Each tool includes its name, description, and JSON schema for input validation. + + Returns: + list[types.Tool]: List of available tools with their schemas + + Complexity: O(1) - Returns static list + + Call Flow: + - Called by: MCP protocol during initialization + - Calls: None (static return) + + Tool Categories: + - Basic Operations: list, get, create, update, delete environments + - Status: get_environment_status for health monitoring + - Groups: CRUD operations for environment groups + - Tags: Create and manage environment tags + - Edge: Generate Edge agent deployment scripts + - Access: Team association management + """ return [ # Basic environment operations types.Tool( @@ -363,7 +425,31 @@ async def handle_list_tools() -> list[types.Tool]: async def make_request(method: str, endpoint: str, json_data: Optional[Dict] = None) -> Dict[str, Any]: - """Make HTTP request to Portainer API.""" + """ + Make HTTP request to Portainer API. + + Centralized HTTP request handler that manages all API communication with Portainer. + Handles authentication headers, timeout configuration, and safe JSON parsing. + + Args: + method: HTTP method (GET, POST, PUT, DELETE) + endpoint: API endpoint path (without /api prefix) + json_data: Optional JSON payload for POST/PUT requests + + Returns: + Dict containing status_code, data (parsed JSON), and raw text + + Complexity: O(1) - Single HTTP request + + Call Flow: + - Called by: handle_call_tool() for all API operations + - Calls: httpx for async HTTP requests + + Error Handling: + - 30 second timeout for long-running operations + - Safe JSON parsing with content-type checking + - SSL verification disabled for self-signed certificates + """ import httpx async with httpx.AsyncClient(verify=False, timeout=30.0) as client: @@ -381,6 +467,7 @@ async def make_request(method: str, endpoint: str, json_data: Optional[Dict] = N raise ValueError(f"Unsupported method: {method}") # Parse JSON response safely + # Check content-type header to avoid parsing non-JSON responses try: data = response.json() if response.text and response.headers.get("content-type", "").startswith("application/json") else None except Exception: @@ -390,7 +477,24 @@ async def make_request(method: str, endpoint: str, json_data: Optional[Dict] = N def format_environment_type(env_type: int) -> str: - """Convert environment type ID to readable string.""" + """ + Convert environment type ID to readable string. + + Maps numeric environment type values from the API to human-readable names. + Includes support for local environment types not in the enum. + + Args: + env_type: Numeric environment type from API + + Returns: + Human-readable environment type name + + Complexity: O(1) - Dictionary lookup + + Call Flow: + - Called by: list_environments, get_environment in handle_call_tool() + - Calls: None (pure function) + """ type_map = { 1: "Docker", 2: "Docker Swarm", @@ -404,7 +508,23 @@ def format_environment_type(env_type: int) -> str: def format_environment_status(status: int) -> str: - """Convert environment status to readable string.""" + """ + Convert environment status to readable string. + + Maps numeric status values to human-readable status descriptions. + + Args: + status: Numeric status value from API + + Returns: + Human-readable status ("up", "down", or "unknown") + + Complexity: O(1) - Dictionary lookup + + Call Flow: + - Called by: list_environments, get_environment in handle_call_tool() + - Calls: None (pure function) + """ status_map = { 1: "up", 2: "down" @@ -414,7 +534,32 @@ def format_environment_status(status: int) -> str: @server.call_tool() async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]: - """Handle tool calls.""" + """ + Handle tool calls. + + Main request dispatcher that processes all incoming tool requests from MCP clients. + Routes requests to appropriate Portainer API endpoints and formats responses. + Handles both new (/environments) and legacy (/endpoints) API endpoints. + + Args: + name: The name of the tool to execute + arguments: Optional dictionary of arguments for the tool + + Returns: + list[types.TextContent]: Response messages as text content + + Complexity: O(1) - Each tool performs single API request + + Call Flow: + - Called by: MCP protocol when client invokes a tool + - Calls: make_request() for API operations, format_* functions for display + + Error Handling: + - Validates required arguments before API calls + - Handles both new and legacy API endpoints with fallback + - Specific error messages for timeouts and connection failures + - Detailed traceback for debugging unexpected errors + """ import httpx try: @@ -424,15 +569,18 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ start = arguments.get("start", 0) if arguments else 0 # Try both /environments (new) and /endpoints (old) endpoints + # This dual-endpoint approach ensures compatibility across Portainer versions result = await make_request("GET", f"/environments?limit={limit}&start={start}") if result["status_code"] == 404: # Legacy endpoint doesn't support pagination params in URL + # Fall back to fetching all and paginating manually result = await make_request("GET", "/endpoints") if result["status_code"] == 200 and result["data"] is not None: environments = result["data"] # Handle pagination manually for legacy endpoint + # Legacy API returns array, new API might return paginated object if isinstance(environments, list): total = len(environments) environments = environments[start:start + limit] @@ -441,6 +589,7 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ output = f"Found environments:\n" for env in environments[:10]: # Limit output to first 10 to avoid huge responses + # Format each environment with essential details env_type = format_environment_type(env.get("Type", 0)) status = format_environment_status(env.get("Status", 0)) output += f"\n- ID: {env.get('Id')}, Name: {env.get('Name')}" @@ -516,7 +665,8 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ if arguments.get("tags"): env_data["TagIds"] = arguments["tags"] - # TLS configuration + # TLS configuration for secure connections + # Required when connecting to Docker daemon over TCP with TLS if arguments.get("tls") or arguments.get("tls_skip_verify"): env_data["TLSConfig"] = { "TLS": arguments.get("tls", False), @@ -574,7 +724,8 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ if not env_id: return [types.TextContent(type="text", text="Error: environment_id is required")] - # Get current environment first + # Get current environment first to preserve unchanged fields + # Portainer requires complete environment object for updates result = await make_request("GET", f"/environments/{env_id}") if result["status_code"] == 404: result = await make_request("GET", f"/endpoints/{env_id}") @@ -628,6 +779,7 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ return [types.TextContent(type="text", text="Error: environment_id is required")] # Try to get docker info through Portainer proxy + # This endpoint proxies requests to the actual Docker/Kubernetes API result = await make_request("GET", f"/environments/{env_id}/docker/info") if result["status_code"] == 404: result = await make_request("GET", f"/endpoints/{env_id}/docker/info") @@ -642,7 +794,7 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ output += f"- Stopped: {info.get('ContainersStopped', 0)}\n" output += f"- Images: {info.get('Images', 0)}\n" output += f"- CPU Count: {info.get('NCPU', 0)}\n" - output += f"- Memory: {info.get('MemTotal', 0) / (1024**3):.2f} GB\n" + output += f"- Memory: {info.get('MemTotal', 0) / (1024**3):.2f} GB\n" # Convert bytes to GB return [types.TextContent(type="text", text=output)] else: return [types.TextContent(type="text", text="Environment is down or inaccessible")] @@ -800,7 +952,7 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ output = f"โœ“ Edge environment '{edge_name}' created\n" output += f"- Environment ID: {env_id}\n" output += f"- Edge Key: {edge_key}\n" - output += f"\nDeployment command:\n" + output += f"\nDeployment command:\n" # Provide ready-to-use Docker command output += f"docker run -d --name portainer_edge_agent --restart always \\\n" output += f" -v /var/run/docker.sock:/var/run/docker.sock \\\n" output += f" -v /var/lib/docker/volumes:/var/lib/docker/volumes \\\n" @@ -831,7 +983,26 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ async def run(): - """Run the MCP server.""" + """ + Run the MCP server. + + Sets up the stdio transport and runs the server with configured capabilities. + This is the main async loop that handles all MCP protocol communication. + + The server runs indefinitely until interrupted, processing tool requests + and responses through stdin/stdout streams. + + Complexity: O(1) - Server initialization + + Call Flow: + - Called by: main() via asyncio.run() + - Calls: server.run() with stdio streams + + Configuration: + - Uses stdio transport for Claude Desktop compatibility + - Disables all change notifications (tools are static) + - Server version 1.0.0 indicates stable implementation + """ # Use stdio transport async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( @@ -853,7 +1024,18 @@ async def run(): def main(): - """Main entry point.""" + """ + Main entry point. + + Synchronous wrapper that starts the async server loop. + This is the script's entry point when run directly. + + Complexity: O(1) + + Call Flow: + - Called by: __main__ block or external scripts + - Calls: asyncio.run(run()) + """ asyncio.run(run()) diff --git a/portainer_gitops_server.py b/portainer_gitops_server.py index 48c8295..ce1936e 100755 --- a/portainer_gitops_server.py +++ b/portainer_gitops_server.py @@ -4,6 +4,29 @@ Portainer GitOps MCP Server Provides GitOps automation and configuration functionality through Portainer's API. Manages automatic deployments, webhooks, and Git-based stack updates. + +This module implements a comprehensive GitOps solution for Portainer, enabling: +- Webhook-based automatic deployments triggered by Git events +- Polling-based deployments for environments without webhook access +- Stack management with Git repository integration +- Configuration updates and monitoring of GitOps status + +The server supports both webhook and polling modes, with configurable +intervals and authentication mechanisms for various Git providers. + +Complexity: O(n) for list operations where n is number of stacks +Dependencies: aiohttp for async HTTP, mcp for server protocol +Call Flow: MCP client -> handle_call_tool() -> make_request() -> Portainer API + +Environment Variables: + PORTAINER_URL: Base URL of Portainer instance (required) + PORTAINER_API_KEY: API key for authentication (required) + MCP_MODE: Set to "true" to suppress logging (default: true) + +GitOps Modes: + - Webhook: Real-time updates via Git provider webhooks + - Polling: Periodic checks for repository changes (5-30 min intervals) + - Manual: Explicit user-triggered updates only """ import os @@ -110,7 +133,32 @@ server = Server("portainer-gitops") @server.list_tools() async def handle_list_tools() -> list[types.Tool]: - """List all available tools.""" + """ + List all available tools. + + Returns the complete list of GitOps management tools provided by this + MCP server. Each tool includes its schema for input validation. + + Returns: + list[types.Tool]: List of available tools with their schemas + + Complexity: O(1) - Returns static list + + Call Flow: + - Called by: MCP protocol during initialization + - Calls: None (static return) + + Available Tools: + - list_gitops_stacks: View all stacks with GitOps configurations + - get_stack_gitops_config: Get detailed GitOps config for a stack + - enable_stack_gitops: Enable automatic updates (webhook/polling) + - disable_stack_gitops: Disable automatic updates + - update_stack_from_git: Manually trigger Git update + - configure_git_authentication: Set up Git credentials + - test_git_repository: Validate Git repository access + - get_webhook_url: Get webhook URL for Git providers + - get_polling_status: Check polling job status + """ return [ types.Tool( name="list_gitops_stacks", @@ -396,7 +444,38 @@ async def handle_call_tool( name: str, arguments: dict | None ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: - """Handle tool execution.""" + """ + Handle tool execution. + + Main request dispatcher that processes all incoming tool requests from MCP + clients. Routes requests to appropriate Portainer GitOps API endpoints and + formats responses. + + Args: + name: The name of the tool to execute + arguments: Optional dictionary of arguments for the tool + + Returns: + list[types.TextContent]: Response messages as text content + + Complexity: O(n) for list operations where n is number of stacks + + Call Flow: + - Called by: MCP protocol when client invokes a tool + - Calls: make_request() for API operations, format_gitops_status() + + Error Handling: + - API errors return descriptive error messages + - Missing required arguments return validation errors + - Network/timeout errors are caught and reported + - All exceptions caught to prevent server crashes + + GitOps Operations: + - Stack listing filters by GitOps enablement + - Configuration updates preserve existing settings + - Webhook URLs are environment-specific + - Polling intervals support various time units (s/m/h) + """ if not arguments: arguments = {} @@ -434,6 +513,7 @@ async def handle_call_tool( output += f" Repository: {stack['GitConfig']['URL']}\n" output += f" Branch: {stack['GitConfig']['ReferenceName']}\n" + # Show polling-specific configuration if method == "polling": output += f" Interval: {auto_update.get('Interval', '5m')}\n" if auto_update.get("ForceUpdate"): @@ -644,6 +724,7 @@ async def handle_call_tool( output += f" Repository: {edge_stack['GitConfig']['URL']}\n" output += f" Branch: {edge_stack['GitConfig']['ReferenceName']}\n" + # Show polling-specific configuration if method == "polling": output += f" Interval: {auto_update.get('Interval', '5m')}\n" diff --git a/portainer_kubernetes_server.py b/portainer_kubernetes_server.py index 5527efa..e4d1e5a 100755 --- a/portainer_kubernetes_server.py +++ b/portainer_kubernetes_server.py @@ -741,7 +741,35 @@ async def handle_list_tools() -> list[types.Tool]: async def make_request(method: str, endpoint: str, json_data: Optional[Dict] = None, params: Optional[Dict] = None, text_response: bool = False) -> Dict[str, Any]: - """Make HTTP request to Portainer API.""" + """ + Make HTTP request to Portainer Kubernetes API. + + Centralized HTTP request handler that manages all API communication with + Portainer's Kubernetes proxy endpoints. Handles authentication and returns + structured response data. + + Args: + method: HTTP method (GET, POST, PUT, DELETE, PATCH) + endpoint: API endpoint path (e.g., /endpoints/{id}/kubernetes/...) + json_data: Optional JSON payload for POST/PUT/PATCH requests + params: Optional query parameters for the request + text_response: If True, return raw text instead of parsing JSON + + Returns: + Dict containing status_code, data (parsed JSON or text), and raw text + + Complexity: O(1) - Single HTTP request + + Call Flow: + - Called by: handle_call_tool() for all Kubernetes API operations + - Calls: httpx for async HTTP requests + + Error Handling: + - Returns status code for caller to handle + - Preserves raw response text for debugging + - SSL verification disabled for self-signed certificates + - Handles both JSON and text responses + """ import httpx async with httpx.AsyncClient(verify=False, timeout=30.0) as client: @@ -1311,6 +1339,7 @@ async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[ for port in ports: port_str = f"{port.get('port')}" if port.get("targetPort") and str(port.get("targetPort")) != str(port.get("port")): + # Display target port mapping using arrow notation port_str += f"โ†’{port.get('targetPort')}" if port.get("nodePort"): port_str += f" (NodePort: {port.get('nodePort')})" diff --git a/portainer_stacks_server.py b/portainer_stacks_server.py index a45f118..47c79f9 100755 --- a/portainer_stacks_server.py +++ b/portainer_stacks_server.py @@ -4,6 +4,34 @@ Portainer Stacks MCP Server Provides stack deployment and management functionality through Portainer's API. Supports Docker Compose stacks and Kubernetes manifests. + +This module implements comprehensive stack management capabilities including: +- Stack creation from strings, files, Git repositories, or templates +- Stack lifecycle management (start, stop, update, delete) +- Support for both Docker Swarm and Kubernetes environments +- Environment variable and secret management +- Stack migration between environments +- Template-based deployments with variable substitution + +The server handles various deployment scenarios: +- Docker Compose stacks for Swarm mode +- Kubernetes manifests for K8s clusters +- Git-based deployments with authentication +- App templates from Portainer's registry + +Complexity: O(n) for list operations where n is number of stacks +Dependencies: aiohttp for async HTTP, mcp for server protocol +Call Flow: MCP client -> handle_call_tool() -> make_request() -> Portainer API + +Environment Variables: + PORTAINER_URL: Base URL of Portainer instance (required) + PORTAINER_API_KEY: API key for authentication (required) + MCP_MODE: Set to "true" to suppress logging (default: true) + +Stack Types: + - Swarm: Docker Compose format for Swarm orchestration + - Kubernetes: YAML manifests for K8s deployments + - Standalone: Single-node Docker Compose (converts to Swarm) """ import os @@ -47,7 +75,36 @@ async def make_request( data: Optional[Any] = None, headers: Optional[dict] = None ) -> dict: - """Make an authenticated request to Portainer API.""" + """ + Make an authenticated request to Portainer API. + + Centralized HTTP request handler for all Portainer Stack API interactions. + Handles authentication, error responses, and timeout management. + + Args: + method: HTTP method (GET, POST, PUT, DELETE) + endpoint: API endpoint path (e.g., /api/stacks) + json_data: Optional JSON payload for request body + params: Optional query parameters + data: Optional form data or raw body content + headers: Optional additional headers to merge with defaults + + Returns: + Dict containing response data or error information + Error responses have 'error' key with descriptive message + + Complexity: O(1) - Single HTTP request + + Call Flow: + - Called by: All tool handler functions + - Calls: aiohttp for async HTTP operations + + Error Handling: + - HTTP 4xx/5xx errors return structured error dict + - Timeout errors (30s) return timeout error + - Network errors return connection error + - Parses Portainer error details from response + """ url = f"{PORTAINER_URL}{endpoint}" default_headers = { @@ -95,7 +152,29 @@ async def make_request( return {"error": f"Request failed: {str(e)}"} def format_stack_status(stack: dict) -> str: - """Format stack status with emoji.""" + """ + Format stack status with emoji. + + Converts numeric stack status codes to user-friendly strings + with visual indicators for quick status recognition. + + Args: + stack: Stack object from Portainer API + + Returns: + Formatted status string with emoji + + Complexity: O(1) - Simple lookup + + Call Flow: + - Called by: Stack listing and detail display functions + - Calls: None (pure function) + + Status Codes: + - 1: Active (running) - โœ… + - 2: Inactive (stopped) - ๐Ÿ›‘ + - Other: Unknown - โ“ + """ status = stack.get("Status", 0) if status == 1: return "โœ… Active" @@ -105,7 +184,30 @@ def format_stack_status(stack: dict) -> str: return "โ“ Unknown" def format_stack_type(stack_type: int) -> str: - """Format stack type.""" + """ + Format stack type. + + Converts numeric stack type codes to readable strings identifying + the orchestration platform. + + Args: + stack_type: Numeric type code from Portainer API + + Returns: + Stack type as string (Swarm, Compose, Kubernetes, or Unknown) + + Complexity: O(1) - Simple lookup + + Call Flow: + - Called by: Stack listing and detail display functions + - Calls: None (pure function) + + Type Codes: + - 1: Swarm (Docker Swarm mode) + - 2: Compose (Standalone Docker Compose) + - 3: Kubernetes (K8s manifests) + - Other: Unknown type + """ if stack_type == 1: return "Swarm" elif stack_type == 2: @@ -120,7 +222,33 @@ server = Server("portainer-stacks") @server.list_tools() async def handle_list_tools() -> list[types.Tool]: - """List all available tools.""" + """ + List all available tools. + + Returns the complete list of stack management tools provided by this + MCP server. Each tool includes its schema for input validation. + + Returns: + list[types.Tool]: List of available tools with their schemas + + Complexity: O(1) - Returns static list + + Call Flow: + - Called by: MCP protocol during initialization + - Calls: None (static return) + + Available Tools: + - list_stacks: View all stacks across environments + - get_stack_details: Get detailed info about a specific stack + - create_stack_from_string: Deploy from Compose/K8s string + - create_stack_from_repository: Deploy from Git repository + - create_stack_from_file: Deploy from uploaded file + - update_stack: Modify existing stack configuration + - delete_stack: Remove a stack and its resources + - start_stack: Start a stopped stack + - stop_stack: Stop a running stack + - get_stack_file: Retrieve stack definition file + """ return [ types.Tool( name="list_stacks", diff --git a/run_mcp.py b/run_mcp.py deleted file mode 100755 index c472453..0000000 --- a/run_mcp.py +++ /dev/null @@ -1,14 +0,0 @@ -#!/usr/bin/env python3 -"""Direct runner for Portainer MCP server without dev dependencies.""" - -import sys -import os - -# Add src to path -sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) - -# Import and run the server -from portainer_core.server import main_sync - -if __name__ == "__main__": - main_sync() \ No newline at end of file diff --git a/run_server.py b/run_server.py index 2f1d911..9d24a92 100644 --- a/run_server.py +++ b/run_server.py @@ -23,6 +23,17 @@ Architecture: - Configuration loading from environment variables - Async server initialization and startup - Graceful error handling and shutdown + +Complexity: O(1) - Simple startup sequence +Dependencies: portainer_core.server for main server logic +Call Flow: __main__ -> setup_environment() -> asyncio.run(main()) + +Server Lifecycle: + 1. Environment validation and setup + 2. Async event loop initialization + 3. MCP server startup on stdio + 4. Process requests until interrupted + 5. Graceful shutdown on exit """ import os @@ -99,6 +110,8 @@ if __name__ == "__main__": try: # Start the async MCP server - this blocks until shutdown + # The main() function from portainer_core.server handles all MCP protocol + # communication via stdio streams (stdin/stdout) asyncio.run(main()) except KeyboardInterrupt: # Handle graceful shutdown on Ctrl+C diff --git a/simple_mcp_server.py b/simple_mcp_server.py deleted file mode 100755 index f2c3114..0000000 --- a/simple_mcp_server.py +++ /dev/null @@ -1,241 +0,0 @@ -#!/usr/bin/env python3 -"""Simple MCP server for Portainer that actually works.""" - -import os -import sys -import json -import asyncio -from typing import Any, Dict, List - -# Suppress all logging to stderr -os.environ["MCP_MODE"] = "true" - -import mcp.server.stdio -import mcp.types as types -from mcp.server import NotificationOptions, Server -from mcp.server.models import InitializationOptions - -# Add src to path -sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) - -# Create server -server = Server("portainer-core") - -# Store for our state -portainer_url = os.getenv("PORTAINER_URL", "https://partner.portainer.live") -api_key = os.getenv("PORTAINER_API_KEY", "") - - -@server.list_tools() -async def handle_list_tools() -> list[types.Tool]: - """List available tools.""" - return [ - types.Tool( - name="test_connection", - description="Test connection to Portainer", - inputSchema={ - "type": "object", - "properties": {}, - "required": [] - } - ), - types.Tool( - name="get_users", - description="Get list of Portainer users", - inputSchema={ - "type": "object", - "properties": {}, - "required": [] - } - ), - types.Tool( - name="create_user", - description="Create a new Portainer user", - inputSchema={ - "type": "object", - "properties": { - "username": { - "type": "string", - "description": "Username for the new user" - }, - "password": { - "type": "string", - "description": "Password for the new user" - }, - "role": { - "type": "string", - "description": "User role (Administrator, StandardUser, ReadOnlyUser)", - "enum": ["Administrator", "StandardUser", "ReadOnlyUser"], - "default": "StandardUser" - } - }, - "required": ["username", "password"] - } - ) - ] - - -@server.call_tool() -async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]: - """Handle tool calls.""" - import httpx - - if name == "test_connection": - try: - async with httpx.AsyncClient(verify=False) as client: - response = await client.get( - f"{portainer_url}/api/status", - headers={"X-API-Key": api_key} if api_key else {} - ) - if response.status_code == 200: - return [types.TextContent( - type="text", - text=f"โœ“ Connected to Portainer at {portainer_url}" - )] - else: - return [types.TextContent( - type="text", - text=f"โœ— Failed to connect: HTTP {response.status_code}" - )] - except Exception as e: - return [types.TextContent( - type="text", - text=f"โœ— Connection error: {str(e)}" - )] - - elif name == "get_users": - try: - async with httpx.AsyncClient(verify=False) as client: - response = await client.get( - f"{portainer_url}/api/users", - headers={"X-API-Key": api_key} - ) - if response.status_code == 200: - users = response.json() - result = f"Found {len(users)} users:\n" - for user in users: - # Handle both old (int) and new (string) role formats - role = user.get("Role", "Unknown") - if isinstance(role, int): - role_map = {1: "Administrator", 2: "StandardUser", 3: "ReadOnlyUser"} - role = role_map.get(role, f"Unknown({role})") - result += f"- {user.get('Username', 'Unknown')} ({role})\n" - return [types.TextContent(type="text", text=result)] - else: - return [types.TextContent( - type="text", - text=f"Failed to get users: HTTP {response.status_code}" - )] - except Exception as e: - return [types.TextContent( - type="text", - text=f"Error getting users: {str(e)}" - )] - - elif name == "create_user": - if not arguments: - return [types.TextContent( - type="text", - text="Error: Missing required arguments" - )] - - username = arguments.get("username") - password = arguments.get("password") - role = arguments.get("role", "StandardUser") - - if not username or not password: - return [types.TextContent( - type="text", - text="Error: Username and password are required" - )] - - # Convert role string to integer for older Portainer versions - role_map = {"Administrator": 1, "StandardUser": 2, "ReadOnlyUser": 3} - role_int = role_map.get(role, 2) - - try: - async with httpx.AsyncClient(verify=False) as client: - # Try with integer role first (older versions) - response = await client.post( - f"{portainer_url}/api/users", - headers={"X-API-Key": api_key}, - json={ - "Username": username, - "Password": password, - "Role": role_int - } - ) - - if response.status_code == 409: - return [types.TextContent( - type="text", - text=f"User '{username}' already exists" - )] - elif response.status_code in [200, 201]: - return [types.TextContent( - type="text", - text=f"โœ“ User '{username}' created successfully with role {role}" - )] - else: - # Try with string role (newer versions) - response = await client.post( - f"{portainer_url}/api/users", - headers={"X-API-Key": api_key}, - json={ - "Username": username, - "Password": password, - "Role": role - } - ) - if response.status_code in [200, 201]: - return [types.TextContent( - type="text", - text=f"โœ“ User '{username}' created successfully with role {role}" - )] - else: - return [types.TextContent( - type="text", - text=f"Failed to create user: HTTP {response.status_code} - {response.text}" - )] - except Exception as e: - return [types.TextContent( - type="text", - text=f"Error creating user: {str(e)}" - )] - - else: - return [types.TextContent( - type="text", - text=f"Unknown tool: {name}" - )] - - -async def run(): - """Run the MCP server.""" - # Use stdio transport - async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): - await server.run( - read_stream, - write_stream, - InitializationOptions( - server_name="portainer-core", - server_version="0.2.0", - capabilities=server.get_capabilities( - notification_options=NotificationOptions( - prompts_changed=False, - resources_changed=False, - tools_changed=False, - ), - experimental_capabilities={}, - ), - ), - ) - - -def main(): - """Main entry point.""" - asyncio.run(run()) - - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/src/portainer_core_mcp.egg-info/PKG-INFO b/src/portainer_core_mcp.egg-info/PKG-INFO deleted file mode 100644 index 8d682a9..0000000 --- a/src/portainer_core_mcp.egg-info/PKG-INFO +++ /dev/null @@ -1,258 +0,0 @@ -Metadata-Version: 2.4 -Name: portainer-core-mcp -Version: 0.1.0 -Summary: Portainer Core MCP Server - Authentication and User Management -Author-email: Portainer MCP Team -License: MIT -Project-URL: Homepage, https://github.com/portainer/portainer-mcp-core -Project-URL: Documentation, https://github.com/portainer/portainer-mcp-core#readme -Project-URL: Repository, https://github.com/portainer/portainer-mcp-core -Project-URL: Issues, https://github.com/portainer/portainer-mcp-core/issues -Classifier: Development Status :: 3 - Alpha -Classifier: Intended Audience :: Developers -Classifier: License :: OSI Approved :: MIT License -Classifier: Programming Language :: Python :: 3 -Classifier: Programming Language :: Python :: 3.8 -Classifier: Programming Language :: Python :: 3.9 -Classifier: Programming Language :: Python :: 3.10 -Classifier: Programming Language :: Python :: 3.11 -Classifier: Programming Language :: Python :: 3.12 -Requires-Python: >=3.8 -Description-Content-Type: text/markdown -Requires-Dist: mcp>=1.0.0 -Requires-Dist: httpx>=0.25.0 -Requires-Dist: pydantic>=2.0.0 -Requires-Dist: pydantic-settings>=2.0.0 -Requires-Dist: structlog>=23.0.0 -Requires-Dist: PyJWT>=2.8.0 -Requires-Dist: python-dotenv>=1.0.0 -Requires-Dist: tenacity>=8.0.0 -Provides-Extra: dev -Requires-Dist: pytest>=7.0.0; extra == "dev" -Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev" -Requires-Dist: pytest-cov>=4.0.0; extra == "dev" -Requires-Dist: pytest-mock>=3.10.0; extra == "dev" -Requires-Dist: httpx-mock>=0.10.0; extra == "dev" -Requires-Dist: black>=23.0.0; extra == "dev" -Requires-Dist: isort>=5.12.0; extra == "dev" -Requires-Dist: flake8>=6.0.0; extra == "dev" -Requires-Dist: mypy>=1.0.0; extra == "dev" -Requires-Dist: pre-commit>=3.0.0; extra == "dev" - -# Portainer Core MCP Server - -A Model Context Protocol (MCP) server that provides authentication and user management functionality for Portainer Business Edition. - -## Features - -- **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 -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 - -### Environment Variables - -Create a `.env` file or set environment variables: - -```bash -# Required -PORTAINER_URL=https://your-portainer-instance.com -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 - -### Start the Server - -#### Using Python - -```bash -python run_server.py -``` - -#### Using uv - -```bash -uv run python run_server.py -``` - -#### 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 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 configuration - -### Health -- `health_check` - Server health status - -## Available Resources - -- `portainer://users` - User management data -- `portainer://settings` - Configuration settings -- `portainer://health` - Server health status - -## Development - -### Setup - -```bash -# Clone the repository -git clone https://github.com/yourusername/portainer-core-mcp.git -cd portainer-core-mcp - -# Install development dependencies -pip install -e ".[dev]" - -# Install pre-commit hooks -pre-commit install -``` - -### Testing - -```bash -# Run all tests -pytest - -# Run with coverage -pytest --cov=src/portainer_core --cov-report=html - -# Run only unit tests -pytest -m unit - -# Run only integration tests -pytest -m integration -``` - -### Code Quality - -```bash -# Format code -black src tests -isort src tests - -# Lint code -flake8 src tests - -# Type checking -mypy src -``` - -## Architecture - -The server follows a layered architecture: - -- **MCP Server Layer**: Handles MCP protocol communication -- **Service Layer**: Abstracts Portainer API interactions -- **Models Layer**: Defines data structures and validation -- **Utils Layer**: Provides utility functions and helpers - -## Security - -- All API communications use HTTPS -- JWT tokens are handled securely and never logged -- Input validation on all parameters -- Rate limiting to prevent abuse -- Circuit breaker pattern for fault tolerance - -## Contributing - -1. Fork the repository -2. Create a feature branch -3. Make your changes -4. Add tests for new functionality -5. Ensure all tests pass -6. Submit a pull request - -## License - -MIT License - see LICENSE file for details. diff --git a/src/portainer_core_mcp.egg-info/SOURCES.txt b/src/portainer_core_mcp.egg-info/SOURCES.txt deleted file mode 100644 index 4dace85..0000000 --- a/src/portainer_core_mcp.egg-info/SOURCES.txt +++ /dev/null @@ -1,25 +0,0 @@ -README.md -pyproject.toml -src/portainer_core/__init__.py -src/portainer_core/config.py -src/portainer_core/server.py -src/portainer_core/models/__init__.py -src/portainer_core/models/auth.py -src/portainer_core/models/settings.py -src/portainer_core/models/users.py -src/portainer_core/services/__init__.py -src/portainer_core/services/auth.py -src/portainer_core/services/base.py -src/portainer_core/services/settings.py -src/portainer_core/services/users.py -src/portainer_core/utils/__init__.py -src/portainer_core/utils/errors.py -src/portainer_core/utils/logging.py -src/portainer_core/utils/tokens.py -src/portainer_core_mcp.egg-info/PKG-INFO -src/portainer_core_mcp.egg-info/SOURCES.txt -src/portainer_core_mcp.egg-info/dependency_links.txt -src/portainer_core_mcp.egg-info/entry_points.txt -src/portainer_core_mcp.egg-info/requires.txt -src/portainer_core_mcp.egg-info/top_level.txt -tests/test_basic.py \ No newline at end of file diff --git a/src/portainer_core_mcp.egg-info/dependency_links.txt b/src/portainer_core_mcp.egg-info/dependency_links.txt deleted file mode 100644 index 8b13789..0000000 --- a/src/portainer_core_mcp.egg-info/dependency_links.txt +++ /dev/null @@ -1 +0,0 @@ - diff --git a/src/portainer_core_mcp.egg-info/entry_points.txt b/src/portainer_core_mcp.egg-info/entry_points.txt deleted file mode 100644 index dc7da5f..0000000 --- a/src/portainer_core_mcp.egg-info/entry_points.txt +++ /dev/null @@ -1,2 +0,0 @@ -[console_scripts] -portainer-core-mcp = portainer_core.server:main_sync diff --git a/src/portainer_core_mcp.egg-info/requires.txt b/src/portainer_core_mcp.egg-info/requires.txt deleted file mode 100644 index 312cdcb..0000000 --- a/src/portainer_core_mcp.egg-info/requires.txt +++ /dev/null @@ -1,20 +0,0 @@ -mcp>=1.0.0 -httpx>=0.25.0 -pydantic>=2.0.0 -pydantic-settings>=2.0.0 -structlog>=23.0.0 -PyJWT>=2.8.0 -python-dotenv>=1.0.0 -tenacity>=8.0.0 - -[dev] -pytest>=7.0.0 -pytest-asyncio>=0.21.0 -pytest-cov>=4.0.0 -pytest-mock>=3.10.0 -httpx-mock>=0.10.0 -black>=23.0.0 -isort>=5.12.0 -flake8>=6.0.0 -mypy>=1.0.0 -pre-commit>=3.0.0 diff --git a/src/portainer_core_mcp.egg-info/top_level.txt b/src/portainer_core_mcp.egg-info/top_level.txt deleted file mode 100644 index 661bb10..0000000 --- a/src/portainer_core_mcp.egg-info/top_level.txt +++ /dev/null @@ -1 +0,0 @@ -portainer_core diff --git a/test_edge_api.sh b/test_edge_api.sh deleted file mode 100755 index ccbd170..0000000 --- a/test_edge_api.sh +++ /dev/null @@ -1,44 +0,0 @@ -#!/bin/bash - -# Test Portainer Edge API endpoints - -echo "๐Ÿš€ Testing Portainer Edge API" -echo "URL: $PORTAINER_URL" -echo "API Key: ***${PORTAINER_API_KEY: -4}" -echo "" - -# Test edge environments -echo "๐ŸŒ Testing Edge Environments..." -curl -s -H "X-API-Key: $PORTAINER_API_KEY" \ - "$PORTAINER_URL/api/endpoints?types=4" | jq '.[] | select(.Type == 4) | {name: .Name, id: .Id, status: .Status}' 2>/dev/null || echo "No edge environments found or jq not installed" - -echo "" - -# Test edge groups -echo "๐Ÿ‘ฅ Testing Edge Groups..." -curl -s -H "X-API-Key: $PORTAINER_API_KEY" \ - "$PORTAINER_URL/api/edge_groups" | jq '.[0:3] | .[] | {name: .Name, id: .Id, dynamic: .Dynamic}' 2>/dev/null || echo "No edge groups found or jq not installed" - -echo "" - -# Test edge stacks -echo "๐Ÿ“š Testing Edge Stacks..." -curl -s -H "X-API-Key: $PORTAINER_API_KEY" \ - "$PORTAINER_URL/api/edge_stacks" | jq '.[0:3] | .[] | {name: .Name, id: .Id, groups: .EdgeGroups}' 2>/dev/null || echo "No edge stacks found or jq not installed" - -echo "" - -# Test edge jobs -echo "๐Ÿ’ผ Testing Edge Jobs..." -curl -s -H "X-API-Key: $PORTAINER_API_KEY" \ - "$PORTAINER_URL/api/edge_jobs" | jq '.[0:3] | .[] | {name: .Name, id: .Id, recurring: .Recurring}' 2>/dev/null || echo "No edge jobs found or jq not installed" - -echo "" - -# Test edge settings -echo "โš™๏ธ Testing Edge Settings..." -curl -s -H "X-API-Key: $PORTAINER_API_KEY" \ - "$PORTAINER_URL/api/settings" | jq '.Edge | {checkin: .CheckinInterval, command: .CommandInterval, ping: .PingInterval}' 2>/dev/null || echo "Failed to get edge settings or jq not installed" - -echo "" -echo "โœ… Edge API test completed!" \ No newline at end of file diff --git a/test_edge_server.py b/test_edge_server.py deleted file mode 100755 index 318ed9f..0000000 --- a/test_edge_server.py +++ /dev/null @@ -1,163 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script for Portainer Edge MCP Server -Tests edge environments, stacks, groups, and jobs functionality -""" - -import asyncio -import aiohttp -import os -import json - -PORTAINER_URL = os.getenv("PORTAINER_URL", "").rstrip("/") -PORTAINER_API_KEY = os.getenv("PORTAINER_API_KEY", "") - -async def make_request(method: str, endpoint: str, json_data=None, params=None): - """Make a request to Portainer API""" - url = f"{PORTAINER_URL}{endpoint}" - headers = {"X-API-Key": PORTAINER_API_KEY} - - async with aiohttp.ClientSession() as session: - async with session.request( - method, - url, - json=json_data, - params=params, - headers=headers - ) as response: - text = await response.text() - if response.status >= 400: - print(f"โŒ Error {response.status}: {text}") - return None - return json.loads(text) if text else {} - -async def test_edge_environments(): - """Test edge environment operations""" - print("\n๐ŸŒ Testing Edge Environments...") - - # List all endpoints/environments - endpoints = await make_request("GET", "/api/endpoints") - if endpoints: - edge_envs = [e for e in endpoints if e.get("Type") == 4] # Type 4 is Edge - print(f"โœ… Found {len(edge_envs)} edge environments") - - if edge_envs: - # Get details of first edge environment - env = edge_envs[0] - print(f" โ€ข {env['Name']} (ID: {env['Id']})") - print(f" Status: {env.get('Status', 'Unknown')}") - print(f" Edge ID: {env.get('EdgeID', 'N/A')}") - - # Get edge status - status = await make_request("GET", f"/api/endpoints/{env['Id']}/edge/status") - if status: - print(f" Check-in: {status.get('CheckinTime', 'Never')}") - else: - print("โŒ Failed to list environments") - -async def test_edge_groups(): - """Test edge group operations""" - print("\n๐Ÿ‘ฅ Testing Edge Groups...") - - # List edge groups - groups = await make_request("GET", "/api/edge_groups") - if groups: - print(f"โœ… Found {len(groups)} edge groups") - for group in groups[:3]: # Show first 3 - print(f" โ€ข {group['Name']} (ID: {group['Id']})") - print(f" Dynamic: {'Yes' if group.get('Dynamic') else 'No'}") - print(f" Endpoints: {len(group.get('Endpoints', []))}") - else: - print("โŒ Failed to list edge groups") - - # Try to create a test edge group - print("\n๐Ÿ“ Creating test edge group...") - test_group_data = { - "Name": "test-edge-group", - "Dynamic": False, - "TagIds": [] - } - - new_group = await make_request("POST", "/api/edge_groups", json_data=test_group_data) - if new_group: - print(f"โœ… Created edge group: {new_group['Name']} (ID: {new_group['Id']})") - - # Clean up - delete the test group - await make_request("DELETE", f"/api/edge_groups/{new_group['Id']}") - print("๐Ÿ—‘๏ธ Cleaned up test edge group") - else: - print("โŒ Failed to create edge group") - -async def test_edge_stacks(): - """Test edge stack operations""" - print("\n๐Ÿ“š Testing Edge Stacks...") - - # List edge stacks - stacks = await make_request("GET", "/api/edge_stacks") - if stacks: - print(f"โœ… Found {len(stacks)} edge stacks") - for stack in stacks[:3]: # Show first 3 - print(f" โ€ข {stack['Name']} (ID: {stack['Id']})") - print(f" Type: {stack.get('StackType', 'Unknown')}") - print(f" Groups: {len(stack.get('EdgeGroups', []))}") - - # Check if it has GitOps - if stack.get("GitConfig") and stack.get("AutoUpdate"): - print(f" GitOps: Enabled ({stack['AutoUpdate'].get('Interval', 'N/A')})") - else: - print("โŒ Failed to list edge stacks") - -async def test_edge_jobs(): - """Test edge job operations""" - print("\n๐Ÿ’ผ Testing Edge Jobs...") - - # List edge jobs - jobs = await make_request("GET", "/api/edge_jobs") - if jobs: - print(f"โœ… Found {len(jobs)} edge jobs") - for job in jobs[:3]: # Show first 3 - print(f" โ€ข {job['Name']} (ID: {job['Id']})") - print(f" Recurring: {'Yes' if job.get('Recurring') else 'No'}") - if job.get('CronExpression'): - print(f" Schedule: {job['CronExpression']}") - print(f" Target Groups: {len(job.get('EdgeGroups', []))}") - else: - print("โŒ Failed to list edge jobs") - -async def test_edge_settings(): - """Test edge settings""" - print("\nโš™๏ธ Testing Edge Settings...") - - # Get settings - settings = await make_request("GET", "/api/settings") - if settings and settings.get("Edge"): - edge_settings = settings["Edge"] - print("โœ… Edge Settings:") - print(f" โ€ข Check-in Interval: {edge_settings.get('CheckinInterval', 'N/A')} seconds") - print(f" โ€ข Command Interval: {edge_settings.get('CommandInterval', 'N/A')} seconds") - print(f" โ€ข Ping Interval: {edge_settings.get('PingInterval', 'N/A')} seconds") - print(f" โ€ข Tunnel Server: {edge_settings.get('TunnelServerAddress', 'Not configured')}") - else: - print("โŒ Failed to get edge settings") - -async def main(): - """Run all edge tests""" - print("๐Ÿš€ Portainer Edge API Tests") - print(f"URL: {PORTAINER_URL}") - print(f"API Key: {'***' + PORTAINER_API_KEY[-4:] if PORTAINER_API_KEY else 'Not set'}") - - if not PORTAINER_URL or not PORTAINER_API_KEY: - print("\nโŒ Please set PORTAINER_URL and PORTAINER_API_KEY environment variables") - return - - # Run tests - await test_edge_environments() - await test_edge_groups() - await test_edge_stacks() - await test_edge_jobs() - await test_edge_settings() - - print("\nโœ… Edge API tests completed!") - -if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file diff --git a/test_gitops_create.py b/test_gitops_create.py deleted file mode 100644 index db1c353..0000000 --- a/test_gitops_create.py +++ /dev/null @@ -1,164 +0,0 @@ -#!/usr/bin/env python3 -"""Test creating a stack with GitOps enabled""" - -import asyncio -import aiohttp -import json -import sys - -# Configuration -PORTAINER_URL = "https://partner.portainer.live" -PORTAINER_API_KEY = "ptr_uMqreULEo44qvuszgG8oZWdjkDx3K9HBXSmjd+F/vDE=" - -# Stack configuration -STACK_NAME = "nginx03-gitops" -ENVIRONMENT_ID = 6 # docker03 -REPOSITORY_URL = "https://git.oe74.net/adelorenzo/portainer-yaml" -REPOSITORY_REF = "main" -COMPOSE_PATH = "docker-compose.yml" - -async def create_stack_with_gitops(): - """Create a stack with GitOps enabled from the start""" - - # Build request data - data = { - "name": STACK_NAME, - "repositoryURL": REPOSITORY_URL, - "repositoryReferenceName": REPOSITORY_REF, - "composeFilePathInRepository": COMPOSE_PATH, - "repositoryAuthentication": False, - "autoUpdate": { - "interval": "5m", - "forcePullImage": True, - "forceUpdate": False - } - } - - # Headers - headers = { - "X-API-Key": PORTAINER_API_KEY, - "Content-Type": "application/json" - } - - # API endpoint - endpoint = f"{PORTAINER_URL}/api/stacks/create/standalone/repository?endpointId={ENVIRONMENT_ID}" - - print(f"Creating stack '{STACK_NAME}' with GitOps enabled...") - print(f"Repository: {REPOSITORY_URL}") - print(f"Compose file: {COMPOSE_PATH}") - print(f"Environment ID: {ENVIRONMENT_ID}") - print(f"GitOps: Enabled (polling every 5m)") - - try: - async with aiohttp.ClientSession() as session: - async with session.post(endpoint, json=data, headers=headers) as response: - response_text = await response.text() - - if response.status in [200, 201]: - result = json.loads(response_text) - print(f"\nโœ… Stack created successfully!") - print(f"Stack ID: {result['Id']}") - print(f"Stack Name: {result['Name']}") - - # Check GitOps status - if result.get("AutoUpdate"): - print(f"\n๐Ÿ”„ GitOps Status:") - auto_update = result["AutoUpdate"] - print(f" Enabled: Yes") - print(f" Interval: {auto_update.get('Interval', 'N/A')}") - print(f" Force Pull Image: {auto_update.get('ForcePullImage', False)}") - print(f" Force Update: {auto_update.get('ForceUpdate', False)}") - else: - print(f"\nโš ๏ธ GitOps is not enabled on the created stack") - - return result - else: - print(f"\nโŒ Error creating stack: {response.status}") - print(f"Response: {response_text}") - - # Try to parse error message - try: - error_data = json.loads(response_text) - if "message" in error_data: - print(f"Error message: {error_data['message']}") - elif "details" in error_data: - print(f"Error details: {error_data['details']}") - except: - pass - - return None - - except Exception as e: - print(f"\nโŒ Exception occurred: {str(e)}") - return None - -async def check_stack_gitops(stack_id): - """Check if GitOps is enabled on a stack""" - - headers = { - "X-API-Key": PORTAINER_API_KEY - } - - endpoint = f"{PORTAINER_URL}/api/stacks/{stack_id}" - - try: - async with aiohttp.ClientSession() as session: - async with session.get(endpoint, headers=headers) as response: - if response.status == 200: - result = await response.json() - print(f"\n๐Ÿ“š Stack Details:") - print(f"Name: {result['Name']}") - print(f"ID: {result['Id']}") - - if result.get("GitConfig"): - print(f"\n๐Ÿ”— Git Configuration:") - git = result["GitConfig"] - print(f" Repository: {git['URL']}") - print(f" Reference: {git['ReferenceName']}") - print(f" Path: {git.get('ConfigFilePath', 'N/A')}") - - if result.get("AutoUpdate"): - print(f"\n๐Ÿ”„ GitOps Configuration:") - auto = result["AutoUpdate"] - print(f" Enabled: Yes") - print(f" Interval: {auto.get('Interval', 'N/A')}") - print(f" Force Pull Image: {auto.get('ForcePullImage', False)}") - print(f" Force Update: {auto.get('ForceUpdate', False)}") - if auto.get("Webhook"): - print(f" Webhook: Enabled") - else: - print(f"\nโŒ GitOps: Disabled") - - return result - else: - print(f"Error getting stack details: {response.status}") - return None - except Exception as e: - print(f"Exception getting stack details: {str(e)}") - return None - -async def main(): - print("=" * 60) - print("Testing Portainer Stack Creation with GitOps") - print("=" * 60) - - # Create the stack with GitOps - result = await create_stack_with_gitops() - - if result: - print("\n" + "=" * 60) - print("Verifying GitOps configuration...") - print("=" * 60) - - # Wait a moment for the stack to be fully created - await asyncio.sleep(2) - - # Check the stack details - await check_stack_gitops(result["Id"]) - - print("\n๐ŸŽ‰ Test completed!") - else: - print("\n๐Ÿ˜ž Test failed!") - -if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file diff --git a/test_uvx.py b/test_uvx.py deleted file mode 100644 index e0f9166..0000000 --- a/test_uvx.py +++ /dev/null @@ -1,56 +0,0 @@ -#!/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