jpmschweitzerandClaude Opus 4.5
dbbb92d292
- Remove _list_containers_via_socket method - Remove _inspect_container_via_socket method - Simplify list_containers and inspect_container to use Portainer API only - Raise RuntimeError when no Portainer endpoints available BREAKING: Portainer API configuration is now required for all container and stack operations. Set PORTAINER_URL and PORTAINER_API_KEY env vars. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
507 lines
16 KiB
Python
507 lines
16 KiB
Python
"""
|
|
Portainer API Client
|
|
|
|
Provides interface to Portainer REST API for stack and container management.
|
|
Includes fallback to Docker socket for containers not managed by Portainer.
|
|
"""
|
|
import httpx
|
|
import json
|
|
from typing import Optional, Dict, List, Any
|
|
from src.logging_config import get_logger
|
|
from src.config import get_settings
|
|
|
|
logger = get_logger(__name__)
|
|
settings = get_settings()
|
|
|
|
|
|
class PortainerClient:
|
|
"""
|
|
HTTP client for Portainer API
|
|
|
|
Uses access token authentication (X-API-Key header)
|
|
for long-lived API access without session management.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: Optional[str] = None,
|
|
api_key: Optional[str] = None,
|
|
timeout: int = 30
|
|
):
|
|
"""
|
|
Initialize Portainer client
|
|
|
|
Args:
|
|
base_url: Portainer base URL (default from settings)
|
|
api_key: Portainer API access token (default from settings)
|
|
timeout: Request timeout in seconds
|
|
"""
|
|
self.base_url = (base_url or settings.portainer_url).rstrip("/")
|
|
self.api_key = api_key or settings.portainer_api_key
|
|
self.timeout = timeout
|
|
|
|
if not self.api_key:
|
|
logger.warning("Portainer API key not configured")
|
|
|
|
def _get_headers(self) -> Dict[str, str]:
|
|
"""Get request headers with authentication"""
|
|
return {
|
|
"X-API-Key": self.api_key,
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
async def health_check(self) -> bool:
|
|
"""
|
|
Check if Portainer API is accessible
|
|
|
|
Returns:
|
|
True if accessible, False otherwise
|
|
"""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(f"{self.base_url}/api/status")
|
|
return response.status_code == 200
|
|
except Exception as e:
|
|
logger.error(f"Portainer health check failed: {e}")
|
|
return False
|
|
|
|
async def get_endpoints(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
List all Portainer endpoints (Docker environments)
|
|
|
|
Returns:
|
|
List of endpoint configurations
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/endpoints",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_stacks(self, endpoint_id: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
List all stacks
|
|
|
|
Args:
|
|
endpoint_id: Filter by specific endpoint (optional)
|
|
|
|
Returns:
|
|
List of stack configurations
|
|
"""
|
|
params = {}
|
|
if endpoint_id:
|
|
params["endpointId"] = endpoint_id
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/stacks",
|
|
headers=self._get_headers(),
|
|
params=params
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_stack(self, stack_id: int) -> Dict[str, Any]:
|
|
"""
|
|
Get details of a specific stack
|
|
|
|
Args:
|
|
stack_id: Stack identifier
|
|
|
|
Returns:
|
|
Stack configuration details
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/stacks/{stack_id}",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def create_stack(
|
|
self,
|
|
name: str,
|
|
stack_file_content: str,
|
|
endpoint_id: int
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a new stack from compose file content
|
|
|
|
Args:
|
|
name: Stack name
|
|
stack_file_content: Docker Compose YAML content
|
|
endpoint_id: Portainer endpoint to deploy to
|
|
|
|
Returns:
|
|
Created stack details
|
|
"""
|
|
payload = {
|
|
"name": name,
|
|
"stackFileContent": stack_file_content
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/api/stacks/create/standalone/string",
|
|
headers=self._get_headers(),
|
|
params={"endpointId": endpoint_id},
|
|
json=payload
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def update_stack(
|
|
self,
|
|
stack_id: int,
|
|
stack_file_content: str,
|
|
endpoint_id: int,
|
|
prune: bool = False,
|
|
pull_image: bool = False
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Update an existing stack
|
|
|
|
Args:
|
|
stack_id: Stack identifier
|
|
stack_file_content: New Docker Compose YAML content
|
|
endpoint_id: Portainer endpoint
|
|
prune: Remove services no longer defined
|
|
pull_image: Pull latest images before deployment
|
|
|
|
Returns:
|
|
Updated stack details
|
|
"""
|
|
payload = {
|
|
"stackFileContent": stack_file_content,
|
|
"prune": prune,
|
|
"pullImage": pull_image
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.put(
|
|
f"{self.base_url}/api/stacks/{stack_id}",
|
|
headers=self._get_headers(),
|
|
params={"endpointId": endpoint_id},
|
|
json=payload
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def delete_stack(self, stack_id: int, endpoint_id: int) -> bool:
|
|
"""
|
|
Delete a stack
|
|
|
|
Args:
|
|
stack_id: Stack identifier
|
|
endpoint_id: Portainer endpoint
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.delete(
|
|
f"{self.base_url}/api/stacks/{stack_id}",
|
|
headers=self._get_headers(),
|
|
params={"endpointId": endpoint_id}
|
|
)
|
|
response.raise_for_status()
|
|
return True
|
|
|
|
async def get_stack_file(self, stack_id: int) -> str:
|
|
"""
|
|
Get the compose file content for a stack
|
|
|
|
Args:
|
|
stack_id: Stack identifier
|
|
|
|
Returns:
|
|
Docker Compose YAML content as string
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/stacks/{stack_id}/file",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data.get("StackFileContent", "")
|
|
|
|
async def redeploy_stack(
|
|
self,
|
|
stack_id: int,
|
|
endpoint_id: int,
|
|
pull_image: bool = False
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Redeploy a stack with its current configuration
|
|
|
|
Args:
|
|
stack_id: Stack identifier
|
|
endpoint_id: Portainer endpoint
|
|
pull_image: Pull latest images before deployment
|
|
|
|
Returns:
|
|
Updated stack details
|
|
"""
|
|
# Get current stack file content
|
|
stack_content = await self.get_stack_file(stack_id)
|
|
|
|
# Get current stack to preserve env vars
|
|
stack = await self.get_stack(stack_id)
|
|
env_vars = stack.get("Env", [])
|
|
|
|
payload = {
|
|
"stackFileContent": stack_content,
|
|
"env": env_vars,
|
|
"prune": False,
|
|
"pullImage": pull_image
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.put(
|
|
f"{self.base_url}/api/stacks/{stack_id}",
|
|
headers=self._get_headers(),
|
|
params={"endpointId": endpoint_id},
|
|
json=payload
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def update_stack_env(
|
|
self,
|
|
stack_id: int,
|
|
endpoint_id: int,
|
|
env_vars: List[Dict[str, str]]
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Update stack environment variables
|
|
|
|
Args:
|
|
stack_id: Stack identifier
|
|
endpoint_id: Portainer endpoint
|
|
env_vars: List of {"name": "VAR_NAME", "value": "var_value"} dicts
|
|
|
|
Returns:
|
|
Updated stack details
|
|
"""
|
|
# Get current stack file content (required for update)
|
|
stack_content = await self.get_stack_file(stack_id)
|
|
|
|
payload = {
|
|
"stackFileContent": stack_content,
|
|
"env": env_vars,
|
|
"prune": False,
|
|
"pullImage": False
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.put(
|
|
f"{self.base_url}/api/stacks/{stack_id}",
|
|
headers=self._get_headers(),
|
|
params={"endpointId": endpoint_id},
|
|
json=payload
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def delete_container(
|
|
self,
|
|
endpoint_id: int,
|
|
container_id: str,
|
|
force: bool = False
|
|
) -> bool:
|
|
"""
|
|
Delete a container
|
|
|
|
Args:
|
|
endpoint_id: Portainer endpoint identifier
|
|
container_id: Container ID or name
|
|
force: Force remove running container
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
params = {"force": "true" if force else "false"}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.delete(
|
|
f"{self.base_url}/api/endpoints/{endpoint_id}/docker/containers/{container_id}",
|
|
headers=self._get_headers(),
|
|
params=params
|
|
)
|
|
response.raise_for_status()
|
|
logger.info(f"Deleted container {container_id}")
|
|
return True
|
|
|
|
async def restart_container(self, endpoint_id: int, container_id: str) -> bool:
|
|
"""
|
|
Restart a container
|
|
|
|
Args:
|
|
endpoint_id: Portainer endpoint identifier
|
|
container_id: Container ID or name
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/api/endpoints/{endpoint_id}/docker/containers/{container_id}/restart",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
logger.info(f"Restarted container {container_id}")
|
|
return True
|
|
|
|
async def get_containers(self, endpoint_id: int, all_containers: bool = True) -> List[Dict[str, Any]]:
|
|
"""
|
|
List containers on a specific endpoint
|
|
|
|
Args:
|
|
endpoint_id: Portainer endpoint identifier
|
|
all_containers: Include stopped containers (default: True)
|
|
|
|
Returns:
|
|
List of container details
|
|
"""
|
|
params = {"all": 1 if all_containers else 0}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/endpoints/{endpoint_id}/docker/containers/json",
|
|
headers=self._get_headers(),
|
|
params=params
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_container(self, endpoint_id: int, container_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Get detailed information about a specific container
|
|
|
|
Args:
|
|
endpoint_id: Portainer endpoint identifier
|
|
container_id: Container ID or name
|
|
|
|
Returns:
|
|
Container details including network and port information
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/endpoints/{endpoint_id}/docker/containers/{container_id}/json",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def stop_container(self, endpoint_id: int, container_id: str) -> bool:
|
|
"""
|
|
Stop a container
|
|
|
|
Args:
|
|
endpoint_id: Portainer endpoint identifier
|
|
container_id: Container ID or name
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/api/endpoints/{endpoint_id}/docker/containers/{container_id}/stop",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
logger.info(f"Stopped container {container_id}")
|
|
return True
|
|
|
|
async def start_container(self, endpoint_id: int, container_id: str) -> bool:
|
|
"""
|
|
Start a container
|
|
|
|
Args:
|
|
endpoint_id: Portainer endpoint identifier
|
|
container_id: Container ID or name
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/api/endpoints/{endpoint_id}/docker/containers/{container_id}/start",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
logger.info(f"Started container {container_id}")
|
|
return True
|
|
|
|
# ========================================================================
|
|
# Helper methods for agent tools (auto-detect endpoint)
|
|
# ========================================================================
|
|
|
|
async def list_containers(self, all_containers: bool = True) -> List[Dict[str, Any]]:
|
|
"""
|
|
List containers using auto-detected endpoint
|
|
|
|
This is a convenience wrapper that automatically uses the first/default endpoint.
|
|
|
|
Args:
|
|
all_containers: Include stopped containers (default: True)
|
|
|
|
Returns:
|
|
List of container details
|
|
"""
|
|
endpoints = await self.get_endpoints()
|
|
if not endpoints:
|
|
raise RuntimeError("No Portainer endpoints available")
|
|
|
|
endpoint_id = endpoints[0]["Id"]
|
|
return await self.get_containers(endpoint_id, all_containers)
|
|
|
|
async def inspect_container(self, container_name: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Inspect a container by name using auto-detected endpoint
|
|
|
|
This is a convenience wrapper that automatically uses the first/default endpoint.
|
|
|
|
Args:
|
|
container_name: Container name (e.g., "jellyfin", "ollama")
|
|
|
|
Returns:
|
|
Container details or None if not found
|
|
"""
|
|
endpoints = await self.get_endpoints()
|
|
if not endpoints:
|
|
raise RuntimeError("No Portainer endpoints available")
|
|
|
|
endpoint_id = endpoints[0]["Id"]
|
|
|
|
# List all containers to find the one matching the name
|
|
all_containers = await self.get_containers(endpoint_id, all_containers=True)
|
|
|
|
for container in all_containers:
|
|
# Container names come as array like ['/jellyfin']
|
|
names = container.get('Names', [])
|
|
for name in names:
|
|
clean_name = name.lstrip('/')
|
|
if clean_name == container_name or clean_name.lower() == container_name.lower():
|
|
# Get detailed info using container ID
|
|
container_id = container['Id']
|
|
return await self.get_container(endpoint_id, container_id)
|
|
|
|
return None
|
|
|
|
|
|
# Singleton instance
|
|
_portainer_client: Optional[PortainerClient] = None
|
|
|
|
|
|
def get_portainer_client() -> PortainerClient:
|
|
"""Get singleton Portainer client instance"""
|
|
global _portainer_client
|
|
if _portainer_client is None:
|
|
_portainer_client = PortainerClient()
|
|
return _portainer_client
|