From 7a1abbe2439aa1c266681d788b2ff49fe527fc49 Mon Sep 17 00:00:00 2001 From: Adolfo Delorenzo Date: Fri, 18 Jul 2025 23:59:56 -0300 Subject: [PATCH] feat: add Portainer Edge MCP server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Implement comprehensive edge computing functionality - Add edge environment management (list, get, status, generate keys) - Add edge stack operations (list, get, create, update, delete) - Add edge group management (list, get, create, update, delete) - Add edge job scheduling (list, get, create, delete) - Add edge settings configuration (get, update) - Create test scripts for edge API validation - Add comprehensive README documentation for edge server - Include nginx stack creation script from earlier testing šŸ¤– Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- README_EDGE.md | 312 +++++ create_nginx_stack.py | 131 +++ portainer_edge_server.py | 1019 +++++++++++++++++ src/portainer_core_mcp.egg-info/PKG-INFO | 141 ++- src/portainer_core_mcp.egg-info/SOURCES.txt | 7 + .../entry_points.txt | 2 +- test_edge_api.sh | 44 + test_edge_server.py | 163 +++ 8 files changed, 1789 insertions(+), 30 deletions(-) create mode 100644 README_EDGE.md create mode 100644 create_nginx_stack.py create mode 100644 portainer_edge_server.py create mode 100755 test_edge_api.sh create mode 100755 test_edge_server.py diff --git a/README_EDGE.md b/README_EDGE.md new file mode 100644 index 0000000..b0b0263 --- /dev/null +++ b/README_EDGE.md @@ -0,0 +1,312 @@ +# Portainer Edge MCP Server + +This MCP server provides edge computing functionality through Portainer's API, managing edge environments, edge stacks, edge groups, and edge jobs. + +## Features + +- **Edge Environment Management**: List and monitor edge environments +- **Edge Stack Deployment**: Deploy and manage stacks across edge devices +- **Edge Group Organization**: Create and manage groups of edge endpoints +- **Edge Job Scheduling**: Schedule and run jobs on edge devices +- **Edge Settings Configuration**: Configure global edge settings + +## Installation + +1. Ensure you have the Portainer MCP servers repository: + ```bash + git clone https://github.com/yourusername/portainer-mcp.git + cd portainer-mcp + ``` + +2. Install dependencies: + ```bash + pip install -r requirements.txt + ``` + +3. Configure environment variables: + ```bash + cp .env.example .env + # Edit .env with your Portainer URL and API key + ``` + +4. Make the server executable: + ```bash + chmod +x portainer_edge_server.py + ``` + +## Configuration + +Add to your Claude Desktop configuration: + +```json +{ + "portainer-edge": { + "command": "python", + "args": ["/path/to/portainer-mcp/portainer_edge_server.py"], + "env": { + "PORTAINER_URL": "https://your-portainer-instance.com", + "PORTAINER_API_KEY": "your-api-key" + } + } +} +``` + +## Available Tools + +### Edge Environment Management + +#### list_edge_environments +List all edge environments. +- **Parameters**: + - `status` (optional): Filter by status - "connected" or "disconnected" + +#### get_edge_environment +Get details of a specific edge environment. +- **Parameters**: + - `environment_id` (required): Environment ID + +#### get_edge_status +Get edge environment status and check-in information. +- **Parameters**: + - `environment_id` (required): Environment ID + +#### generate_edge_key +Generate an edge key for adding new edge agents. +- **Parameters**: + - `name` (required): Environment name + - `group_ids` (optional): List of edge group IDs + +### Edge Stack Management + +#### list_edge_stacks +List all edge stacks. + +#### get_edge_stack +Get details of a specific edge stack. +- **Parameters**: + - `edge_stack_id` (required): Edge Stack ID + +#### create_edge_stack +Create a new edge stack. +- **Parameters**: + - `name` (required): Stack name + - `stack_content` (required): Stack file content (Docker Compose) + - `edge_groups` (required): List of edge group IDs + - `deploy_type` (optional): Deployment type (0: compose, 1: kubernetes) + - `edge_id_list` (optional): Specific edge IDs for deployment + - `registries` (optional): List of registry IDs + +#### update_edge_stack +Update an existing edge stack. +- **Parameters**: + - `edge_stack_id` (required): Edge Stack ID + - `stack_content` (optional): Updated stack content + - `edge_groups` (optional): Updated edge group IDs + - `deploy_type` (optional): Updated deployment type + +#### delete_edge_stack +Delete an edge stack. +- **Parameters**: + - `edge_stack_id` (required): Edge Stack ID + +### Edge Group Management + +#### list_edge_groups +List all edge groups. + +#### get_edge_group +Get details of a specific edge group. +- **Parameters**: + - `edge_group_id` (required): Edge Group ID + +#### create_edge_group +Create a new edge group. +- **Parameters**: + - `name` (required): Group name + - `dynamic` (optional): Enable dynamic membership (default: false) + - `tag_ids` (optional): Tag IDs for dynamic groups + - `endpoints` (optional): Endpoint IDs for static groups + +#### update_edge_group +Update an existing edge group. +- **Parameters**: + - `edge_group_id` (required): Edge Group ID + - `name` (optional): Updated name + - `dynamic` (optional): Update dynamic membership + - `tag_ids` (optional): Updated tag IDs + - `endpoints` (optional): Updated endpoint IDs + +#### delete_edge_group +Delete an edge group. +- **Parameters**: + - `edge_group_id` (required): Edge Group ID + +### Edge Job Management + +#### list_edge_jobs +List all edge jobs. + +#### get_edge_job +Get details of a specific edge job. +- **Parameters**: + - `edge_job_id` (required): Edge Job ID + +#### create_edge_job +Create a new edge job. +- **Parameters**: + - `name` (required): Job name + - `edge_groups` (required): Target edge group IDs + - `script_content` (required): Script content to execute + - `recurring` (optional): Enable recurring execution + - `cron_expression` (optional): Cron expression for scheduling + +#### delete_edge_job +Delete an edge job. +- **Parameters**: + - `edge_job_id` (required): Edge Job ID + +### Edge Settings + +#### get_edge_settings +Get global edge settings. + +#### update_edge_settings +Update global edge settings. +- **Parameters**: + - `check_in_interval` (optional): Check-in interval in seconds + - `command_interval` (optional): Command interval in seconds + - `ping_interval` (optional): Ping interval in seconds + - `snapshot_interval` (optional): Snapshot interval in seconds + - `tunnel_server_address` (optional): Tunnel server address + +## Usage Examples + +### List Edge Environments + +```javascript +await use_mcp_tool("portainer-edge", "list_edge_environments", { + status: "connected" +}); +``` + +### Create Edge Stack + +```javascript +await use_mcp_tool("portainer-edge", "create_edge_stack", { + name: "nginx-edge", + stack_content: `version: '3' +services: + nginx: + image: nginx:alpine + ports: + - "80:80"`, + edge_groups: ["1", "2"] +}); +``` + +### Create Edge Group + +```javascript +// Static group with specific endpoints +await use_mcp_tool("portainer-edge", "create_edge_group", { + name: "production-edge", + dynamic: false, + endpoints: ["10", "11", "12"] +}); + +// Dynamic group based on tags +await use_mcp_tool("portainer-edge", "create_edge_group", { + name: "tagged-devices", + dynamic: true, + tag_ids: ["1", "3"] +}); +``` + +### Schedule Edge Job + +```javascript +await use_mcp_tool("portainer-edge", "create_edge_job", { + name: "system-update", + edge_groups: ["1"], + script_content: "apt update && apt upgrade -y", + recurring: true, + cron_expression: "0 2 * * *" // Daily at 2 AM +}); +``` + +## Edge Computing Concepts + +### Edge Environments +Edge environments are remote Docker or Kubernetes environments that connect to Portainer via the Edge Agent. They can be: +- **Connected**: Currently connected and checking in +- **Disconnected**: Not currently reachable + +### Edge Stacks +Edge stacks are Docker Compose or Kubernetes deployments that can be deployed to multiple edge environments simultaneously through edge groups. + +### Edge Groups +Edge groups organize edge environments for bulk operations: +- **Static Groups**: Manually selected endpoints +- **Dynamic Groups**: Automatically populated based on tags + +### Edge Jobs +Edge jobs execute scripts on edge devices: +- **One-time Jobs**: Execute once immediately +- **Recurring Jobs**: Execute on a schedule (cron) + +## Testing + +Use the provided test script to verify edge functionality: + +```bash +python test_edge_server.py +``` + +This will test: +- Listing edge environments +- Creating and managing edge groups +- Listing edge stacks +- Viewing edge jobs +- Checking edge settings + +## Best Practices + +1. **Group Organization**: Use edge groups to organize devices by location, purpose, or capability +2. **Stack Templates**: Create reusable stack templates for common deployments +3. **Job Scheduling**: Use recurring jobs for maintenance tasks +4. **Monitoring**: Regularly check edge environment status +5. **Security**: Use proper authentication for edge agents + +## Security Considerations + +- Edge agents use unique keys for authentication +- Communication is encrypted between edge agents and Portainer +- Edge jobs execute with the permissions of the edge agent +- Limit edge job permissions appropriately +- Regularly rotate edge keys + +## Troubleshooting + +### Edge Environment Not Connecting +- Check network connectivity from edge device +- Verify edge key is correct +- Check firewall rules +- Review edge agent logs + +### Stack Deployment Failures +- Verify stack syntax +- Check image availability on edge devices +- Review resource constraints +- Check edge agent permissions + +### Edge Group Issues +- Verify tag configuration for dynamic groups +- Check endpoint assignments for static groups +- Review group membership rules + +## Requirements + +- Python 3.8+ +- Portainer Business Edition 2.19+ (for full edge features) +- Valid Portainer API token +- Edge agents deployed on target devices \ No newline at end of file diff --git a/create_nginx_stack.py b/create_nginx_stack.py new file mode 100644 index 0000000..7d79f27 --- /dev/null +++ b/create_nginx_stack.py @@ -0,0 +1,131 @@ +#!/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/portainer_edge_server.py b/portainer_edge_server.py new file mode 100644 index 0000000..d5809d2 --- /dev/null +++ b/portainer_edge_server.py @@ -0,0 +1,1019 @@ +#!/usr/bin/env python3 +""" +Portainer Edge MCP Server + +Provides edge computing functionality through Portainer's API. +Manages edge environments, edge stacks, edge groups, and edge jobs. +""" + +import os +import sys +import json +import asyncio +import aiohttp +import logging +from typing import Any, Optional +from mcp.server import Server, NotificationOptions +from mcp.server.models import InitializationOptions +import mcp.server.stdio +import mcp.types as types + +# Set up logging +MCP_MODE = os.getenv("MCP_MODE", "true").lower() == "true" +if MCP_MODE: + # In MCP mode, suppress all logs to stdout/stderr + logging.basicConfig(level=logging.CRITICAL + 1) +else: + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') + +logger = logging.getLogger(__name__) + +# Environment variables +PORTAINER_URL = os.getenv("PORTAINER_URL", "").rstrip("/") +PORTAINER_API_KEY = os.getenv("PORTAINER_API_KEY", "") + +# Validate environment +if not PORTAINER_URL or not PORTAINER_API_KEY: + if not MCP_MODE: + logger.error("PORTAINER_URL and PORTAINER_API_KEY must be set") + sys.exit(1) + +# Helper functions +async def make_request( + method: str, + endpoint: str, + json_data: Optional[dict] = None, + params: Optional[dict] = None, + data: Optional[Any] = None, + headers: Optional[dict] = None +) -> dict: + """Make an authenticated request to Portainer API.""" + url = f"{PORTAINER_URL}{endpoint}" + + default_headers = { + "X-API-Key": PORTAINER_API_KEY + } + + if headers: + default_headers.update(headers) + + timeout = aiohttp.ClientTimeout(total=30) + + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.request( + method, + url, + json=json_data, + params=params, + data=data, + headers=default_headers + ) as response: + response_text = await response.text() + + if response.status >= 400: + error_msg = f"API request failed: {response.status}" + try: + error_data = json.loads(response_text) + if "message" in error_data: + error_msg = f"{error_msg} - {error_data['message']}" + elif "details" in error_data: + error_msg = f"{error_msg} - {error_data['details']}" + except: + if response_text: + error_msg = f"{error_msg} - {response_text}" + + return {"error": error_msg} + + if response_text: + return json.loads(response_text) + return {} + + except asyncio.TimeoutError: + return {"error": "Request timeout"} + except Exception as e: + return {"error": f"Request failed: {str(e)}"} + +def format_edge_status(status: int) -> str: + """Format edge environment status with emoji.""" + if status == 1: + return "šŸŸ¢ Connected" + elif status == 2: + return "šŸ”“ Disconnected" + else: + return "āšŖ Unknown" + +def format_edge_group_type(group_type: int) -> str: + """Format edge group type.""" + if group_type == 1: + return "Static" + elif group_type == 2: + return "Dynamic" + else: + return "Unknown" + +# Create server instance +server = Server("portainer-edge") + +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List all available tools.""" + return [ + # Edge Environment Tools + types.Tool( + name="list_edge_environments", + description="List all edge environments", + inputSchema={ + "type": "object", + "properties": { + "group_id": { + "type": "string", + "description": "Filter by edge group ID (optional)" + }, + "status": { + "type": "string", + "enum": ["connected", "disconnected", "all"], + "description": "Filter by connection status", + "default": "all" + } + } + } + ), + types.Tool( + name="get_edge_environment", + description="Get detailed information about an edge environment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "string", + "description": "Edge environment ID" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="get_edge_status", + description="Get the connection status of an edge environment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "string", + "description": "Edge environment ID" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="generate_edge_key", + description="Generate an edge agent deployment script", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Name for the edge environment" + }, + "group_id": { + "type": "string", + "description": "Edge group ID to assign to (optional)" + } + }, + "required": ["name"] + } + ), + + # Edge Stack Tools + types.Tool( + name="list_edge_stacks", + description="List all edge stacks", + inputSchema={ + "type": "object", + "properties": {} + } + ), + types.Tool( + name="get_edge_stack", + description="Get detailed information about an edge stack", + inputSchema={ + "type": "object", + "properties": { + "edge_stack_id": { + "type": "string", + "description": "Edge stack ID" + } + }, + "required": ["edge_stack_id"] + } + ), + types.Tool( + name="create_edge_stack", + description="Create a new edge stack", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Edge stack name" + }, + "stack_file_content": { + "type": "string", + "description": "Stack file content (Docker Compose or Kubernetes manifest)" + }, + "edge_groups": { + "type": "array", + "items": {"type": "string"}, + "description": "List of edge group IDs to deploy to" + }, + "deploy_type": { + "type": "integer", + "description": "Deployment type (0=compose, 1=kubernetes)", + "default": 0 + } + }, + "required": ["name", "stack_file_content", "edge_groups"] + } + ), + types.Tool( + name="update_edge_stack", + description="Update an existing edge stack", + inputSchema={ + "type": "object", + "properties": { + "edge_stack_id": { + "type": "string", + "description": "Edge stack ID" + }, + "stack_file_content": { + "type": "string", + "description": "Updated stack file content" + }, + "edge_groups": { + "type": "array", + "items": {"type": "string"}, + "description": "Updated list of edge group IDs" + }, + "update_version": { + "type": "boolean", + "description": "Update the stack version", + "default": True + } + }, + "required": ["edge_stack_id"] + } + ), + types.Tool( + name="delete_edge_stack", + description="Delete an edge stack", + inputSchema={ + "type": "object", + "properties": { + "edge_stack_id": { + "type": "string", + "description": "Edge stack ID" + } + }, + "required": ["edge_stack_id"] + } + ), + + # Edge Group Tools + types.Tool( + name="list_edge_groups", + description="List all edge groups", + inputSchema={ + "type": "object", + "properties": {} + } + ), + types.Tool( + name="get_edge_group", + description="Get detailed information about an edge group", + inputSchema={ + "type": "object", + "properties": { + "edge_group_id": { + "type": "string", + "description": "Edge group ID" + } + }, + "required": ["edge_group_id"] + } + ), + types.Tool( + name="create_edge_group", + description="Create a new edge group", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Edge group name" + }, + "dynamic": { + "type": "boolean", + "description": "Whether this is a dynamic group", + "default": False + }, + "tag_ids": { + "type": "array", + "items": {"type": "string"}, + "description": "Tag IDs for dynamic groups" + }, + "endpoints": { + "type": "array", + "items": {"type": "string"}, + "description": "Endpoint IDs for static groups" + } + }, + "required": ["name"] + } + ), + types.Tool( + name="update_edge_group", + description="Update an edge group", + inputSchema={ + "type": "object", + "properties": { + "edge_group_id": { + "type": "string", + "description": "Edge group ID" + }, + "name": { + "type": "string", + "description": "Updated group name" + }, + "tag_ids": { + "type": "array", + "items": {"type": "string"}, + "description": "Updated tag IDs for dynamic groups" + }, + "endpoints": { + "type": "array", + "items": {"type": "string"}, + "description": "Updated endpoint IDs for static groups" + } + }, + "required": ["edge_group_id"] + } + ), + types.Tool( + name="delete_edge_group", + description="Delete an edge group", + inputSchema={ + "type": "object", + "properties": { + "edge_group_id": { + "type": "string", + "description": "Edge group ID" + } + }, + "required": ["edge_group_id"] + } + ), + + # Edge Job Tools + types.Tool( + name="list_edge_jobs", + description="List all edge jobs", + inputSchema={ + "type": "object", + "properties": {} + } + ), + types.Tool( + name="get_edge_job", + description="Get detailed information about an edge job", + inputSchema={ + "type": "object", + "properties": { + "edge_job_id": { + "type": "string", + "description": "Edge job ID" + } + }, + "required": ["edge_job_id"] + } + ), + types.Tool( + name="create_edge_job", + description="Create a new edge job", + inputSchema={ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Edge job name" + }, + "script_content": { + "type": "string", + "description": "Script content to execute" + }, + "edge_groups": { + "type": "array", + "items": {"type": "string"}, + "description": "Edge group IDs to run the job on" + }, + "cron_expression": { + "type": "string", + "description": "Cron expression for scheduling (optional)" + }, + "recurring": { + "type": "boolean", + "description": "Whether this is a recurring job", + "default": False + } + }, + "required": ["name", "script_content", "edge_groups"] + } + ), + types.Tool( + name="delete_edge_job", + description="Delete an edge job", + inputSchema={ + "type": "object", + "properties": { + "edge_job_id": { + "type": "string", + "description": "Edge job ID" + } + }, + "required": ["edge_job_id"] + } + ), + + # Edge Settings Tools + types.Tool( + name="get_edge_settings", + description="Get global edge settings", + inputSchema={ + "type": "object", + "properties": {} + } + ), + types.Tool( + name="update_edge_settings", + description="Update global edge settings", + inputSchema={ + "type": "object", + "properties": { + "check_in_interval": { + "type": "integer", + "description": "Edge agent check-in interval in seconds" + }, + "command_interval": { + "type": "integer", + "description": "Command execution interval in seconds" + }, + "ping_interval": { + "type": "integer", + "description": "Ping interval in seconds" + }, + "snapshot_interval": { + "type": "integer", + "description": "Snapshot interval in seconds" + }, + "tunnel_server_address": { + "type": "string", + "description": "Tunnel server address" + } + } + } + ) + ] + +@server.call_tool() +async def handle_call_tool( + name: str, + arguments: dict | None +) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: + """Handle tool execution.""" + + if not arguments: + arguments = {} + + try: + # Edge Environment Tools + if name == "list_edge_environments": + # Get all environments and filter for edge type + result = await make_request("GET", "/api/endpoints") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + # Filter for edge environments (Type = 4 or 5) + edge_envs = [env for env in result if env.get("Type") in [4, 5]] + + # Apply additional filters + if arguments.get("group_id"): + edge_envs = [env for env in edge_envs if env.get("GroupId") == int(arguments["group_id"])] + + if arguments.get("status") != "all": + status_filter = arguments.get("status", "all") + if status_filter == "connected": + edge_envs = [env for env in edge_envs if env.get("Status") == 1] + elif status_filter == "disconnected": + edge_envs = [env for env in edge_envs if env.get("Status") == 2] + + if not edge_envs: + return [types.TextContent(type="text", text="No edge environments found")] + + output = "šŸŒ Edge Environments:\n\n" + + for env in edge_envs: + status = format_edge_status(env.get("Status", 0)) + output += f"ā€¢ {env['Name']} (ID: {env['Id']})\n" + output += f" Status: {status}\n" + output += f" URL: {env.get('URL', 'N/A')}\n" + if env.get("EdgeID"): + output += f" Edge ID: {env['EdgeID']}\n" + if env.get("EdgeKey"): + output += f" Edge Key: {env['EdgeKey'][:8]}...\n" + if env.get("GroupId"): + output += f" Group ID: {env['GroupId']}\n" + output += f" Last Check-in: {env.get('LastCheckInDate', 'Never')}\n" + output += "\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "get_edge_environment": + env_id = arguments["environment_id"] + + result = await make_request("GET", f"/api/endpoints/{env_id}") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"šŸŒ Edge Environment: {result['Name']}\n\n" + output += f"ID: {result['Id']}\n" + output += f"Status: {format_edge_status(result.get('Status', 0))}\n" + output += f"Type: {'Edge Agent' if result.get('Type') == 4 else 'Edge Agent (async)'}\n" + output += f"URL: {result.get('URL', 'N/A')}\n" + + if result.get("EdgeID"): + output += f"\nšŸ”‘ Edge Configuration:\n" + output += f" Edge ID: {result['EdgeID']}\n" + output += f" Edge Key: {result.get('EdgeKey', 'N/A')[:8]}...\n" + output += f" Check-in Interval: {result.get('EdgeCheckinInterval', 5)} seconds\n" + + if result.get("GroupId"): + output += f"\nšŸ‘„ Group: {result['GroupId']}\n" + + if result.get("TagIds"): + output += f"\nšŸ·ļø Tags: {', '.join(map(str, result['TagIds']))}\n" + + output += f"\nšŸ“… Timestamps:\n" + output += f" Created: {result.get('CreatedAt', 'Unknown')}\n" + output += f" Last Check-in: {result.get('LastCheckInDate', 'Never')}\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "get_edge_status": + env_id = arguments["environment_id"] + + result = await make_request("GET", f"/api/endpoints/{env_id}/edge/status") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = "šŸ“Š Edge Status Information:\n\n" + output += f"Check-in Time: {result.get('CheckinTime', 'Unknown')}\n" + output += f"Status: {result.get('Status', 'Unknown')}\n" + + if result.get("Stacks"): + output += f"\nšŸ“š Deployed Stacks: {len(result['Stacks'])}\n" + for stack in result["Stacks"]: + output += f" ā€¢ {stack.get('Name', 'Unknown')} (v{stack.get('Version', '?')})\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "generate_edge_key": + edge_name = arguments["name"] + + edge_data = { + "Name": edge_name, + "Type": 4, # Edge Agent + "URL": "", + "PublicURL": "", + "TLS": False + } + + if arguments.get("group_id"): + edge_data["GroupId"] = int(arguments["group_id"]) + + result = await make_request("POST", "/api/endpoints", json_data=edge_data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + edge_id = result.get("Id") + edge_key = result.get("EdgeKey", "") + + output = f"āœ… Edge environment '{edge_name}' created!\n\n" + output += f"ID: {edge_id}\n" + output += f"Edge Key: {edge_key}\n\n" + output += "šŸš€ Deploy Edge Agent with:\n\n" + output += "```bash\n" + output += f"docker run -d \\\n" + output += f" --name portainer_edge_agent \\\n" + output += f" --restart always \\\n" + output += f" --cap-add CAP_HOST_MANAGEMENT \\\n" + output += f" -v /var/run/docker.sock:/var/run/docker.sock \\\n" + output += f" -v /var/lib/docker/volumes:/var/lib/docker/volumes \\\n" + output += f" -v /:/host \\\n" + output += f" -v portainer_agent_data:/data \\\n" + output += f" -e EDGE=1 \\\n" + output += f" -e EDGE_ID={edge_id} \\\n" + output += f" -e EDGE_KEY={edge_key} \\\n" + output += f" -e EDGE_INSECURE_POLL=1 \\\n" + output += f" -e EDGE_ASYNC=1 \\\n" + output += f" portainer/agent:latest\n" + output += "```\n" + + return [types.TextContent(type="text", text=output)] + + # Edge Stack Tools + elif name == "list_edge_stacks": + result = await make_request("GET", "/api/edge_stacks") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + if not result: + return [types.TextContent(type="text", text="No edge stacks found")] + + output = "šŸ“š Edge Stacks:\n\n" + + for stack in result: + output += f"ā€¢ {stack['Name']} (ID: {stack['Id']})\n" + output += f" Version: {stack.get('Version', 1)}\n" + output += f" Type: {'Compose' if stack.get('DeploymentType') == 0 else 'Kubernetes'}\n" + output += f" Edge Groups: {len(stack.get('EdgeGroups', []))}\n" + + if stack.get("Status"): + deployed = sum(1 for s in stack["Status"].values() if s.get("Type") == "OK") + output += f" Deployed: {deployed}/{len(stack['Status'])}\n" + + output += "\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "get_edge_stack": + edge_stack_id = arguments["edge_stack_id"] + + result = await make_request("GET", f"/api/edge_stacks/{edge_stack_id}") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"šŸ“š Edge Stack: {result['Name']}\n\n" + output += f"ID: {result['Id']}\n" + output += f"Version: {result.get('Version', 1)}\n" + output += f"Type: {'Compose' if result.get('DeploymentType') == 0 else 'Kubernetes'}\n" + output += f"Created: {result.get('CreationDate', 'Unknown')}\n" + + if result.get("EdgeGroups"): + output += f"\nšŸ‘„ Edge Groups ({len(result['EdgeGroups'])}):\n" + for group_id in result["EdgeGroups"]: + output += f" ā€¢ Group {group_id}\n" + + if result.get("Status"): + output += f"\nšŸ“Š Deployment Status:\n" + for env_id, status in result["Status"].items(): + output += f" ā€¢ Environment {env_id}: {status.get('Type', 'Unknown')}\n" + if status.get("Error"): + output += f" Error: {status['Error']}\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "create_edge_stack": + data = { + "Name": arguments["name"], + "StackFileContent": arguments["stack_file_content"], + "EdgeGroups": [int(g) for g in arguments["edge_groups"]], + "DeploymentType": arguments.get("deploy_type", 0) + } + + result = await make_request("POST", "/api/edge_stacks", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"āœ… Edge stack created successfully!\n\n" + output += f"Name: {result['Name']}\n" + output += f"ID: {result['Id']}\n" + output += f"Version: {result.get('Version', 1)}\n" + output += f"Edge Groups: {len(result.get('EdgeGroups', []))}\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "update_edge_stack": + edge_stack_id = arguments["edge_stack_id"] + + # Get current stack first + current = await make_request("GET", f"/api/edge_stacks/{edge_stack_id}") + if "error" in current: + return [types.TextContent(type="text", text=f"Error: {current['error']}")] + + data = {} + + if arguments.get("stack_file_content"): + data["StackFileContent"] = arguments["stack_file_content"] + + if arguments.get("edge_groups"): + data["EdgeGroups"] = [int(g) for g in arguments["edge_groups"]] + + if arguments.get("update_version", True): + data["Version"] = current.get("Version", 1) + 1 + + result = await make_request("PUT", f"/api/edge_stacks/{edge_stack_id}", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text=f"āœ… Edge stack '{current['Name']}' updated successfully!")] + + elif name == "delete_edge_stack": + edge_stack_id = arguments["edge_stack_id"] + + result = await make_request("DELETE", f"/api/edge_stacks/{edge_stack_id}") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text="šŸ—‘ļø Edge stack deleted successfully!")] + + # Edge Group Tools + elif name == "list_edge_groups": + result = await make_request("GET", "/api/edge_groups") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + if not result: + return [types.TextContent(type="text", text="No edge groups found")] + + output = "šŸ‘„ Edge Groups:\n\n" + + for group in result: + group_type = format_edge_group_type(2 if group.get("Dynamic") else 1) + output += f"ā€¢ {group['Name']} (ID: {group['Id']})\n" + output += f" Type: {group_type}\n" + + if group.get("Dynamic") and group.get("TagIds"): + output += f" Tags: {len(group['TagIds'])} tags\n" + elif not group.get("Dynamic") and group.get("Endpoints"): + output += f" Endpoints: {len(group['Endpoints'])} environments\n" + + output += "\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "get_edge_group": + edge_group_id = arguments["edge_group_id"] + + result = await make_request("GET", f"/api/edge_groups/{edge_group_id}") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"šŸ‘„ Edge Group: {result['Name']}\n\n" + output += f"ID: {result['Id']}\n" + output += f"Type: {format_edge_group_type(2 if result.get('Dynamic') else 1)}\n" + + if result.get("Dynamic"): + output += f"\nšŸ·ļø Dynamic Tags:\n" + if result.get("TagIds"): + for tag_id in result["TagIds"]: + output += f" ā€¢ Tag {tag_id}\n" + else: + output += f"\nšŸ–„ļø Static Endpoints:\n" + if result.get("Endpoints"): + for endpoint_id in result["Endpoints"]: + output += f" ā€¢ Environment {endpoint_id}\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "create_edge_group": + data = { + "Name": arguments["name"], + "Dynamic": arguments.get("dynamic", False) + } + + if data["Dynamic"] and arguments.get("tag_ids"): + data["TagIds"] = [int(t) for t in arguments["tag_ids"]] + elif not data["Dynamic"] and arguments.get("endpoints"): + data["Endpoints"] = [int(e) for e in arguments["endpoints"]] + + result = await make_request("POST", "/api/edge_groups", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"āœ… Edge group created successfully!\n\n" + output += f"Name: {result['Name']}\n" + output += f"ID: {result['Id']}\n" + output += f"Type: {format_edge_group_type(2 if result.get('Dynamic') else 1)}\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "update_edge_group": + edge_group_id = arguments["edge_group_id"] + + # Get current group + current = await make_request("GET", f"/api/edge_groups/{edge_group_id}") + if "error" in current: + return [types.TextContent(type="text", text=f"Error: {current['error']}")] + + data = current.copy() + + if arguments.get("name"): + data["Name"] = arguments["name"] + + if current.get("Dynamic") and arguments.get("tag_ids"): + data["TagIds"] = [int(t) for t in arguments["tag_ids"]] + elif not current.get("Dynamic") and arguments.get("endpoints"): + data["Endpoints"] = [int(e) for e in arguments["endpoints"]] + + result = await make_request("PUT", f"/api/edge_groups/{edge_group_id}", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text=f"āœ… Edge group '{current['Name']}' updated successfully!")] + + elif name == "delete_edge_group": + edge_group_id = arguments["edge_group_id"] + + result = await make_request("DELETE", f"/api/edge_groups/{edge_group_id}") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text="šŸ—‘ļø Edge group deleted successfully!")] + + # Edge Job Tools + elif name == "list_edge_jobs": + result = await make_request("GET", "/api/edge_jobs") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + if not result: + return [types.TextContent(type="text", text="No edge jobs found")] + + output = "āš™ļø Edge Jobs:\n\n" + + for job in result: + output += f"ā€¢ {job['Name']} (ID: {job['Id']})\n" + output += f" Created: {job.get('Created', 'Unknown')}\n" + + if job.get("CronExpression"): + output += f" Schedule: {job['CronExpression']}\n" + output += f" Recurring: Yes\n" + else: + output += f" Recurring: No\n" + + if job.get("EdgeGroups"): + output += f" Edge Groups: {len(job['EdgeGroups'])}\n" + + output += "\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "get_edge_job": + edge_job_id = arguments["edge_job_id"] + + result = await make_request("GET", f"/api/edge_jobs/{edge_job_id}") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"āš™ļø Edge Job: {result['Name']}\n\n" + output += f"ID: {result['Id']}\n" + output += f"Created: {result.get('Created', 'Unknown')}\n" + output += f"Recurring: {'Yes' if result.get('Recurring') else 'No'}\n" + + if result.get("CronExpression"): + output += f"Schedule: {result['CronExpression']}\n" + + if result.get("EdgeGroups"): + output += f"\nšŸ‘„ Target Edge Groups:\n" + for group_id in result["EdgeGroups"]: + output += f" ā€¢ Group {group_id}\n" + + if result.get("ScriptPath"): + output += f"\nšŸ“œ Script: {result['ScriptPath']}\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "create_edge_job": + data = { + "Name": arguments["name"], + "ScriptContent": arguments["script_content"], + "EdgeGroups": [int(g) for g in arguments["edge_groups"]], + "Recurring": arguments.get("recurring", False) + } + + if arguments.get("cron_expression"): + data["CronExpression"] = arguments["cron_expression"] + data["Recurring"] = True + + result = await make_request("POST", "/api/edge_jobs", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"āœ… Edge job created successfully!\n\n" + output += f"Name: {result['Name']}\n" + output += f"ID: {result['Id']}\n" + output += f"Recurring: {'Yes' if result.get('Recurring') else 'No'}\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "delete_edge_job": + edge_job_id = arguments["edge_job_id"] + + result = await make_request("DELETE", f"/api/edge_jobs/{edge_job_id}") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text="šŸ—‘ļø Edge job deleted successfully!")] + + # Edge Settings Tools + elif name == "get_edge_settings": + result = await make_request("GET", "/api/settings") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = "āš™ļø Edge Settings:\n\n" + + edge_settings = result.get("Edge", {}) + + output += f"Check-in Interval: {edge_settings.get('CheckinInterval', 5)} seconds\n" + output += f"Command Interval: {edge_settings.get('CommandInterval', 5)} seconds\n" + output += f"Ping Interval: {edge_settings.get('PingInterval', 5)} seconds\n" + output += f"Snapshot Interval: {edge_settings.get('SnapshotInterval', 5)} seconds\n" + output += f"Tunnel Server Address: {edge_settings.get('TunnelServerAddress', 'Not configured')}\n" + + return [types.TextContent(type="text", text=output)] + + elif name == "update_edge_settings": + # Get current settings first + current = await make_request("GET", "/api/settings") + if "error" in current: + return [types.TextContent(type="text", text=f"Error: {current['error']}")] + + edge_settings = current.get("Edge", {}) + + # Update only provided settings + if arguments.get("check_in_interval"): + edge_settings["CheckinInterval"] = arguments["check_in_interval"] + if arguments.get("command_interval"): + edge_settings["CommandInterval"] = arguments["command_interval"] + if arguments.get("ping_interval"): + edge_settings["PingInterval"] = arguments["ping_interval"] + if arguments.get("snapshot_interval"): + edge_settings["SnapshotInterval"] = arguments["snapshot_interval"] + if arguments.get("tunnel_server_address"): + edge_settings["TunnelServerAddress"] = arguments["tunnel_server_address"] + + # Update settings + update_data = {"Edge": edge_settings} + result = await make_request("PUT", "/api/settings", json_data=update_data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text="āœ… Edge settings updated successfully!")] + + else: + return [types.TextContent(type="text", text=f"Unknown tool: {name}")] + + except Exception as e: + logger.error(f"Error in {name}: {str(e)}", exc_info=True) + return [types.TextContent(type="text", text=f"Error: {str(e)}")] + +async def main(): + # Run the server using stdin/stdout streams + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="portainer-edge", + server_version="0.1.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + +if __name__ == "__main__": + asyncio.run(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 index de5c578..8d682a9 100644 --- a/src/portainer_core_mcp.egg-info/PKG-INFO +++ b/src/portainer_core_mcp.egg-info/PKG-INFO @@ -2,12 +2,12 @@ Metadata-Version: 2.4 Name: portainer-core-mcp Version: 0.1.0 Summary: Portainer Core MCP Server - Authentication and User Management -Author-email: Your Name +Author-email: Portainer MCP Team License: MIT -Project-URL: Homepage, https://github.com/yourusername/portainer-core-mcp -Project-URL: Documentation, https://github.com/yourusername/portainer-core-mcp#readme -Project-URL: Repository, https://github.com/yourusername/portainer-core-mcp -Project-URL: Issues, https://github.com/yourusername/portainer-core-mcp/issues +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 @@ -41,62 +41,145 @@ Requires-Dist: pre-commit>=3.0.0; extra == "dev" # 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/src/portainer_core_mcp.egg-info/SOURCES.txt b/src/portainer_core_mcp.egg-info/SOURCES.txt index dcad2f4..4dace85 100644 --- a/src/portainer_core_mcp.egg-info/SOURCES.txt +++ b/src/portainer_core_mcp.egg-info/SOURCES.txt @@ -4,11 +4,18 @@ 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 diff --git a/src/portainer_core_mcp.egg-info/entry_points.txt b/src/portainer_core_mcp.egg-info/entry_points.txt index 1d8eacc..dc7da5f 100644 --- a/src/portainer_core_mcp.egg-info/entry_points.txt +++ b/src/portainer_core_mcp.egg-info/entry_points.txt @@ -1,2 +1,2 @@ [console_scripts] -portainer-core-mcp = portainer_core.server:main +portainer-core-mcp = portainer_core.server:main_sync diff --git a/test_edge_api.sh b/test_edge_api.sh new file mode 100755 index 0000000..ccbd170 --- /dev/null +++ b/test_edge_api.sh @@ -0,0 +1,44 @@ +#!/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 new file mode 100755 index 0000000..318ed9f --- /dev/null +++ b/test_edge_server.py @@ -0,0 +1,163 @@ +#!/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