portainer-mcp/portainer_gitops_server.py

#!/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.

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
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.
    
    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",
            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.
    
    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 = {}
    
    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"
                
                # Show polling-specific configuration
                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"
                
                # Show polling-specific configuration
                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())