diff --git a/README_KUBERNETES.md b/README_KUBERNETES.md new file mode 100644 index 0000000..306a873 --- /dev/null +++ b/README_KUBERNETES.md @@ -0,0 +1,430 @@ +# Portainer Kubernetes MCP Server + +This MCP server provides comprehensive Kubernetes cluster management capabilities through Portainer's API. + +## Features + +- **Namespace Management**: List, create, and delete namespaces +- **Pod Operations**: List, view, delete pods, and access logs +- **Deployment Management**: Create, scale, update, restart, and delete deployments +- **Service Management**: List, create, and delete services +- **Ingress Configuration**: List and create ingress rules +- **ConfigMap & Secret Management**: Full CRUD operations with base64 encoding +- **Storage Management**: PersistentVolume and PersistentVolumeClaim operations +- **Node Information**: View cluster node details + +## 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_kubernetes_server.py + ``` + +## Configuration + +Add to your Claude Desktop configuration: + +```json +{ + "portainer-kubernetes": { + "command": "python", + "args": ["/path/to/portainer-mcp/portainer_kubernetes_server.py"], + "env": { + "PORTAINER_URL": "https://your-portainer-instance.com", + "PORTAINER_API_KEY": "your-api-key" + } + } +} +``` + +## Available Tools + +### Namespace Management + +#### list_namespaces +List all namespaces in the cluster. +- **Parameters**: + - `environment_id` (required): Target environment ID + +#### create_namespace +Create a new namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `name` (required): Namespace name + +#### delete_namespace +Delete a namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `name` (required): Namespace name + +### Pod Management + +#### list_pods +List pods in a namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (optional): Namespace (default: "default") + - `label_selector` (optional): Label selector (e.g., "app=nginx") + +#### get_pod +Get detailed information about a specific pod. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Pod name + +#### delete_pod +Delete a pod. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Pod name + - `grace_period` (optional): Grace period in seconds + +#### get_pod_logs +Get logs from a pod. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Pod name + - `container` (optional): Container name + - `previous` (optional): Get previous container logs + - `tail_lines` (optional): Number of lines from end + +### Deployment Management + +#### list_deployments +List deployments in a namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (optional): Namespace (default: "default") + +#### get_deployment +Get deployment details. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Deployment name + +#### create_deployment +Create a new deployment. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Deployment name + - `image` (required): Container image + - `replicas` (optional): Number of replicas (default: 1) + - `port` (optional): Container port + - `env_vars` (optional): Environment variables object + - `labels` (optional): Labels object + +#### scale_deployment +Scale a deployment. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Deployment name + - `replicas` (required): Desired replica count + +#### update_deployment_image +Update deployment container image. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Deployment name + - `container` (required): Container name + - `image` (required): New image + +#### restart_deployment +Restart a deployment by updating annotation. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Deployment name + +#### delete_deployment +Delete a deployment. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Deployment name + +### Service Management + +#### list_services +List services in a namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (optional): Namespace (default: "default") + +#### create_service +Create a service for a deployment. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Service name + - `selector` (required): Pod selector object + - `ports` (required): Array of port mappings + - `type` (optional): Service type (ClusterIP, NodePort, LoadBalancer) + +#### delete_service +Delete a service. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Service name + +### Ingress Management + +#### list_ingresses +List ingress rules in a namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (optional): Namespace (default: "default") + +#### create_ingress +Create an ingress rule. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Ingress name + - `rules` (required): Array of ingress rules + - `tls` (optional): TLS configuration array + +### ConfigMap Management + +#### list_configmaps +List ConfigMaps in a namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (optional): Namespace (default: "default") + +#### create_configmap +Create a ConfigMap. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): ConfigMap name + - `data` (required): Key-value data object + +#### delete_configmap +Delete a ConfigMap. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): ConfigMap name + +### Secret Management + +#### list_secrets +List secrets in a namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (optional): Namespace (default: "default") + +#### create_secret +Create a secret (automatically base64 encodes). +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Secret name + - `data` (required): Key-value data object + - `type` (optional): Secret type (default: "Opaque") + +#### delete_secret +Delete a secret. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): Secret name + +### Storage Management + +#### list_persistent_volumes +List PersistentVolumes in the cluster. +- **Parameters**: + - `environment_id` (required): Target environment ID + +#### list_persistent_volume_claims +List PersistentVolumeClaims in a namespace. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (optional): Namespace (default: "default") + +#### create_persistent_volume_claim +Create a PersistentVolumeClaim. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): PVC name + - `storage` (required): Storage size (e.g., "1Gi") + - `access_modes` (optional): Array of access modes + - `storage_class` (optional): Storage class name + +#### delete_persistent_volume_claim +Delete a PersistentVolumeClaim. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `namespace` (required): Namespace + - `name` (required): PVC name + +### Node Management + +#### list_nodes +List cluster nodes with details. +- **Parameters**: + - `environment_id` (required): Target environment ID + +## Usage Examples + +### Create and manage a deployment + +```javascript +// Create a deployment +await use_mcp_tool("portainer-kubernetes", "create_deployment", { + environment_id: "2", + namespace: "production", + name: "nginx-app", + image: "nginx:latest", + replicas: 3, + port: 80, + labels: { app: "nginx", tier: "frontend" } +}); + +// Scale the deployment +await use_mcp_tool("portainer-kubernetes", "scale_deployment", { + environment_id: "2", + namespace: "production", + name: "nginx-app", + replicas: 5 +}); + +// Create a service for the deployment +await use_mcp_tool("portainer-kubernetes", "create_service", { + environment_id: "2", + namespace: "production", + name: "nginx-service", + selector: { app: "nginx" }, + ports: [{ port: 80, targetPort: 80 }], + type: "LoadBalancer" +}); +``` + +### Manage ConfigMaps and Secrets + +```javascript +// Create a ConfigMap +await use_mcp_tool("portainer-kubernetes", "create_configmap", { + environment_id: "2", + namespace: "production", + name: "app-config", + data: { + "app.properties": "debug=false\nport=8080", + "database.conf": "host=db.example.com" + } +}); + +// Create a Secret (automatically base64 encoded) +await use_mcp_tool("portainer-kubernetes", "create_secret", { + environment_id: "2", + namespace: "production", + name: "db-credentials", + data: { + username: "dbuser", + password: "secretpassword" + } +}); +``` + +### Storage operations + +```javascript +// Create a PersistentVolumeClaim +await use_mcp_tool("portainer-kubernetes", "create_persistent_volume_claim", { + environment_id: "2", + namespace: "production", + name: "data-storage", + storage: "10Gi", + access_modes: ["ReadWriteOnce"], + storage_class: "fast-ssd" +}); +``` + +### Pod troubleshooting + +```javascript +// Get pod logs +await use_mcp_tool("portainer-kubernetes", "get_pod_logs", { + environment_id: "2", + namespace: "production", + name: "nginx-app-7d9c5b5b6-abc123", + tail_lines: 100 +}); + +// Get pod details +await use_mcp_tool("portainer-kubernetes", "get_pod", { + environment_id: "2", + namespace: "production", + name: "nginx-app-7d9c5b5b6-abc123" +}); +``` + +## Error Handling + +The server includes comprehensive error handling: +- Network timeouts and retries +- Invalid Kubernetes resource specifications +- Authentication failures +- Resource not found errors +- Namespace conflicts + +All errors are returned with descriptive messages to help diagnose issues. + +## Security Notes + +- The server automatically handles base64 encoding for Kubernetes secrets +- API tokens are never logged or exposed +- All communications use HTTPS when configured +- Follows Kubernetes RBAC permissions + +## Troubleshooting + +### Common Issues + +1. **Authentication failures**: Ensure your API key is valid and has appropriate permissions +2. **Resource not found**: Verify the namespace and resource names +3. **Permission denied**: Check RBAC permissions for the service account +4. **Timeout errors**: Increase HTTP_TIMEOUT in environment variables + +### Debug Mode + +Enable debug logging by setting in your environment: +```bash +DEBUG=true +LOG_LEVEL=DEBUG +``` + +## Requirements + +- Python 3.8+ +- Portainer Business Edition 2.19+ with Kubernetes endpoints +- Valid Portainer API token +- Kubernetes cluster connected to Portainer \ No newline at end of file diff --git a/README_STACKS.md b/README_STACKS.md new file mode 100644 index 0000000..ed7c4fa --- /dev/null +++ b/README_STACKS.md @@ -0,0 +1,353 @@ +# Portainer Stacks MCP Server + +This MCP server provides comprehensive stack deployment and management capabilities through Portainer's API, supporting both Docker Compose and Kubernetes deployments. + +## Features + +- **Stack Management**: List, create, update, start, stop, and delete stacks +- **Docker Compose Support**: Deploy stacks from file content or Git repositories +- **Kubernetes Support**: Deploy Kubernetes manifests as stacks +- **Git Integration**: Create and update stacks directly from Git repositories +- **Environment Variables**: Manage stack environment variables +- **Stack Migration**: Copy stacks between environments +- **Multi-Environment**: Work with stacks across different Portainer environments + +## 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_stacks_server.py + ``` + +## Configuration + +Add to your Claude Desktop configuration: + +```json +{ + "portainer-stacks": { + "command": "python", + "args": ["/path/to/portainer-mcp/portainer_stacks_server.py"], + "env": { + "PORTAINER_URL": "https://your-portainer-instance.com", + "PORTAINER_API_KEY": "your-api-key" + } + } +} +``` + +## Available Tools + +### Stack Information + +#### list_stacks +List all stacks across environments. +- **Parameters**: + - `environment_id` (optional): Filter by environment ID + +#### get_stack +Get detailed information about a specific stack. +- **Parameters**: + - `stack_id` (required): Stack ID + +#### get_stack_file +Get the stack file content (Docker Compose or Kubernetes manifest). +- **Parameters**: + - `stack_id` (required): Stack ID + +### Stack Creation + +#### create_compose_stack_from_file +Create a new Docker Compose stack from file content. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `name` (required): Stack name + - `compose_file` (required): Docker Compose file content (YAML) + - `env_vars` (optional): Array of environment variables + - `name`: Variable name + - `value`: Variable value + +#### create_compose_stack_from_git +Create a new Docker Compose stack from Git repository. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `name` (required): Stack name + - `repository_url` (required): Git repository URL + - `repository_ref` (optional): Git reference (default: "main") + - `compose_path` (optional): Path to compose file (default: "docker-compose.yml") + - `repository_auth` (optional): Use authentication (default: false) + - `repository_username` (optional): Git username + - `repository_password` (optional): Git password/token + - `env_vars` (optional): Array of environment variables + +#### create_kubernetes_stack +Create a new Kubernetes stack from manifest. +- **Parameters**: + - `environment_id` (required): Target environment ID + - `name` (required): Stack name + - `namespace` (optional): Kubernetes namespace (default: "default") + - `manifest` (required): Kubernetes manifest content (YAML) + +### Stack Management + +#### update_stack +Update an existing stack. +- **Parameters**: + - `stack_id` (required): Stack ID + - `compose_file` (optional): Updated compose file or manifest + - `env_vars` (optional): Updated environment variables array + - `pull_image` (optional): Pull latest images (default: true) + +#### update_git_stack +Update a Git-based stack (pull latest changes). +- **Parameters**: + - `stack_id` (required): Stack ID + - `pull_image` (optional): Pull latest images (default: true) + +#### start_stack +Start a stopped stack. +- **Parameters**: + - `stack_id` (required): Stack ID + +#### stop_stack +Stop a running stack. +- **Parameters**: + - `stack_id` (required): Stack ID + +#### delete_stack +Delete a stack and optionally its volumes. +- **Parameters**: + - `stack_id` (required): Stack ID + - `delete_volumes` (optional): Delete associated volumes (default: false) + +### Stack Operations + +#### migrate_stack +Migrate a stack to another environment. +- **Parameters**: + - `stack_id` (required): Stack ID + - `target_environment_id` (required): Target environment ID + - `new_name` (optional): New stack name + +#### get_stack_logs +Get logs from all containers in a stack. +- **Parameters**: + - `stack_id` (required): Stack ID + - `tail` (optional): Number of lines from end (default: 100) + - `timestamps` (optional): Show timestamps (default: true) + +## Usage Examples + +### Deploy a Docker Compose stack + +```javascript +// From file content +await use_mcp_tool("portainer-stacks", "create_compose_stack_from_file", { + environment_id: "2", + name: "wordpress-blog", + compose_file: `version: '3.8' +services: + wordpress: + image: wordpress:latest + ports: + - "8080:80" + environment: + WORDPRESS_DB_HOST: db + WORDPRESS_DB_USER: wordpress + WORDPRESS_DB_PASSWORD: \${DB_PASSWORD} + volumes: + - wordpress_data:/var/www/html + + db: + image: mysql:5.7 + environment: + MYSQL_DATABASE: wordpress + MYSQL_USER: wordpress + MYSQL_PASSWORD: \${DB_PASSWORD} + MYSQL_RANDOM_ROOT_PASSWORD: '1' + volumes: + - db_data:/var/lib/mysql + +volumes: + wordpress_data: + db_data:`, + env_vars: [ + { name: "DB_PASSWORD", value: "secure_password" } + ] +}); + +// From Git repository +await use_mcp_tool("portainer-stacks", "create_compose_stack_from_git", { + environment_id: "2", + name: "microservices-app", + repository_url: "https://github.com/myorg/microservices.git", + repository_ref: "main", + compose_path: "docker/docker-compose.prod.yml", + env_vars: [ + { name: "ENVIRONMENT", value: "production" }, + { name: "API_KEY", value: "secret_key" } + ] +}); +``` + +### Deploy a Kubernetes stack + +```javascript +await use_mcp_tool("portainer-stacks", "create_kubernetes_stack", { + environment_id: "3", + name: "nginx-deployment", + namespace: "production", + manifest: `apiVersion: apps/v1 +kind: Deployment +metadata: + name: nginx-deployment +spec: + replicas: 3 + selector: + matchLabels: + app: nginx + template: + metadata: + labels: + app: nginx + spec: + containers: + - name: nginx + image: nginx:1.21 + ports: + - containerPort: 80 +--- +apiVersion: v1 +kind: Service +metadata: + name: nginx-service +spec: + selector: + app: nginx + ports: + - port: 80 + targetPort: 80 + type: LoadBalancer` +}); +``` + +### Manage stacks + +```javascript +// Update a stack +await use_mcp_tool("portainer-stacks", "update_stack", { + stack_id: "5", + compose_file: "updated compose content...", + env_vars: [ + { name: "VERSION", value: "2.0" } + ], + pull_image: true +}); + +// Update Git-based stack +await use_mcp_tool("portainer-stacks", "update_git_stack", { + stack_id: "7", + pull_image: true +}); + +// Stop and start stacks +await use_mcp_tool("portainer-stacks", "stop_stack", { + stack_id: "5" +}); + +await use_mcp_tool("portainer-stacks", "start_stack", { + stack_id: "5" +}); + +// Delete stack with volumes +await use_mcp_tool("portainer-stacks", "delete_stack", { + stack_id: "5", + delete_volumes: true +}); +``` + +### Migrate stack between environments + +```javascript +await use_mcp_tool("portainer-stacks", "migrate_stack", { + stack_id: "5", + target_environment_id: "4", + new_name: "wordpress-prod" +}); +``` + +## Stack Types + +The server supports three types of stacks: + +1. **Swarm Stacks** (Type 1): Docker Swarm mode stacks +2. **Compose Stacks** (Type 2): Standard Docker Compose deployments +3. **Kubernetes Stacks** (Type 3): Kubernetes manifest deployments + +## Error Handling + +The server includes comprehensive error handling: +- Invalid stack configurations +- Git repository access errors +- Environment permission issues +- Network timeouts and retries +- Resource conflicts + +All errors are returned with descriptive messages to help diagnose issues. + +## Security Notes + +- Git credentials are transmitted securely but stored in Portainer +- Environment variables may contain sensitive data +- Stack files can include secrets - handle with care +- API tokens are never logged or exposed +- Use RBAC to control stack access + +## Best Practices + +1. **Environment Variables**: Use environment variables for configuration instead of hardcoding values +2. **Git Integration**: Use Git repositories for version control and automated deployments +3. **Naming Convention**: Use consistent naming for stacks across environments +4. **Volume Management**: Be careful when deleting stacks with volumes +5. **Migration Testing**: Test stack migrations in non-production environments first + +## Troubleshooting + +### Common Issues + +1. **Stack creation fails**: Check compose file syntax and image availability +2. **Git authentication errors**: Ensure credentials are correct and have repository access +3. **Permission denied**: Verify user has appropriate Portainer permissions +4. **Stack update fails**: Check for resource conflicts or invalid configurations + +### Debug Mode + +Enable debug logging by setting in your environment: +```bash +DEBUG=true +LOG_LEVEL=DEBUG +``` + +## Requirements + +- Python 3.8+ +- Portainer Business Edition 2.19+ +- Valid Portainer API token with stack management permissions +- Docker or Kubernetes environments configured in Portainer \ No newline at end of file diff --git a/portainer_kubernetes_server.py b/portainer_kubernetes_server.py new file mode 100755 index 0000000..5527efa --- /dev/null +++ b/portainer_kubernetes_server.py @@ -0,0 +1,1847 @@ +#!/usr/bin/env python3 +"""MCP server for Portainer Kubernetes management.""" + +import os +import sys +import json +import asyncio +from typing import Any, Dict, List, Optional +from enum import Enum +import base64 + +# Suppress all logging to stderr +os.environ["MCP_MODE"] = "true" + +import mcp.server.stdio +import mcp.types as types +from mcp.server import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +# Create server +server = Server("portainer-kubernetes") + +# Store for our state +portainer_url = os.getenv("PORTAINER_URL", "https://partner.portainer.live") +api_key = os.getenv("PORTAINER_API_KEY", "") + + +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + # Namespace Management + types.Tool( + name="list_namespaces", + description="List all namespaces in a Kubernetes cluster", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_namespace", + description="Create a new namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "name": { + "type": "string", + "description": "Namespace name" + }, + "labels": { + "type": "object", + "description": "Labels for the namespace (optional)" + } + }, + "required": ["environment_id", "name"] + } + ), + types.Tool( + name="delete_namespace", + description="Delete a namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name" + } + }, + "required": ["environment_id", "namespace"] + } + ), + # Pod Management + types.Tool( + name="list_pods", + description="List pods in a namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "label_selector": { + "type": "string", + "description": "Label selector to filter pods (optional)" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="get_pod", + description="Get detailed information about a pod", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "pod_name": { + "type": "string", + "description": "Pod name" + } + }, + "required": ["environment_id", "pod_name"] + } + ), + types.Tool( + name="delete_pod", + description="Delete a pod", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "pod_name": { + "type": "string", + "description": "Pod name" + } + }, + "required": ["environment_id", "pod_name"] + } + ), + types.Tool( + name="get_pod_logs", + description="Get logs from a pod", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "pod_name": { + "type": "string", + "description": "Pod name" + }, + "container": { + "type": "string", + "description": "Container name (optional if pod has single container)" + }, + "tail": { + "type": "integer", + "description": "Number of lines from the end", + "default": 100 + } + }, + "required": ["environment_id", "pod_name"] + } + ), + # Deployment Management + types.Tool( + name="list_deployments", + description="List deployments in a namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="get_deployment", + description="Get detailed information about a deployment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "deployment_name": { + "type": "string", + "description": "Deployment name" + } + }, + "required": ["environment_id", "deployment_name"] + } + ), + types.Tool( + name="create_deployment", + description="Create a new deployment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "name": { + "type": "string", + "description": "Deployment name" + }, + "image": { + "type": "string", + "description": "Container image" + }, + "replicas": { + "type": "integer", + "description": "Number of replicas", + "default": 1 + }, + "port": { + "type": "integer", + "description": "Container port (optional)" + }, + "env": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "value": {"type": "string"} + } + }, + "description": "Environment variables (optional)" + }, + "labels": { + "type": "object", + "description": "Labels for the deployment (optional)" + } + }, + "required": ["environment_id", "name", "image"] + } + ), + types.Tool( + name="scale_deployment", + description="Scale a deployment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "deployment_name": { + "type": "string", + "description": "Deployment name" + }, + "replicas": { + "type": "integer", + "description": "Number of replicas" + } + }, + "required": ["environment_id", "deployment_name", "replicas"] + } + ), + types.Tool( + name="update_deployment_image", + description="Update deployment container image", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "deployment_name": { + "type": "string", + "description": "Deployment name" + }, + "container_name": { + "type": "string", + "description": "Container name" + }, + "new_image": { + "type": "string", + "description": "New container image" + } + }, + "required": ["environment_id", "deployment_name", "container_name", "new_image"] + } + ), + types.Tool( + name="restart_deployment", + description="Restart a deployment by updating annotation", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "deployment_name": { + "type": "string", + "description": "Deployment name" + } + }, + "required": ["environment_id", "deployment_name"] + } + ), + types.Tool( + name="delete_deployment", + description="Delete a deployment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "deployment_name": { + "type": "string", + "description": "Deployment name" + } + }, + "required": ["environment_id", "deployment_name"] + } + ), + # Service Management + types.Tool( + name="list_services", + description="List services in a namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_service", + description="Create a service for a deployment", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "name": { + "type": "string", + "description": "Service name" + }, + "selector": { + "type": "object", + "description": "Pod selector labels" + }, + "ports": { + "type": "array", + "items": { + "type": "object", + "properties": { + "port": {"type": "integer"}, + "targetPort": {"type": "integer"}, + "protocol": {"type": "string", "default": "TCP"} + } + }, + "description": "Service ports" + }, + "type": { + "type": "string", + "enum": ["ClusterIP", "NodePort", "LoadBalancer"], + "default": "ClusterIP" + } + }, + "required": ["environment_id", "name", "selector", "ports"] + } + ), + types.Tool( + name="delete_service", + description="Delete a service", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "service_name": { + "type": "string", + "description": "Service name" + } + }, + "required": ["environment_id", "service_name"] + } + ), + # Ingress Management + types.Tool( + name="list_ingresses", + description="List ingresses in a namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_ingress", + description="Create an ingress", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "name": { + "type": "string", + "description": "Ingress name" + }, + "rules": { + "type": "array", + "items": { + "type": "object", + "properties": { + "host": {"type": "string"}, + "paths": { + "type": "array", + "items": { + "type": "object", + "properties": { + "path": {"type": "string"}, + "service": {"type": "string"}, + "port": {"type": "integer"} + } + } + } + } + }, + "description": "Ingress rules" + } + }, + "required": ["environment_id", "name", "rules"] + } + ), + # ConfigMap and Secret Management + types.Tool( + name="list_configmaps", + description="List ConfigMaps in a namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_configmap", + description="Create a ConfigMap", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "name": { + "type": "string", + "description": "ConfigMap name" + }, + "data": { + "type": "object", + "description": "Key-value pairs for the ConfigMap" + } + }, + "required": ["environment_id", "name", "data"] + } + ), + types.Tool( + name="list_secrets", + description="List Secrets in a namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_secret", + description="Create a Secret", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "name": { + "type": "string", + "description": "Secret name" + }, + "type": { + "type": "string", + "description": "Secret type", + "default": "Opaque" + }, + "data": { + "type": "object", + "description": "Key-value pairs (values will be base64 encoded)" + } + }, + "required": ["environment_id", "name", "data"] + } + ), + # Persistent Volume Management + types.Tool( + name="list_persistent_volumes", + description="List PersistentVolumes", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="list_persistent_volume_claims", + description="List PersistentVolumeClaims in a namespace", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="create_persistent_volume_claim", + description="Create a PersistentVolumeClaim", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "namespace": { + "type": "string", + "description": "Namespace name", + "default": "default" + }, + "name": { + "type": "string", + "description": "PVC name" + }, + "storage_class": { + "type": "string", + "description": "Storage class name (optional)" + }, + "size": { + "type": "string", + "description": "Storage size (e.g., '10Gi')" + }, + "access_mode": { + "type": "string", + "enum": ["ReadWriteOnce", "ReadOnlyMany", "ReadWriteMany"], + "default": "ReadWriteOnce" + } + }, + "required": ["environment_id", "name", "size"] + } + ), + # Node Information + types.Tool( + name="list_nodes", + description="List cluster nodes", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + } + }, + "required": ["environment_id"] + } + ), + types.Tool( + name="get_node", + description="Get detailed information about a node", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "integer", + "description": "ID of the Kubernetes environment" + }, + "node_name": { + "type": "string", + "description": "Node name" + } + }, + "required": ["environment_id", "node_name"] + } + ) + ] + + +async def make_request(method: str, endpoint: str, json_data: Optional[Dict] = None, + params: Optional[Dict] = None, text_response: bool = False) -> Dict[str, Any]: + """Make HTTP request to Portainer API.""" + import httpx + + async with httpx.AsyncClient(verify=False, timeout=30.0) as client: + headers = {"X-API-Key": api_key} if api_key else {} + + # Add JSON content type for POST/PUT requests + if method in ["POST", "PUT", "PATCH"] and json_data is not None: + headers["Content-Type"] = "application/json" + + if method == "GET": + response = await client.get(f"{portainer_url}/api{endpoint}", headers=headers, params=params) + elif method == "POST": + response = await client.post(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "PUT": + response = await client.put(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "PATCH": + response = await client.patch(f"{portainer_url}/api{endpoint}", headers=headers, json=json_data) + elif method == "DELETE": + response = await client.delete(f"{portainer_url}/api{endpoint}", headers=headers) + else: + raise ValueError(f"Unsupported method: {method}") + + # Handle text responses (like logs) + if text_response: + return {"status_code": response.status_code, "data": None, "text": response.text} + + # Parse JSON response safely + try: + data = response.json() if response.text and response.headers.get("content-type", "").startswith("application/json") else None + except Exception: + data = None + + return {"status_code": response.status_code, "data": data, "text": response.text} + + +def format_pod_phase(phase: str) -> str: + """Format pod phase for display.""" + phase_map = { + "Pending": "ā³ Pending", + "Running": "šŸŸ¢ Running", + "Succeeded": "āœ… Succeeded", + "Failed": "āŒ Failed", + "Unknown": "ā“ Unknown" + } + return phase_map.get(phase, phase) + + +def format_resource_quantity(quantity: str) -> str: + """Format Kubernetes resource quantities.""" + if quantity.endswith("Ki"): + kb = int(quantity[:-2]) + return f"{kb / 1024:.1f} Mi" + elif quantity.endswith("Mi"): + return quantity + elif quantity.endswith("Gi"): + return quantity + elif quantity.endswith("m"): # millicores + cores = int(quantity[:-1]) / 1000 + return f"{cores:.2f} cores" + return quantity + + +@server.call_tool() +async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]: + """Handle tool calls.""" + import httpx + + try: + env_id = arguments.get("environment_id") if arguments else None + namespace = arguments.get("namespace", "default") if arguments else "default" + + # Namespace Management + if name == "list_namespaces": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} namespace(s):\n" + + for ns in items: + metadata = ns.get("metadata", {}) + status = ns.get("status", {}) + output += f"\n- {metadata.get('name')}" + output += f"\n Status: {status.get('phase', 'Unknown')}" + output += f"\n Created: {metadata.get('creationTimestamp', 'N/A')}" + + labels = metadata.get("labels", {}) + if labels: + label_str = ", ".join([f"{k}={v}" for k, v in labels.items()]) + output += f"\n Labels: {label_str}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list namespaces: HTTP {result['status_code']}")] + + elif name == "create_namespace": + ns_name = arguments.get("name") + if not env_id or not ns_name: + return [types.TextContent(type="text", text="Error: environment_id and name are required")] + + ns_data = { + "apiVersion": "v1", + "kind": "Namespace", + "metadata": { + "name": ns_name + } + } + + if arguments.get("labels"): + ns_data["metadata"]["labels"] = arguments["labels"] + + result = await make_request("POST", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces", json_data=ns_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"āœ“ Namespace '{ns_name}' created successfully")] + else: + error_msg = f"Failed to create namespace: HTTP {result['status_code']}" + if result.get("data") and result["data"].get("message"): + error_msg += f"\n{result['data']['message']}" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "delete_namespace": + ns_name = arguments.get("namespace") + if not env_id or not ns_name: + return [types.TextContent(type="text", text="Error: environment_id and namespace are required")] + + result = await make_request("DELETE", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{ns_name}") + + if result["status_code"] in [200, 202]: + return [types.TextContent(type="text", text=f"āœ“ Namespace '{ns_name}' deletion initiated")] + else: + return [types.TextContent(type="text", text=f"Failed to delete namespace: HTTP {result['status_code']}")] + + # Pod Management + elif name == "list_pods": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + params = {} + if arguments.get("label_selector"): + params["labelSelector"] = arguments["label_selector"] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/pods", params=params) + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} pod(s) in namespace '{namespace}':\n" + + for pod in items[:20]: # Limit to 20 pods + metadata = pod.get("metadata", {}) + spec = pod.get("spec", {}) + status = pod.get("status", {}) + + pod_name = metadata.get("name", "unknown") + phase = status.get("phase", "Unknown") + node = spec.get("nodeName", "unscheduled") + + # Count ready containers + container_statuses = status.get("containerStatuses", []) + ready_count = sum(1 for c in container_statuses if c.get("ready", False)) + total_count = len(container_statuses) + + output += f"\n- {pod_name}" + output += f"\n Status: {format_pod_phase(phase)} ({ready_count}/{total_count} ready)" + output += f"\n Node: {node}" + + # Show restart count if any + restart_count = sum(c.get("restartCount", 0) for c in container_statuses) + if restart_count > 0: + output += f"\n Restarts: {restart_count}" + + # Show IP if available + pod_ip = status.get("podIP") + if pod_ip: + output += f"\n IP: {pod_ip}" + + if len(items) > 20: + output += f"\n\n... and {len(items) - 20} more pods" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list pods: HTTP {result['status_code']}")] + + elif name == "get_pod": + pod_name = arguments.get("pod_name") + if not env_id or not pod_name: + return [types.TextContent(type="text", text="Error: environment_id and pod_name are required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/pods/{pod_name}") + + if result["status_code"] == 200 and result["data"]: + pod = result["data"] + metadata = pod.get("metadata", {}) + spec = pod.get("spec", {}) + status = pod.get("status", {}) + + output = f"Pod Details:\n" + output += f"- Name: {metadata.get('name')}\n" + output += f"- Namespace: {metadata.get('namespace')}\n" + output += f"- Phase: {format_pod_phase(status.get('phase'))}\n" + output += f"- Node: {spec.get('nodeName', 'unscheduled')}\n" + output += f"- IP: {status.get('podIP', 'N/A')}\n" + output += f"- Created: {metadata.get('creationTimestamp')}\n" + + # Container details + containers = spec.get("containers", []) + if containers: + output += "\nContainers:\n" + for i, container in enumerate(containers): + container_status = status.get("containerStatuses", [])[i] if i < len(status.get("containerStatuses", [])) else {} + output += f"- {container.get('name')}\n" + output += f" Image: {container.get('image')}\n" + output += f" Ready: {container_status.get('ready', False)}\n" + output += f" Restarts: {container_status.get('restartCount', 0)}\n" + + # Resources + resources = container.get("resources", {}) + if resources.get("requests") or resources.get("limits"): + output += " Resources:\n" + if resources.get("requests"): + output += f" Requests: CPU={resources['requests'].get('cpu', 'N/A')}, Memory={resources['requests'].get('memory', 'N/A')}\n" + if resources.get("limits"): + output += f" Limits: CPU={resources['limits'].get('cpu', 'N/A')}, Memory={resources['limits'].get('memory', 'N/A')}\n" + + # Events (recent) + events_result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/events?fieldSelector=involvedObject.name={pod_name}") + if events_result["status_code"] == 200 and events_result["data"]: + events = events_result["data"].get("items", []) + if events: + output += "\nRecent Events:\n" + for event in events[-5:]: # Last 5 events + output += f"- {event.get('type')}: {event.get('reason')} - {event.get('message')}\n" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get pod: HTTP {result['status_code']}")] + + elif name == "delete_pod": + pod_name = arguments.get("pod_name") + if not env_id or not pod_name: + return [types.TextContent(type="text", text="Error: environment_id and pod_name are required")] + + result = await make_request("DELETE", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/pods/{pod_name}") + + if result["status_code"] in [200, 202]: + return [types.TextContent(type="text", text=f"āœ“ Pod '{pod_name}' deleted successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to delete pod: HTTP {result['status_code']}")] + + elif name == "get_pod_logs": + pod_name = arguments.get("pod_name") + if not env_id or not pod_name: + return [types.TextContent(type="text", text="Error: environment_id and pod_name are required")] + + params = { + "tailLines": arguments.get("tail", 100) + } + + if arguments.get("container"): + params["container"] = arguments["container"] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/pods/{pod_name}/log", + params=params, text_response=True) + + if result["status_code"] == 200: + logs = result.get("text", "") + if logs: + output = f"Pod logs (last {arguments.get('tail', 100)} lines):\n" + output += "-" * 50 + "\n" + output += logs + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text="No logs available")] + else: + return [types.TextContent(type="text", text=f"Failed to get logs: HTTP {result['status_code']}")] + + # Deployment Management + elif name == "list_deployments": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} deployment(s) in namespace '{namespace}':\n" + + for deployment in items: + metadata = deployment.get("metadata", {}) + spec = deployment.get("spec", {}) + status = deployment.get("status", {}) + + dep_name = metadata.get("name") + replicas = spec.get("replicas", 0) + ready_replicas = status.get("readyReplicas", 0) + available_replicas = status.get("availableReplicas", 0) + + output += f"\n- {dep_name}" + output += f"\n Replicas: {ready_replicas}/{replicas} ready, {available_replicas} available" + + # Container images + containers = spec.get("template", {}).get("spec", {}).get("containers", []) + if containers: + images = [c.get("image") for c in containers] + output += f"\n Images: {', '.join(images)}" + + # Conditions + conditions = status.get("conditions", []) + for condition in conditions: + if condition.get("type") == "Progressing" and condition.get("status") != "True": + output += f"\n āš ļø {condition.get('reason', 'Progressing issue')}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list deployments: HTTP {result['status_code']}")] + + elif name == "get_deployment": + deployment_name = arguments.get("deployment_name") + if not env_id or not deployment_name: + return [types.TextContent(type="text", text="Error: environment_id and deployment_name are required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments/{deployment_name}") + + if result["status_code"] == 200 and result["data"]: + deployment = result["data"] + metadata = deployment.get("metadata", {}) + spec = deployment.get("spec", {}) + status = deployment.get("status", {}) + + output = f"Deployment Details:\n" + output += f"- Name: {metadata.get('name')}\n" + output += f"- Namespace: {metadata.get('namespace')}\n" + output += f"- Created: {metadata.get('creationTimestamp')}\n" + output += f"- Replicas: {spec.get('replicas', 0)} desired\n" + output += f"- Ready: {status.get('readyReplicas', 0)}/{spec.get('replicas', 0)}\n" + output += f"- Available: {status.get('availableReplicas', 0)}\n" + output += f"- Up-to-date: {status.get('updatedReplicas', 0)}\n" + + # Deployment strategy + strategy = spec.get("strategy", {}) + output += f"\nStrategy: {strategy.get('type', 'Unknown')}\n" + + # Container details + containers = spec.get("template", {}).get("spec", {}).get("containers", []) + if containers: + output += "\nContainers:\n" + for container in containers: + output += f"- {container.get('name')}\n" + output += f" Image: {container.get('image')}\n" + + # Ports + ports = container.get("ports", []) + if ports: + port_str = ", ".join([f"{p.get('containerPort')}/{p.get('protocol', 'TCP')}" for p in ports]) + output += f" Ports: {port_str}\n" + + # Resources + resources = container.get("resources", {}) + if resources.get("requests") or resources.get("limits"): + output += " Resources:\n" + if resources.get("requests"): + output += f" Requests: CPU={resources['requests'].get('cpu', 'N/A')}, Memory={resources['requests'].get('memory', 'N/A')}\n" + if resources.get("limits"): + output += f" Limits: CPU={resources['limits'].get('cpu', 'N/A')}, Memory={resources['limits'].get('memory', 'N/A')}\n" + + # Conditions + conditions = status.get("conditions", []) + if conditions: + output += "\nConditions:\n" + for condition in conditions: + output += f"- {condition.get('type')}: {condition.get('status')} ({condition.get('reason', 'N/A')})\n" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get deployment: HTTP {result['status_code']}")] + + elif name == "create_deployment": + dep_name = arguments.get("name") + image = arguments.get("image") + if not env_id or not dep_name or not image: + return [types.TextContent(type="text", text="Error: environment_id, name, and image are required")] + + # Build deployment spec + deployment_data = { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "name": dep_name, + "namespace": namespace + }, + "spec": { + "replicas": arguments.get("replicas", 1), + "selector": { + "matchLabels": { + "app": dep_name + } + }, + "template": { + "metadata": { + "labels": { + "app": dep_name + } + }, + "spec": { + "containers": [{ + "name": dep_name, + "image": image + }] + } + } + } + } + + # Add custom labels if provided + if arguments.get("labels"): + deployment_data["metadata"]["labels"] = arguments["labels"] + deployment_data["spec"]["template"]["metadata"]["labels"].update(arguments["labels"]) + + # Add port if provided + if arguments.get("port"): + deployment_data["spec"]["template"]["spec"]["containers"][0]["ports"] = [{ + "containerPort": arguments["port"] + }] + + # Add environment variables if provided + if arguments.get("env"): + deployment_data["spec"]["template"]["spec"]["containers"][0]["env"] = arguments["env"] + + result = await make_request("POST", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments", + json_data=deployment_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"āœ“ Deployment '{dep_name}' created successfully")] + else: + error_msg = f"Failed to create deployment: HTTP {result['status_code']}" + if result.get("data") and result["data"].get("message"): + error_msg += f"\n{result['data']['message']}" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "scale_deployment": + deployment_name = arguments.get("deployment_name") + replicas = arguments.get("replicas") + if not env_id or not deployment_name or replicas is None: + return [types.TextContent(type="text", text="Error: environment_id, deployment_name, and replicas are required")] + + # Get current deployment first + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments/{deployment_name}") + if result["status_code"] != 200: + return [types.TextContent(type="text", text=f"Failed to get deployment: HTTP {result['status_code']}")] + + deployment = result["data"] + deployment["spec"]["replicas"] = replicas + + # Update deployment + result = await make_request("PUT", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments/{deployment_name}", + json_data=deployment) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"āœ“ Deployment '{deployment_name}' scaled to {replicas} replicas")] + else: + return [types.TextContent(type="text", text=f"Failed to scale deployment: HTTP {result['status_code']}")] + + elif name == "update_deployment_image": + deployment_name = arguments.get("deployment_name") + container_name = arguments.get("container_name") + new_image = arguments.get("new_image") + if not env_id or not deployment_name or not container_name or not new_image: + return [types.TextContent(type="text", text="Error: environment_id, deployment_name, container_name, and new_image are required")] + + # Get current deployment + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments/{deployment_name}") + if result["status_code"] != 200: + return [types.TextContent(type="text", text=f"Failed to get deployment: HTTP {result['status_code']}")] + + deployment = result["data"] + containers = deployment["spec"]["template"]["spec"]["containers"] + + # Find and update container image + updated = False + for container in containers: + if container["name"] == container_name: + container["image"] = new_image + updated = True + break + + if not updated: + return [types.TextContent(type="text", text=f"Container '{container_name}' not found in deployment")] + + # Update deployment + result = await make_request("PUT", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments/{deployment_name}", + json_data=deployment) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"āœ“ Updated container '{container_name}' to image '{new_image}'")] + else: + return [types.TextContent(type="text", text=f"Failed to update deployment: HTTP {result['status_code']}")] + + elif name == "restart_deployment": + deployment_name = arguments.get("deployment_name") + if not env_id or not deployment_name: + return [types.TextContent(type="text", text="Error: environment_id and deployment_name are required")] + + # Get current deployment + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments/{deployment_name}") + if result["status_code"] != 200: + return [types.TextContent(type="text", text=f"Failed to get deployment: HTTP {result['status_code']}")] + + deployment = result["data"] + + # Add/update restart annotation + import datetime + if "annotations" not in deployment["spec"]["template"]["metadata"]: + deployment["spec"]["template"]["metadata"]["annotations"] = {} + deployment["spec"]["template"]["metadata"]["annotations"]["kubectl.kubernetes.io/restartedAt"] = datetime.datetime.utcnow().isoformat() + + # Update deployment + result = await make_request("PUT", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments/{deployment_name}", + json_data=deployment) + + if result["status_code"] == 200: + return [types.TextContent(type="text", text=f"āœ“ Deployment '{deployment_name}' restart initiated")] + else: + return [types.TextContent(type="text", text=f"Failed to restart deployment: HTTP {result['status_code']}")] + + elif name == "delete_deployment": + deployment_name = arguments.get("deployment_name") + if not env_id or not deployment_name: + return [types.TextContent(type="text", text="Error: environment_id and deployment_name are required")] + + result = await make_request("DELETE", f"/endpoints/{env_id}/kubernetes/apis/apps/v1/namespaces/{namespace}/deployments/{deployment_name}") + + if result["status_code"] in [200, 202]: + return [types.TextContent(type="text", text=f"āœ“ Deployment '{deployment_name}' deleted successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to delete deployment: HTTP {result['status_code']}")] + + # Service Management + elif name == "list_services": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/services") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} service(s) in namespace '{namespace}':\n" + + for service in items: + metadata = service.get("metadata", {}) + spec = service.get("spec", {}) + + svc_name = metadata.get("name") + svc_type = spec.get("type", "ClusterIP") + cluster_ip = spec.get("clusterIP", "None") + + output += f"\n- {svc_name}" + output += f"\n Type: {svc_type}" + output += f"\n Cluster IP: {cluster_ip}" + + # Ports + ports = spec.get("ports", []) + if ports: + port_info = [] + for port in ports: + port_str = f"{port.get('port')}" + if port.get("targetPort") and str(port.get("targetPort")) != str(port.get("port")): + port_str += f"ā†’{port.get('targetPort')}" + if port.get("nodePort"): + port_str += f" (NodePort: {port.get('nodePort')})" + port_info.append(port_str) + output += f"\n Ports: {', '.join(port_info)}" + + # External IPs or LoadBalancer IP + if svc_type == "LoadBalancer": + status = service.get("status", {}) + lb = status.get("loadBalancer", {}) + ingresses = lb.get("ingress", []) + if ingresses: + ips = [ing.get("ip") or ing.get("hostname") for ing in ingresses] + output += f"\n External: {', '.join(ips)}" + elif spec.get("externalIPs"): + output += f"\n External IPs: {', '.join(spec['externalIPs'])}" + + # Selector + selector = spec.get("selector", {}) + if selector: + selector_str = ", ".join([f"{k}={v}" for k, v in selector.items()]) + output += f"\n Selector: {selector_str}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list services: HTTP {result['status_code']}")] + + elif name == "create_service": + svc_name = arguments.get("name") + selector = arguments.get("selector") + ports = arguments.get("ports") + if not env_id or not svc_name or not selector or not ports: + return [types.TextContent(type="text", text="Error: environment_id, name, selector, and ports are required")] + + service_data = { + "apiVersion": "v1", + "kind": "Service", + "metadata": { + "name": svc_name, + "namespace": namespace + }, + "spec": { + "selector": selector, + "ports": ports, + "type": arguments.get("type", "ClusterIP") + } + } + + result = await make_request("POST", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/services", + json_data=service_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"āœ“ Service '{svc_name}' created successfully")] + else: + error_msg = f"Failed to create service: HTTP {result['status_code']}" + if result.get("data") and result["data"].get("message"): + error_msg += f"\n{result['data']['message']}" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "delete_service": + service_name = arguments.get("service_name") + if not env_id or not service_name: + return [types.TextContent(type="text", text="Error: environment_id and service_name are required")] + + result = await make_request("DELETE", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/services/{service_name}") + + if result["status_code"] in [200, 202]: + return [types.TextContent(type="text", text=f"āœ“ Service '{service_name}' deleted successfully")] + else: + return [types.TextContent(type="text", text=f"Failed to delete service: HTTP {result['status_code']}")] + + # Ingress Management + elif name == "list_ingresses": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/apis/networking.k8s.io/v1/namespaces/{namespace}/ingresses") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} ingress(es) in namespace '{namespace}':\n" + + for ingress in items: + metadata = ingress.get("metadata", {}) + spec = ingress.get("spec", {}) + status = ingress.get("status", {}) + + ing_name = metadata.get("name") + output += f"\n- {ing_name}" + + # Ingress class + ing_class = spec.get("ingressClassName", "default") + output += f"\n Class: {ing_class}" + + # Rules + rules = spec.get("rules", []) + for rule in rules: + host = rule.get("host", "*") + output += f"\n Host: {host}" + + http = rule.get("http", {}) + paths = http.get("paths", []) + for path in paths: + path_val = path.get("path", "/") + backend = path.get("backend", {}) + service = backend.get("service", {}) + svc_name = service.get("name", "unknown") + svc_port = service.get("port", {}).get("number", "unknown") + output += f"\n {path_val} ā†’ {svc_name}:{svc_port}" + + # Load balancer info + lb = status.get("loadBalancer", {}) + ingresses = lb.get("ingress", []) + if ingresses: + ips = [ing.get("ip") or ing.get("hostname") for ing in ingresses] + output += f"\n Address: {', '.join(ips)}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list ingresses: HTTP {result['status_code']}")] + + elif name == "create_ingress": + ing_name = arguments.get("name") + rules = arguments.get("rules") + if not env_id or not ing_name or not rules: + return [types.TextContent(type="text", text="Error: environment_id, name, and rules are required")] + + # Build ingress spec + ingress_data = { + "apiVersion": "networking.k8s.io/v1", + "kind": "Ingress", + "metadata": { + "name": ing_name, + "namespace": namespace + }, + "spec": { + "rules": [] + } + } + + # Process rules + for rule in rules: + rule_spec = {} + if rule.get("host"): + rule_spec["host"] = rule["host"] + + paths = [] + for path in rule.get("paths", []): + path_spec = { + "path": path.get("path", "/"), + "pathType": "Prefix", + "backend": { + "service": { + "name": path["service"], + "port": { + "number": path["port"] + } + } + } + } + paths.append(path_spec) + + rule_spec["http"] = {"paths": paths} + ingress_data["spec"]["rules"].append(rule_spec) + + result = await make_request("POST", f"/endpoints/{env_id}/kubernetes/apis/networking.k8s.io/v1/namespaces/{namespace}/ingresses", + json_data=ingress_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"āœ“ Ingress '{ing_name}' created successfully")] + else: + error_msg = f"Failed to create ingress: HTTP {result['status_code']}" + if result.get("data") and result["data"].get("message"): + error_msg += f"\n{result['data']['message']}" + return [types.TextContent(type="text", text=error_msg)] + + # ConfigMap and Secret Management + elif name == "list_configmaps": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/configmaps") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} ConfigMap(s) in namespace '{namespace}':\n" + + for cm in items: + metadata = cm.get("metadata", {}) + data = cm.get("data", {}) + + output += f"\n- {metadata.get('name')}" + output += f"\n Keys: {len(data)}" + if data: + output += f" ({', '.join(list(data.keys())[:5])}" + if len(data) > 5: + output += f", ... +{len(data) - 5} more" + output += ")" + output += f"\n Created: {metadata.get('creationTimestamp')}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list ConfigMaps: HTTP {result['status_code']}")] + + elif name == "create_configmap": + cm_name = arguments.get("name") + data = arguments.get("data") + if not env_id or not cm_name or not data: + return [types.TextContent(type="text", text="Error: environment_id, name, and data are required")] + + configmap_data = { + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": { + "name": cm_name, + "namespace": namespace + }, + "data": data + } + + result = await make_request("POST", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/configmaps", + json_data=configmap_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"āœ“ ConfigMap '{cm_name}' created successfully")] + else: + error_msg = f"Failed to create ConfigMap: HTTP {result['status_code']}" + if result.get("data") and result["data"].get("message"): + error_msg += f"\n{result['data']['message']}" + return [types.TextContent(type="text", text=error_msg)] + + elif name == "list_secrets": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/secrets") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} Secret(s) in namespace '{namespace}':\n" + + for secret in items: + metadata = secret.get("metadata", {}) + data = secret.get("data", {}) + secret_type = secret.get("type", "Opaque") + + output += f"\n- {metadata.get('name')}" + output += f"\n Type: {secret_type}" + output += f"\n Keys: {len(data)}" + if data: + output += f" ({', '.join(list(data.keys())[:3])}" + if len(data) > 3: + output += f", ... +{len(data) - 3} more" + output += ")" + output += f"\n Created: {metadata.get('creationTimestamp')}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list Secrets: HTTP {result['status_code']}")] + + elif name == "create_secret": + secret_name = arguments.get("name") + data = arguments.get("data") + if not env_id or not secret_name or not data: + return [types.TextContent(type="text", text="Error: environment_id, name, and data are required")] + + # Base64 encode the data values + encoded_data = {} + for key, value in data.items(): + encoded_data[key] = base64.b64encode(value.encode()).decode() + + secret_data = { + "apiVersion": "v1", + "kind": "Secret", + "metadata": { + "name": secret_name, + "namespace": namespace + }, + "type": arguments.get("type", "Opaque"), + "data": encoded_data + } + + result = await make_request("POST", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/secrets", + json_data=secret_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"āœ“ Secret '{secret_name}' created successfully")] + else: + error_msg = f"Failed to create Secret: HTTP {result['status_code']}" + if result.get("data") and result["data"].get("message"): + error_msg += f"\n{result['data']['message']}" + return [types.TextContent(type="text", text=error_msg)] + + # Persistent Volume Management + elif name == "list_persistent_volumes": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/persistentvolumes") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} PersistentVolume(s):\n" + + for pv in items: + metadata = pv.get("metadata", {}) + spec = pv.get("spec", {}) + status = pv.get("status", {}) + + output += f"\n- {metadata.get('name')}" + output += f"\n Capacity: {spec.get('capacity', {}).get('storage', 'N/A')}" + output += f"\n Access Modes: {', '.join(spec.get('accessModes', []))}" + output += f"\n Storage Class: {spec.get('storageClassName', 'N/A')}" + output += f"\n Status: {status.get('phase', 'Unknown')}" + + # Claim reference + claim_ref = spec.get("claimRef") + if claim_ref: + output += f"\n Bound to: {claim_ref.get('namespace')}/{claim_ref.get('name')}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list PersistentVolumes: HTTP {result['status_code']}")] + + elif name == "list_persistent_volume_claims": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/persistentvolumeclaims") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} PersistentVolumeClaim(s) in namespace '{namespace}':\n" + + for pvc in items: + metadata = pvc.get("metadata", {}) + spec = pvc.get("spec", {}) + status = pvc.get("status", {}) + + output += f"\n- {metadata.get('name')}" + output += f"\n Status: {status.get('phase', 'Unknown')}" + output += f"\n Capacity: {status.get('capacity', {}).get('storage', spec.get('resources', {}).get('requests', {}).get('storage', 'N/A'))}" + output += f"\n Access Modes: {', '.join(spec.get('accessModes', []))}" + output += f"\n Storage Class: {spec.get('storageClassName', 'N/A')}" + output += f"\n Volume: {spec.get('volumeName', 'N/A')}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list PersistentVolumeClaims: HTTP {result['status_code']}")] + + elif name == "create_persistent_volume_claim": + pvc_name = arguments.get("name") + size = arguments.get("size") + if not env_id or not pvc_name or not size: + return [types.TextContent(type="text", text="Error: environment_id, name, and size are required")] + + pvc_data = { + "apiVersion": "v1", + "kind": "PersistentVolumeClaim", + "metadata": { + "name": pvc_name, + "namespace": namespace + }, + "spec": { + "accessModes": [arguments.get("access_mode", "ReadWriteOnce")], + "resources": { + "requests": { + "storage": size + } + } + } + } + + if arguments.get("storage_class"): + pvc_data["spec"]["storageClassName"] = arguments["storage_class"] + + result = await make_request("POST", f"/endpoints/{env_id}/kubernetes/api/v1/namespaces/{namespace}/persistentvolumeclaims", + json_data=pvc_data) + + if result["status_code"] in [200, 201]: + return [types.TextContent(type="text", text=f"āœ“ PersistentVolumeClaim '{pvc_name}' created successfully")] + else: + error_msg = f"Failed to create PVC: HTTP {result['status_code']}" + if result.get("data") and result["data"].get("message"): + error_msg += f"\n{result['data']['message']}" + return [types.TextContent(type="text", text=error_msg)] + + # Node Information + elif name == "list_nodes": + if not env_id: + return [types.TextContent(type="text", text="Error: environment_id is required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/nodes") + + if result["status_code"] == 200 and result["data"]: + items = result["data"].get("items", []) + output = f"Found {len(items)} node(s):\n" + + for node in items: + metadata = node.get("metadata", {}) + status = node.get("status", {}) + + node_name = metadata.get("name") + + # Node conditions + ready = "Unknown" + for condition in status.get("conditions", []): + if condition.get("type") == "Ready": + ready = "Ready" if condition.get("status") == "True" else "NotReady" + break + + output += f"\n- {node_name}" + output += f"\n Status: {ready}" + + # Node info + node_info = status.get("nodeInfo", {}) + output += f"\n Version: {node_info.get('kubeletVersion', 'N/A')}" + output += f"\n OS: {node_info.get('operatingSystem', 'N/A')} ({node_info.get('architecture', 'N/A')})" + + # Resources + capacity = status.get("capacity", {}) + allocatable = status.get("allocatable", {}) + if capacity: + output += f"\n CPU: {capacity.get('cpu', 'N/A')} cores" + output += f"\n Memory: {format_resource_quantity(capacity.get('memory', 'N/A'))}" + output += f"\n Pods: {allocatable.get('pods', 'N/A')} allocatable" + + # Taints + taints = node.get("spec", {}).get("taints", []) + if taints: + taint_strs = [f"{t.get('key')}={t.get('value')}:{t.get('effect')}" for t in taints] + output += f"\n Taints: {', '.join(taint_strs)}" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to list nodes: HTTP {result['status_code']}")] + + elif name == "get_node": + node_name = arguments.get("node_name") + if not env_id or not node_name: + return [types.TextContent(type="text", text="Error: environment_id and node_name are required")] + + result = await make_request("GET", f"/endpoints/{env_id}/kubernetes/api/v1/nodes/{node_name}") + + if result["status_code"] == 200 and result["data"]: + node = result["data"] + metadata = node.get("metadata", {}) + spec = node.get("spec", {}) + status = node.get("status", {}) + + output = f"Node Details:\n" + output += f"- Name: {metadata.get('name')}\n" + output += f"- Created: {metadata.get('creationTimestamp')}\n" + + # Status + for condition in status.get("conditions", []): + output += f"- {condition.get('type')}: {condition.get('status')} ({condition.get('reason', 'N/A')})\n" + + # Node info + node_info = status.get("nodeInfo", {}) + output += f"\nSystem Info:\n" + output += f"- Kubernetes: {node_info.get('kubeletVersion', 'N/A')}\n" + output += f"- Container Runtime: {node_info.get('containerRuntimeVersion', 'N/A')}\n" + output += f"- OS: {node_info.get('osImage', 'N/A')}\n" + output += f"- Kernel: {node_info.get('kernelVersion', 'N/A')}\n" + output += f"- Architecture: {node_info.get('architecture', 'N/A')}\n" + + # Capacity and Allocatable + capacity = status.get("capacity", {}) + allocatable = status.get("allocatable", {}) + output += f"\nResources:\n" + output += f"- CPU: {capacity.get('cpu', 'N/A')} capacity, {allocatable.get('cpu', 'N/A')} allocatable\n" + output += f"- Memory: {format_resource_quantity(capacity.get('memory', 'N/A'))} capacity, " + output += f"{format_resource_quantity(allocatable.get('memory', 'N/A'))} allocatable\n" + output += f"- Pods: {capacity.get('pods', 'N/A')} capacity, {allocatable.get('pods', 'N/A')} allocatable\n" + + # Addresses + addresses = status.get("addresses", []) + if addresses: + output += "\nAddresses:\n" + for addr in addresses: + output += f"- {addr.get('type')}: {addr.get('address')}\n" + + # Images + images = status.get("images", []) + if images: + output += f"\nImages: {len(images)} cached\n" + + return [types.TextContent(type="text", text=output)] + else: + return [types.TextContent(type="text", text=f"Failed to get node: HTTP {result['status_code']}")] + + else: + return [types.TextContent(type="text", text=f"Unknown tool: {name}")] + + except httpx.TimeoutException: + return [types.TextContent(type="text", text="Error: Request timed out. The operation may take longer than expected.")] + except httpx.ConnectError: + return [types.TextContent(type="text", text="Error: Could not connect to Portainer server. Please check the URL and network connection.")] + except Exception as e: + import traceback + error_details = f"Error: {str(e)}\nType: {type(e).__name__}" + return [types.TextContent(type="text", text=error_details)] + + +async def run(): + """Run the MCP server.""" + # Use stdio transport + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="portainer-kubernetes", + server_version="1.0.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions( + prompts_changed=False, + resources_changed=False, + tools_changed=False, + ), + experimental_capabilities={}, + ), + ), + ) + + +def main(): + """Main entry point.""" + asyncio.run(run()) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/portainer_stacks_server.py b/portainer_stacks_server.py new file mode 100755 index 0000000..9089955 --- /dev/null +++ b/portainer_stacks_server.py @@ -0,0 +1,840 @@ +#!/usr/bin/env python3 +""" +Portainer Stacks MCP Server + +Provides stack deployment and management functionality through Portainer's API. +Supports Docker Compose stacks and Kubernetes manifests. +""" + +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_stack_status(stack: dict) -> str: + """Format stack status with emoji.""" + status = stack.get("Status", 0) + if status == 1: + return "āœ… Active" + elif status == 2: + return "āš ļø Inactive" + else: + return "ā“ Unknown" + +def format_stack_type(stack_type: int) -> str: + """Format stack type.""" + if stack_type == 1: + return "Swarm" + elif stack_type == 2: + return "Compose" + elif stack_type == 3: + return "Kubernetes" + else: + return "Unknown" + +# Create server instance +server = Server("portainer-stacks") + +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List all available tools.""" + return [ + types.Tool( + name="list_stacks", + description="List all stacks across environments", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "string", + "description": "Filter by environment ID (optional)" + } + } + } + ), + types.Tool( + name="get_stack", + description="Get detailed information about a specific stack", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + } + }, + "required": ["stack_id"] + } + ), + types.Tool( + name="get_stack_file", + description="Get the stack file content (Docker Compose or Kubernetes manifest)", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + } + }, + "required": ["stack_id"] + } + ), + types.Tool( + name="create_compose_stack_from_file", + description="Create a new Docker Compose stack from file content", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "string", + "description": "Target environment ID" + }, + "name": { + "type": "string", + "description": "Stack name" + }, + "compose_file": { + "type": "string", + "description": "Docker Compose file content (YAML)" + }, + "env_vars": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "value": {"type": "string"} + } + }, + "description": "Environment variables" + } + }, + "required": ["environment_id", "name", "compose_file"] + } + ), + types.Tool( + name="create_compose_stack_from_git", + description="Create a new Docker Compose stack from Git repository", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "string", + "description": "Target environment ID" + }, + "name": { + "type": "string", + "description": "Stack name" + }, + "repository_url": { + "type": "string", + "description": "Git repository URL" + }, + "repository_ref": { + "type": "string", + "description": "Git reference (branch/tag)", + "default": "main" + }, + "compose_path": { + "type": "string", + "description": "Path to compose file in repository", + "default": "docker-compose.yml" + }, + "repository_auth": { + "type": "boolean", + "description": "Use repository authentication", + "default": False + }, + "repository_username": { + "type": "string", + "description": "Git username (if auth required)" + }, + "repository_password": { + "type": "string", + "description": "Git password/token (if auth required)" + }, + "env_vars": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "value": {"type": "string"} + } + }, + "description": "Environment variables" + } + }, + "required": ["environment_id", "name", "repository_url"] + } + ), + types.Tool( + name="create_kubernetes_stack", + description="Create a new Kubernetes stack from manifest", + inputSchema={ + "type": "object", + "properties": { + "environment_id": { + "type": "string", + "description": "Target environment ID" + }, + "name": { + "type": "string", + "description": "Stack name" + }, + "namespace": { + "type": "string", + "description": "Kubernetes namespace", + "default": "default" + }, + "manifest": { + "type": "string", + "description": "Kubernetes manifest content (YAML)" + } + }, + "required": ["environment_id", "name", "manifest"] + } + ), + types.Tool( + name="update_stack", + description="Update an existing stack", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + }, + "compose_file": { + "type": "string", + "description": "Updated compose file or manifest (required for file-based stacks)" + }, + "env_vars": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "value": {"type": "string"} + } + }, + "description": "Updated environment variables" + }, + "pull_image": { + "type": "boolean", + "description": "Pull latest images before updating", + "default": True + } + }, + "required": ["stack_id"] + } + ), + types.Tool( + name="update_git_stack", + description="Update a Git-based stack (pull latest changes)", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + }, + "pull_image": { + "type": "boolean", + "description": "Pull latest images after updating", + "default": True + } + }, + "required": ["stack_id"] + } + ), + types.Tool( + name="start_stack", + description="Start a stopped stack", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + } + }, + "required": ["stack_id"] + } + ), + types.Tool( + name="stop_stack", + description="Stop a running stack", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + } + }, + "required": ["stack_id"] + } + ), + types.Tool( + name="delete_stack", + description="Delete a stack and optionally its volumes", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + }, + "delete_volumes": { + "type": "boolean", + "description": "Also delete associated volumes", + "default": False + } + }, + "required": ["stack_id"] + } + ), + types.Tool( + name="migrate_stack", + description="Migrate a stack to another environment", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + }, + "target_environment_id": { + "type": "string", + "description": "Target environment ID" + }, + "new_name": { + "type": "string", + "description": "New stack name (optional)" + } + }, + "required": ["stack_id", "target_environment_id"] + } + ), + types.Tool( + name="get_stack_logs", + description="Get logs from all containers in a stack", + inputSchema={ + "type": "object", + "properties": { + "stack_id": { + "type": "string", + "description": "Stack ID" + }, + "tail": { + "type": "integer", + "description": "Number of lines to show from the end", + "default": 100 + }, + "timestamps": { + "type": "boolean", + "description": "Show timestamps", + "default": True + } + }, + "required": ["stack_id"] + } + ) + ] + +@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 stacks + if name == "list_stacks": + endpoint = "/api/stacks" + params = {} + + result = await make_request("GET", endpoint, params=params) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + # Filter by environment if specified + stacks = result + if arguments.get("environment_id"): + env_id = int(arguments["environment_id"]) + stacks = [s for s in stacks if s.get("EndpointId") == env_id] + + if not stacks: + return [types.TextContent(type="text", text="No stacks found")] + + output = "šŸ“š Stacks:\n\n" + + # Group by environment + env_groups = {} + for stack in stacks: + env_id = stack.get("EndpointId", "Unknown") + if env_id not in env_groups: + env_groups[env_id] = [] + env_groups[env_id].append(stack) + + for env_id, env_stacks in env_groups.items(): + output += f"Environment {env_id}:\n" + for stack in env_stacks: + status = format_stack_status(stack) + stack_type = format_stack_type(stack.get("Type", 0)) + output += f" ā€¢ {stack['Name']} (ID: {stack['Id']})\n" + output += f" Type: {stack_type} | Status: {status}\n" + if stack.get("GitConfig"): + output += f" Git: {stack['GitConfig']['URL']} ({stack['GitConfig']['ReferenceName']})\n" + output += "\n" + + return [types.TextContent(type="text", text=output)] + + # Get stack details + elif name == "get_stack": + 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']}")] + + output = f"šŸ“š Stack: {result['Name']}\n\n" + output += f"ID: {result['Id']}\n" + output += f"Type: {format_stack_type(result.get('Type', 0))}\n" + output += f"Status: {format_stack_status(result)}\n" + output += f"Environment ID: {result.get('EndpointId', 'Unknown')}\n" + output += f"Created by: {result.get('CreatedBy', 'Unknown')}\n" + + if result.get("GitConfig"): + git = result["GitConfig"] + output += f"\nšŸ”— Git Configuration:\n" + output += f" Repository: {git['URL']}\n" + output += f" Reference: {git['ReferenceName']}\n" + output += f" Path: {git.get('ComposeFilePathInRepository', 'N/A')}\n" + + if result.get("Env"): + output += f"\nšŸ”§ Environment Variables:\n" + for env in result["Env"]: + output += f" {env['name']} = {env['value']}\n" + + if result.get("ResourceControl"): + rc = result["ResourceControl"] + output += f"\nšŸ”’ Access Control:\n" + output += f" Public: {'Yes' if rc.get('Public') else 'No'}\n" + if rc.get("Users"): + output += f" Users: {len(rc['Users'])} users\n" + if rc.get("Teams"): + output += f" Teams: {len(rc['Teams'])} teams\n" + + return [types.TextContent(type="text", text=output)] + + # Get stack file + elif name == "get_stack_file": + stack_id = arguments["stack_id"] + + result = await make_request("GET", f"/api/stacks/{stack_id}/file") + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + content = result.get("StackFileContent", "") + if not content: + return [types.TextContent(type="text", text="Stack file is empty")] + + output = f"šŸ“„ Stack File Content:\n\n```yaml\n{content}\n```" + + return [types.TextContent(type="text", text=output)] + + # Create compose stack from file + elif name == "create_compose_stack_from_file": + env_id = arguments["environment_id"] + + # Build request data + data = { + "Name": arguments["name"], + "StackFileContent": arguments["compose_file"], + "EndpointId": int(env_id) + } + + # Add environment variables if provided + if arguments.get("env_vars"): + data["Env"] = arguments["env_vars"] + + result = await make_request("POST", "/api/stacks", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"āœ… Stack created successfully!\n\n" + output += f"Name: {result['Name']}\n" + output += f"ID: {result['Id']}\n" + output += f"Type: Compose\n" + output += f"Environment: {result.get('EndpointId', 'Unknown')}\n" + + return [types.TextContent(type="text", text=output)] + + # Create compose stack from Git + elif name == "create_compose_stack_from_git": + env_id = arguments["environment_id"] + + # Build request data + data = { + "Name": arguments["name"], + "EndpointId": int(env_id), + "GitConfig": { + "URL": arguments["repository_url"], + "ReferenceName": arguments.get("repository_ref", "main"), + "ComposeFilePathInRepository": arguments.get("compose_path", "docker-compose.yml") + } + } + + # Add authentication if provided + if arguments.get("repository_auth") and arguments.get("repository_username"): + data["GitConfig"]["Authentication"] = { + "Username": arguments["repository_username"], + "Password": arguments.get("repository_password", "") + } + + # Add environment variables if provided + if arguments.get("env_vars"): + data["Env"] = arguments["env_vars"] + + result = await make_request("POST", "/api/stacks", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"āœ… Git-based stack created successfully!\n\n" + output += f"Name: {result['Name']}\n" + output += f"ID: {result['Id']}\n" + output += f"Repository: {arguments['repository_url']}\n" + output += f"Branch/Tag: {arguments.get('repository_ref', 'main')}\n" + + return [types.TextContent(type="text", text=output)] + + # Create Kubernetes stack + elif name == "create_kubernetes_stack": + env_id = arguments["environment_id"] + + # Build request data + data = { + "Name": arguments["name"], + "StackFileContent": arguments["manifest"], + "EndpointId": int(env_id), + "Type": 3, # Kubernetes type + "Namespace": arguments.get("namespace", "default") + } + + result = await make_request("POST", "/api/stacks", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"āœ… Kubernetes stack created successfully!\n\n" + output += f"Name: {result['Name']}\n" + output += f"ID: {result['Id']}\n" + output += f"Namespace: {arguments.get('namespace', 'default')}\n" + output += f"Environment: {result.get('EndpointId', 'Unknown')}\n" + + return [types.TextContent(type="text", text=output)] + + # Update stack + elif name == "update_stack": + stack_id = arguments["stack_id"] + + # Get current stack info first + 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']}")] + + # Build update data + data = {} + + if arguments.get("compose_file"): + data["StackFileContent"] = arguments["compose_file"] + + if arguments.get("env_vars"): + data["Env"] = arguments["env_vars"] + + data["Prune"] = arguments.get("prune", False) + data["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=data, params=params) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text=f"āœ… Stack '{stack_info['Name']}' updated successfully!")] + + # Update Git stack + elif name == "update_git_stack": + 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 is not a Git-based stack")] + + 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"āœ… Git stack '{stack_info['Name']}' updated from repository!")] + + # Start stack + elif name == "start_stack": + 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']}")] + + endpoint = f"/api/stacks/{stack_id}/start" + params = {"endpointId": stack_info["EndpointId"]} + + result = await make_request("POST", endpoint, params=params) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text=f"āœ… Stack '{stack_info['Name']}' started successfully!")] + + # Stop stack + elif name == "stop_stack": + 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']}")] + + endpoint = f"/api/stacks/{stack_id}/stop" + params = {"endpointId": stack_info["EndpointId"]} + + result = await make_request("POST", endpoint, params=params) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + return [types.TextContent(type="text", text=f"ā¹ļø Stack '{stack_info['Name']}' stopped successfully!")] + + # Delete stack + elif name == "delete_stack": + 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']}")] + + endpoint = f"/api/stacks/{stack_id}" + params = { + "endpointId": stack_info["EndpointId"], + "external": "false" + } + + if arguments.get("delete_volumes"): + # For compose stacks, this deletes volumes + data = {"removeVolumes": True} + result = await make_request("DELETE", endpoint, params=params, json_data=data) + else: + result = await make_request("DELETE", endpoint, params=params) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"šŸ—‘ļø Stack '{stack_info['Name']}' deleted successfully!" + if arguments.get("delete_volumes"): + output += " (including volumes)" + + return [types.TextContent(type="text", text=output)] + + # Migrate stack + elif name == "migrate_stack": + stack_id = arguments["stack_id"] + target_env = arguments["target_environment_id"] + + # Get current stack info and file + 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']}")] + + stack_file = await make_request("GET", f"/api/stacks/{stack_id}/file") + if "error" in stack_file: + return [types.TextContent(type="text", text=f"Error: {stack_file['error']}")] + + # Create new stack in target environment + new_name = arguments.get("new_name", f"{stack_info['Name']}-migrated") + + data = { + "Name": new_name, + "StackFileContent": stack_file.get("StackFileContent", ""), + "EndpointId": int(target_env), + "Type": stack_info.get("Type", 2) + } + + # Copy environment variables if any + if stack_info.get("Env"): + data["Env"] = stack_info["Env"] + + # For Kubernetes stacks, copy namespace + if stack_info.get("Type") == 3 and stack_info.get("Namespace"): + data["Namespace"] = stack_info["Namespace"] + + result = await make_request("POST", "/api/stacks", json_data=data) + + if "error" in result: + return [types.TextContent(type="text", text=f"Error: {result['error']}")] + + output = f"āœ… Stack migrated successfully!\n\n" + output += f"Original: {stack_info['Name']} (Environment {stack_info['EndpointId']})\n" + output += f"New: {new_name} (Environment {target_env})\n" + output += f"New Stack ID: {result['Id']}\n" + output += "\nNote: Original stack was not deleted." + + return [types.TextContent(type="text", text=output)] + + # Get stack logs + elif name == "get_stack_logs": + stack_id = arguments["stack_id"] + + # This is a simplified version - actual implementation would need to: + # 1. Get stack details + # 2. List all containers in the stack + # 3. Aggregate logs from all containers + + return [types.TextContent( + type="text", + text="Note: Stack logs aggregation requires listing all containers in the stack. Use docker logs on individual containers for now." + )] + + 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-stacks", + 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