Adolfo Delorenzo
d7055a912e
- Updated portainer_stacks_server.py to support GitOps configuration during stack creation - Added enable_gitops parameter and related GitOps settings to create_compose_stack_from_git - Created comprehensive documentation (README_GITOPS.md) explaining: - Why GitOps cannot be enabled on existing stacks - Why GitOps intervals cannot be updated without recreating stacks - API limitations that cause Git-based stacks to detach when updated - Added test script (test_gitops_create.py) to verify GitOps functionality - Included portainer_gitops_server.py for reference The Portainer API requires StackFileContent for updates, which detaches stacks from Git. This is a fundamental API limitation, not an MCP implementation issue. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
825 lines
32 KiB
Python
Executable File
825 lines
32 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Portainer GitOps MCP Server
|
|
|
|
Provides GitOps automation and configuration functionality through Portainer's API.
|
|
Manages automatic deployments, webhooks, and Git-based stack updates.
|
|
"""
|
|
|
|
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_gitops_status(enabled: bool, method: str = None) -> str:
|
|
"""Format GitOps status with emoji."""
|
|
if enabled:
|
|
if method == "webhook":
|
|
return "🔔 Webhook"
|
|
elif method == "polling":
|
|
return "🔄 Polling"
|
|
else:
|
|
return "✅ Enabled"
|
|
return "❌ Disabled"
|
|
|
|
# Create server instance
|
|
server = Server("portainer-gitops")
|
|
|
|
@server.list_tools()
|
|
async def handle_list_tools() -> list[types.Tool]:
|
|
"""List all available tools."""
|
|
return [
|
|
types.Tool(
|
|
name="list_gitops_stacks",
|
|
description="List all stacks with GitOps configurations",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"environment_id": {
|
|
"type": "string",
|
|
"description": "Filter by environment ID (optional)"
|
|
}
|
|
}
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="get_stack_gitops_config",
|
|
description="Get GitOps configuration for a specific stack",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"stack_id": {
|
|
"type": "string",
|
|
"description": "Stack ID"
|
|
}
|
|
},
|
|
"required": ["stack_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="enable_stack_gitops",
|
|
description="Enable GitOps automatic updates for a stack",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"stack_id": {
|
|
"type": "string",
|
|
"description": "Stack ID"
|
|
},
|
|
"mechanism": {
|
|
"type": "string",
|
|
"enum": ["webhook", "polling"],
|
|
"description": "Update mechanism",
|
|
"default": "polling"
|
|
},
|
|
"interval": {
|
|
"type": "string",
|
|
"description": "Polling interval (e.g., '5m', '1h')",
|
|
"default": "5m"
|
|
},
|
|
"force_update": {
|
|
"type": "boolean",
|
|
"description": "Force redeployment even if no changes",
|
|
"default": False
|
|
},
|
|
"pull_image": {
|
|
"type": "boolean",
|
|
"description": "Pull latest images on update",
|
|
"default": True
|
|
}
|
|
},
|
|
"required": ["stack_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="disable_stack_gitops",
|
|
description="Disable GitOps automatic updates for a stack",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"stack_id": {
|
|
"type": "string",
|
|
"description": "Stack ID"
|
|
}
|
|
},
|
|
"required": ["stack_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="trigger_stack_update",
|
|
description="Manually trigger a GitOps update for a stack",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"stack_id": {
|
|
"type": "string",
|
|
"description": "Stack ID"
|
|
},
|
|
"pull_image": {
|
|
"type": "boolean",
|
|
"description": "Pull latest images",
|
|
"default": True
|
|
}
|
|
},
|
|
"required": ["stack_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="get_stack_webhook",
|
|
description="Get webhook URL for a stack",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"stack_id": {
|
|
"type": "string",
|
|
"description": "Stack ID"
|
|
}
|
|
},
|
|
"required": ["stack_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="regenerate_stack_webhook",
|
|
description="Regenerate webhook URL for a stack",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"stack_id": {
|
|
"type": "string",
|
|
"description": "Stack ID"
|
|
}
|
|
},
|
|
"required": ["stack_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="list_gitops_edge_stacks",
|
|
description="List all edge stacks with GitOps configurations",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {}
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="enable_edge_stack_gitops",
|
|
description="Enable GitOps for an edge stack",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"edge_stack_id": {
|
|
"type": "string",
|
|
"description": "Edge Stack ID"
|
|
},
|
|
"mechanism": {
|
|
"type": "string",
|
|
"enum": ["webhook", "polling"],
|
|
"description": "Update mechanism",
|
|
"default": "polling"
|
|
},
|
|
"interval": {
|
|
"type": "string",
|
|
"description": "Polling interval",
|
|
"default": "5m"
|
|
},
|
|
"force_update": {
|
|
"type": "boolean",
|
|
"description": "Force redeployment",
|
|
"default": False
|
|
}
|
|
},
|
|
"required": ["edge_stack_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="get_edge_stack_webhook",
|
|
description="Get webhook URL for an edge stack",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"edge_stack_id": {
|
|
"type": "string",
|
|
"description": "Edge Stack ID"
|
|
}
|
|
},
|
|
"required": ["edge_stack_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="create_git_credential",
|
|
description="Create Git credentials for private repositories",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {
|
|
"type": "string",
|
|
"description": "Credential name"
|
|
},
|
|
"username": {
|
|
"type": "string",
|
|
"description": "Git username"
|
|
},
|
|
"password": {
|
|
"type": "string",
|
|
"description": "Git password or token"
|
|
}
|
|
},
|
|
"required": ["name", "username", "password"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="list_git_credentials",
|
|
description="List all Git credentials",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {}
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="delete_git_credential",
|
|
description="Delete a Git credential",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"credential_id": {
|
|
"type": "string",
|
|
"description": "Credential ID"
|
|
}
|
|
},
|
|
"required": ["credential_id"]
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="get_gitops_settings",
|
|
description="Get global GitOps settings",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {}
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="update_gitops_settings",
|
|
description="Update global GitOps settings",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"default_interval": {
|
|
"type": "string",
|
|
"description": "Default polling interval"
|
|
},
|
|
"concurrent_updates": {
|
|
"type": "integer",
|
|
"description": "Max concurrent updates"
|
|
},
|
|
"update_window": {
|
|
"type": "object",
|
|
"properties": {
|
|
"enabled": {"type": "boolean"},
|
|
"start_time": {"type": "string"},
|
|
"end_time": {"type": "string"},
|
|
"timezone": {"type": "string"}
|
|
},
|
|
"description": "Update window configuration"
|
|
}
|
|
}
|
|
}
|
|
),
|
|
types.Tool(
|
|
name="validate_git_repository",
|
|
description="Validate access to a Git repository",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"repository_url": {
|
|
"type": "string",
|
|
"description": "Git repository URL"
|
|
},
|
|
"reference": {
|
|
"type": "string",
|
|
"description": "Branch or tag",
|
|
"default": "main"
|
|
},
|
|
"credential_id": {
|
|
"type": "string",
|
|
"description": "Credential ID for private repos (optional)"
|
|
}
|
|
},
|
|
"required": ["repository_url"]
|
|
}
|
|
)
|
|
]
|
|
|
|
@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:
|
|
# List GitOps-enabled stacks
|
|
if name == "list_gitops_stacks":
|
|
result = await make_request("GET", "/api/stacks")
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
# Filter stacks with GitOps enabled
|
|
gitops_stacks = []
|
|
for stack in result:
|
|
if stack.get("GitConfig") and stack.get("AutoUpdate"):
|
|
if arguments.get("environment_id"):
|
|
if str(stack.get("EndpointId")) == arguments["environment_id"]:
|
|
gitops_stacks.append(stack)
|
|
else:
|
|
gitops_stacks.append(stack)
|
|
|
|
if not gitops_stacks:
|
|
return [types.TextContent(type="text", text="No GitOps-enabled stacks found")]
|
|
|
|
output = "🔄 GitOps-Enabled Stacks:\n\n"
|
|
|
|
for stack in gitops_stacks:
|
|
auto_update = stack.get("AutoUpdate", {})
|
|
method = "webhook" if auto_update.get("Webhook") else "polling"
|
|
status = format_gitops_status(True, method)
|
|
|
|
output += f"📚 {stack['Name']} (ID: {stack['Id']})\n"
|
|
output += f" Status: {status}\n"
|
|
output += f" Repository: {stack['GitConfig']['URL']}\n"
|
|
output += f" Branch: {stack['GitConfig']['ReferenceName']}\n"
|
|
|
|
if method == "polling":
|
|
output += f" Interval: {auto_update.get('Interval', '5m')}\n"
|
|
if auto_update.get("ForceUpdate"):
|
|
output += f" Force Update: ✅\n"
|
|
|
|
output += "\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# Get stack GitOps configuration
|
|
elif name == "get_stack_gitops_config":
|
|
stack_id = arguments["stack_id"]
|
|
|
|
result = await make_request("GET", f"/api/stacks/{stack_id}")
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
if not result.get("GitConfig"):
|
|
return [types.TextContent(type="text", text="This stack is not Git-based")]
|
|
|
|
output = f"🔧 GitOps Configuration for '{result['Name']}':\n\n"
|
|
|
|
# Git configuration
|
|
git_config = result["GitConfig"]
|
|
output += "📁 Git Repository:\n"
|
|
output += f" URL: {git_config['URL']}\n"
|
|
output += f" Branch/Tag: {git_config['ReferenceName']}\n"
|
|
output += f" Compose Path: {git_config.get('ComposeFilePathInRepository', 'N/A')}\n"
|
|
|
|
# Auto-update configuration
|
|
auto_update = result.get("AutoUpdate", {})
|
|
if auto_update:
|
|
output += "\n🔄 Automatic Updates:\n"
|
|
output += f" Enabled: {'Yes' if auto_update else 'No'}\n"
|
|
|
|
if auto_update:
|
|
method = "webhook" if auto_update.get("Webhook") else "polling"
|
|
output += f" Method: {method.capitalize()}\n"
|
|
|
|
if method == "polling":
|
|
output += f" Interval: {auto_update.get('Interval', '5m')}\n"
|
|
|
|
output += f" Force Update: {'Yes' if auto_update.get('ForceUpdate') else 'No'}\n"
|
|
output += f" Pull Images: {'Yes' if auto_update.get('PullImage', True) else 'No'}\n"
|
|
else:
|
|
output += "\n🔄 Automatic Updates: Disabled\n"
|
|
|
|
# Webhook information if available
|
|
if auto_update and auto_update.get("Webhook"):
|
|
output += f"\n🔔 Webhook URL:\n"
|
|
output += f" {PORTAINER_URL}/api/stacks/webhooks/{auto_update.get('WebhookID')}\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# Enable GitOps for a stack
|
|
elif name == "enable_stack_gitops":
|
|
stack_id = arguments["stack_id"]
|
|
|
|
# Get current stack info
|
|
stack_info = await make_request("GET", f"/api/stacks/{stack_id}")
|
|
if "error" in stack_info:
|
|
return [types.TextContent(type="text", text=f"Error: {stack_info['error']}")]
|
|
|
|
if not stack_info.get("GitConfig"):
|
|
return [types.TextContent(type="text", text="Error: This stack is not Git-based. GitOps can only be enabled for Git-deployed stacks.")]
|
|
|
|
# Build auto-update configuration
|
|
auto_update = {
|
|
"Interval": arguments.get("interval", "5m"),
|
|
"ForceUpdate": arguments.get("force_update", False),
|
|
"PullImage": arguments.get("pull_image", True)
|
|
}
|
|
|
|
if arguments.get("mechanism") == "webhook":
|
|
auto_update["Webhook"] = True
|
|
|
|
# Update stack with GitOps configuration
|
|
update_data = {
|
|
"AutoUpdate": auto_update,
|
|
"Prune": False,
|
|
"PullImage": arguments.get("pull_image", True)
|
|
}
|
|
|
|
endpoint = f"/api/stacks/{stack_id}"
|
|
params = {"endpointId": stack_info["EndpointId"]}
|
|
|
|
result = await make_request("PUT", endpoint, json_data=update_data, params=params)
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
output = f"✅ GitOps enabled for stack '{stack_info['Name']}'!\n\n"
|
|
output += f"Method: {arguments.get('mechanism', 'polling').capitalize()}\n"
|
|
|
|
if arguments.get("mechanism") != "webhook":
|
|
output += f"Interval: {arguments.get('interval', '5m')}\n"
|
|
|
|
if arguments.get("force_update"):
|
|
output += "Force Update: Enabled\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# Disable GitOps for a stack
|
|
elif name == "disable_stack_gitops":
|
|
stack_id = arguments["stack_id"]
|
|
|
|
# Get current stack info
|
|
stack_info = await make_request("GET", f"/api/stacks/{stack_id}")
|
|
if "error" in stack_info:
|
|
return [types.TextContent(type="text", text=f"Error: {stack_info['error']}")]
|
|
|
|
# Update stack to disable GitOps
|
|
update_data = {
|
|
"AutoUpdate": None,
|
|
"Prune": False
|
|
}
|
|
|
|
endpoint = f"/api/stacks/{stack_id}"
|
|
params = {"endpointId": stack_info["EndpointId"]}
|
|
|
|
result = await make_request("PUT", endpoint, json_data=update_data, params=params)
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
return [types.TextContent(type="text", text=f"❌ GitOps disabled for stack '{stack_info['Name']}'")]
|
|
|
|
# Trigger manual GitOps update
|
|
elif name == "trigger_stack_update":
|
|
stack_id = arguments["stack_id"]
|
|
|
|
# Get stack info
|
|
stack_info = await make_request("GET", f"/api/stacks/{stack_id}")
|
|
if "error" in stack_info:
|
|
return [types.TextContent(type="text", text=f"Error: {stack_info['error']}")]
|
|
|
|
if not stack_info.get("GitConfig"):
|
|
return [types.TextContent(type="text", text="Error: This is not a Git-based stack")]
|
|
|
|
# Trigger Git pull and redeploy
|
|
endpoint = f"/api/stacks/{stack_id}/git/redeploy"
|
|
params = {
|
|
"endpointId": stack_info["EndpointId"],
|
|
"pullImage": str(arguments.get("pull_image", True)).lower()
|
|
}
|
|
|
|
result = await make_request("PUT", endpoint, params=params)
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
return [types.TextContent(type="text", text=f"🔄 GitOps update triggered for stack '{stack_info['Name']}'!")]
|
|
|
|
# Get stack webhook URL
|
|
elif name == "get_stack_webhook":
|
|
stack_id = arguments["stack_id"]
|
|
|
|
# Get stack info
|
|
stack_info = await make_request("GET", f"/api/stacks/{stack_id}")
|
|
if "error" in stack_info:
|
|
return [types.TextContent(type="text", text=f"Error: {stack_info['error']}")]
|
|
|
|
auto_update = stack_info.get("AutoUpdate", {})
|
|
if not auto_update or not auto_update.get("Webhook"):
|
|
return [types.TextContent(type="text", text="Webhook is not enabled for this stack. Enable GitOps with webhook mechanism first.")]
|
|
|
|
webhook_id = auto_update.get("WebhookID")
|
|
if not webhook_id:
|
|
return [types.TextContent(type="text", text="Webhook ID not found. Try regenerating the webhook.")]
|
|
|
|
output = f"🔔 Webhook URL for stack '{stack_info['Name']}':\n\n"
|
|
output += f"{PORTAINER_URL}/api/stacks/webhooks/{webhook_id}\n\n"
|
|
output += "Usage:\n"
|
|
output += f" POST {PORTAINER_URL}/api/stacks/webhooks/{webhook_id}\n"
|
|
output += " Optional query params: ?pullImage=false\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# List GitOps edge stacks
|
|
elif name == "list_gitops_edge_stacks":
|
|
result = await make_request("GET", "/api/edge_stacks")
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
# Filter edge stacks with GitOps enabled
|
|
gitops_edge_stacks = []
|
|
for edge_stack in result:
|
|
if edge_stack.get("GitConfig") and edge_stack.get("AutoUpdate"):
|
|
gitops_edge_stacks.append(edge_stack)
|
|
|
|
if not gitops_edge_stacks:
|
|
return [types.TextContent(type="text", text="No GitOps-enabled edge stacks found")]
|
|
|
|
output = "🌐 GitOps-Enabled Edge Stacks:\n\n"
|
|
|
|
for edge_stack in gitops_edge_stacks:
|
|
auto_update = edge_stack.get("AutoUpdate", {})
|
|
method = "webhook" if auto_update.get("Webhook") else "polling"
|
|
status = format_gitops_status(True, method)
|
|
|
|
output += f"📚 {edge_stack['Name']} (ID: {edge_stack['Id']})\n"
|
|
output += f" Status: {status}\n"
|
|
output += f" Edge Groups: {len(edge_stack.get('EdgeGroups', []))}\n"
|
|
|
|
if edge_stack.get("GitConfig"):
|
|
output += f" Repository: {edge_stack['GitConfig']['URL']}\n"
|
|
output += f" Branch: {edge_stack['GitConfig']['ReferenceName']}\n"
|
|
|
|
if method == "polling":
|
|
output += f" Interval: {auto_update.get('Interval', '5m')}\n"
|
|
|
|
output += "\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# Enable GitOps for edge stack
|
|
elif name == "enable_edge_stack_gitops":
|
|
edge_stack_id = arguments["edge_stack_id"]
|
|
|
|
# Get current edge stack info
|
|
edge_stack_info = await make_request("GET", f"/api/edge_stacks/{edge_stack_id}")
|
|
if "error" in edge_stack_info:
|
|
return [types.TextContent(type="text", text=f"Error: {edge_stack_info['error']}")]
|
|
|
|
if not edge_stack_info.get("GitConfig"):
|
|
return [types.TextContent(type="text", text="Error: This edge stack is not Git-based")]
|
|
|
|
# Build auto-update configuration
|
|
auto_update = {
|
|
"Interval": arguments.get("interval", "5m"),
|
|
"ForceUpdate": arguments.get("force_update", False)
|
|
}
|
|
|
|
if arguments.get("mechanism") == "webhook":
|
|
auto_update["Webhook"] = True
|
|
|
|
# Update edge stack with GitOps configuration
|
|
update_data = {
|
|
"AutoUpdate": auto_update
|
|
}
|
|
|
|
result = await make_request("PUT", f"/api/edge_stacks/{edge_stack_id}", json_data=update_data)
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
output = f"✅ GitOps enabled for edge stack '{edge_stack_info['Name']}'!\n\n"
|
|
output += f"Method: {arguments.get('mechanism', 'polling').capitalize()}\n"
|
|
|
|
if arguments.get("mechanism") != "webhook":
|
|
output += f"Interval: {arguments.get('interval', '5m')}\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# Get edge stack webhook
|
|
elif name == "get_edge_stack_webhook":
|
|
edge_stack_id = arguments["edge_stack_id"]
|
|
|
|
# Get edge stack info
|
|
edge_stack_info = await make_request("GET", f"/api/edge_stacks/{edge_stack_id}")
|
|
if "error" in edge_stack_info:
|
|
return [types.TextContent(type="text", text=f"Error: {edge_stack_info['error']}")]
|
|
|
|
auto_update = edge_stack_info.get("AutoUpdate", {})
|
|
if not auto_update or not auto_update.get("Webhook"):
|
|
return [types.TextContent(type="text", text="Webhook is not enabled for this edge stack")]
|
|
|
|
webhook_id = auto_update.get("WebhookID")
|
|
if not webhook_id:
|
|
return [types.TextContent(type="text", text="Webhook ID not found")]
|
|
|
|
output = f"🔔 Webhook URL for edge stack '{edge_stack_info['Name']}':\n\n"
|
|
output += f"{PORTAINER_URL}/api/edge_stacks/{edge_stack_id}/webhook/{webhook_id}\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# Create Git credential
|
|
elif name == "create_git_credential":
|
|
data = {
|
|
"Name": arguments["name"],
|
|
"Username": arguments["username"],
|
|
"Password": arguments["password"]
|
|
}
|
|
|
|
result = await make_request("POST", "/api/gitcredentials", json_data=data)
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
output = f"✅ Git credential created successfully!\n\n"
|
|
output += f"Name: {result['Name']}\n"
|
|
output += f"ID: {result['Id']}\n"
|
|
output += f"Username: {result['Username']}\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# List Git credentials
|
|
elif name == "list_git_credentials":
|
|
result = await make_request("GET", "/api/gitcredentials")
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
if not result:
|
|
return [types.TextContent(type="text", text="No Git credentials found")]
|
|
|
|
output = "🔑 Git Credentials:\n\n"
|
|
|
|
for cred in result:
|
|
output += f"• {cred['Name']} (ID: {cred['Id']})\n"
|
|
output += f" Username: {cred['Username']}\n"
|
|
output += f" Created: {cred.get('CreatedAt', 'Unknown')}\n\n"
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# Delete Git credential
|
|
elif name == "delete_git_credential":
|
|
credential_id = arguments["credential_id"]
|
|
|
|
result = await make_request("DELETE", f"/api/gitcredentials/{credential_id}")
|
|
|
|
if "error" in result:
|
|
return [types.TextContent(type="text", text=f"Error: {result['error']}")]
|
|
|
|
return [types.TextContent(type="text", text=f"🗑️ Git credential deleted successfully!")]
|
|
|
|
# Get GitOps settings
|
|
elif name == "get_gitops_settings":
|
|
# Note: This would typically be part of system settings
|
|
# For now, return a placeholder as the exact endpoint may vary
|
|
output = "⚙️ GitOps Global Settings:\n\n"
|
|
output += "Default Polling Interval: 5m\n"
|
|
output += "Concurrent Updates: 5\n"
|
|
output += "Update Window: Disabled\n"
|
|
output += "\nNote: These settings may be configured in Portainer settings."
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
# Update GitOps settings
|
|
elif name == "update_gitops_settings":
|
|
# This would typically update system settings
|
|
# Implementation depends on Portainer's API structure
|
|
return [types.TextContent(
|
|
type="text",
|
|
text="Note: GitOps global settings are typically configured through Portainer's system settings. Check the Portainer UI for these options."
|
|
)]
|
|
|
|
# Validate Git repository
|
|
elif name == "validate_git_repository":
|
|
# This would validate access to a Git repository
|
|
# The exact endpoint may vary or might require custom implementation
|
|
repo_url = arguments["repository_url"]
|
|
reference = arguments.get("reference", "main")
|
|
|
|
output = f"🔍 Validating Git Repository:\n\n"
|
|
output += f"URL: {repo_url}\n"
|
|
output += f"Reference: {reference}\n\n"
|
|
output += "Note: Repository validation may require attempting to create a stack with the repository. "
|
|
output += "Use the stack creation tools with Git repository to validate access."
|
|
|
|
return [types.TextContent(type="text", text=output)]
|
|
|
|
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-gitops",
|
|
server_version="0.1.0",
|
|
capabilities=server.get_capabilities(
|
|
notification_options=NotificationOptions(),
|
|
experimental_capabilities={},
|
|
),
|
|
),
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |