From dbbb92d29228ffcc912ed31de8d6b65b410a7310 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Wed, 31 Dec 2025 11:13:46 +0100 Subject: [PATCH] refactor: remove Docker socket fallback from Portainer client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- src/clients/portainer_client.py | 134 ++++++-------------------------- 1 file changed, 22 insertions(+), 112 deletions(-) diff --git a/src/clients/portainer_client.py b/src/clients/portainer_client.py index f98dad3..e73a474 100644 --- a/src/clients/portainer_client.py +++ b/src/clients/portainer_client.py @@ -438,70 +438,14 @@ class PortainerClient: return True # ======================================================================== - # Docker Socket Fallback (for containers not managed by Portainer) - # ======================================================================== - - async def _list_containers_via_socket(self, all_containers: bool = True) -> List[Dict[str, Any]]: - """ - Fallback: List containers directly via Docker socket - - Used when Portainer API doesn't return complete data (e.g., containers - started outside Portainer, AMP game servers, etc.) - - Args: - all_containers: Include stopped containers - - Returns: - List of container details in Docker API format - """ - try: - # Docker socket is mounted at /var/run/docker.sock - # Use httpx with unix socket transport - transport = httpx.AsyncHTTPTransport(uds="/var/run/docker.sock") - async with httpx.AsyncClient(transport=transport, timeout=10) as client: - params = {"all": 1 if all_containers else 0} - response = await client.get( - "http://localhost/v1.41/containers/json", - params=params - ) - response.raise_for_status() - return response.json() - except Exception as e: - logger.warning(f"Docker socket fallback failed: {e}") - return [] - - async def _inspect_container_via_socket(self, container_id_or_name: str) -> Optional[Dict[str, Any]]: - """ - Fallback: Inspect container directly via Docker socket - - Args: - container_id_or_name: Container ID or name - - Returns: - Container details or None - """ - try: - transport = httpx.AsyncHTTPTransport(uds="/var/run/docker.sock") - async with httpx.AsyncClient(transport=transport, timeout=10) as client: - response = await client.get( - f"http://localhost/v1.41/containers/{container_id_or_name}/json" - ) - response.raise_for_status() - return response.json() - except Exception as e: - logger.warning(f"Docker socket inspect fallback failed for '{container_id_or_name}': {e}") - return None - - # ======================================================================== - # Helper methods for agent tools (auto-detect endpoint + fallback) + # 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 with Docker socket fallback + List containers using auto-detected endpoint This is a convenience wrapper that automatically uses the first/default endpoint. - If Portainer doesn't have complete data, falls back to Docker socket. Args: all_containers: Include stopped containers (default: True) @@ -509,34 +453,18 @@ class PortainerClient: Returns: List of container details """ - try: - # Try Portainer first - endpoints = await self.get_endpoints() - if endpoints: - endpoint_id = endpoints[0]["Id"] - containers = await self.get_containers(endpoint_id, all_containers) - if containers: - return containers + endpoints = await self.get_endpoints() + if not endpoints: + raise RuntimeError("No Portainer endpoints available") - # Fallback to Docker socket - logger.info("Portainer returned no containers, trying Docker socket fallback...") - return await self._list_containers_via_socket(all_containers) - - except Exception as e: - logger.error(f"Error listing containers: {e}") - # Try fallback even on exception - try: - return await self._list_containers_via_socket(all_containers) - except Exception as fallback_error: - logger.error(f"Fallback also failed: {fallback_error}") - return [] + 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 with Docker socket fallback + Inspect a container by name using auto-detected endpoint This is a convenience wrapper that automatically uses the first/default endpoint. - If Portainer doesn't find the container, falls back to Docker socket. Args: container_name: Container name (e.g., "jellyfin", "ollama") @@ -544,44 +472,26 @@ class PortainerClient: Returns: Container details or None if not found """ - try: - # Try Portainer first - endpoints = await self.get_endpoints() - if endpoints: - endpoint_id = endpoints[0]["Id"] + endpoints = await self.get_endpoints() + if not endpoints: + raise RuntimeError("No Portainer endpoints available") - # First list all containers to find the one matching the name - all_containers = await self.get_containers(endpoint_id, all_containers=True) + endpoint_id = endpoints[0]["Id"] - matching_container = None - 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(): - matching_container = container - break - if matching_container: - break + # List all containers to find the one matching the name + all_containers = await self.get_containers(endpoint_id, all_containers=True) - if matching_container: + 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 = matching_container['Id'] + container_id = container['Id'] return await self.get_container(endpoint_id, container_id) - # Not found in Portainer, try Docker socket fallback - logger.info(f"Container '{container_name}' not found in Portainer, trying Docker socket fallback...") - return await self._inspect_container_via_socket(container_name) - - except Exception as e: - logger.error(f"Error inspecting container '{container_name}': {e}") - # Try fallback even on exception - try: - return await self._inspect_container_via_socket(container_name) - except Exception as fallback_error: - logger.error(f"Fallback also failed: {fallback_error}") - return None + return None # Singleton instance