Docker API

Docker Engine API & docker-py Integration

Integrates with the Docker Engine API via the official docker-py Python SDK to programmatically manage containers, images, networks, volumes, and Swarm clusters. Enables building container management tooling, CI/CD orchestration, and infrastructure automation.

TL;DR for Code Generation

• Connect with docker.from_env() to use the default socket or docker.DockerClient(base_url='tcp://...') for remote
• Use client.containers.run() with detach=True for background execution
• Always specify auto_remove=True for ephemeral containers to prevent resource leaks
• Handle docker.errors.APIError with specific status codes for robust error recovery
• Use client.images.build() with rm=True to clean up intermediate build containers
• Set DOCKER_API_VERSION explicitly to pin a specific API version for compatibility

When to Use

Use this skill when:

• Building custom container orchestration tools or CLI wrappers around Docker
• Automating image builds, tagging, and pushes to container registries
• Managing multi-container environments for development or testing
• Implementing Swarm service deployment and scaling logic
• Building container health-check and log collection pipelines
• Creating ephemeral build or test environments per CI job

When NOT to Use

Avoid this skill for:

• Production Kubernetes orchestration (use the coding-kubernetes-api skill)
• Declarative infrastructure-as-code (use coding-terraform-sdk or docker compose)
• Low-level HTTP calls to the Engine API (the SDK handles serialization and auth)

Core Workflow

1. Connect to Docker Daemon — Use docker.from_env() for local socket (Unix) or DockerClient(base_url) for TCP/TLS. Checkpoint: Call client.ping() to verify the daemon is reachable and responsive.

2. Pull or Build Image — Use client.images.pull(repository, tag) or client.images.build(path=...). Checkpoint: Verify the image ID is present in client.images.list(name=...).

3. Create and Run Container — Use client.containers.run(image, command, detach=True, ...) or create the container separately then call .start(). Checkpoint: Inspect container.status to confirm it is running or exited with the expected exit code.

4. Manage Container Lifecycle — Call .stop(), .restart(), .kill(), and .remove() with appropriate timeouts. Checkpoint: Verify container.status transitions to exited or the container is removed from client.containers.list(all=True).

5. Stream Logs and Collect Results — Use container.logs(stream=True) for real-time output and container.wait() to block on exit. Checkpoint: Check exit code via container.attrs['State']['ExitCode'].

Implementation Patterns

Pattern 1: Build and Push a Docker Image

import os
import docker
from docker.errors import APIError, BuildError

def build_and_push_image(
    path: str,
    image_name: str,
    tag: str = "latest",
    dockerfile: str = "Dockerfile",
    build_args: dict | None = None,
) -> dict:
    """Build a Docker image and push it to a registry.

    Args:
        path: Build context directory.
        image_name: Repository name (e.g., "my-registry.com/my-app").
        tag: Image tag.
        dockerfile: Relative path to Dockerfile within context.
        build_args: Build-time variables for ARG instructions.

    Returns:
        Dict with image ID, tags, and build log.

    Raises:
        BuildError: If the image build fails.
        APIError: If the push to registry fails.
    """
    client = docker.from_env()

    # Build the image
    image, build_logs = client.images.build(
        path=path,
        tag=f"{image_name}:{tag}",
        dockerfile=dockerfile,
        buildargs=build_args or {},
        rm=True,
        forcerm=True,
    )

    log_output = ""
    for chunk in build_logs:
        if "stream" in chunk:
            log_output += chunk["stream"]

    # Tag with additional tags if needed
    image.tag(image_name, tag)

    # Push to registry
    push_log = client.images.push(image_name, tag=tag)
    push_status = push_log.get("status", "unknown")

    return {
        "image_id": image.id,
        "tags": image.tags,
        "build_log": log_output,
        "push_status": push_status,
    }

Pattern 2: Orchestrated Container Execution with Log Streaming

import sys
import docker
from docker.errors import APIError, ContainerError


def run_container_with_streaming(
    image: str,
    command: str | list[str],
    name: str | None = None,
    environment: dict | None = None,
    volumes: dict | None = None,
    network: str | None = None,
    timeout: int = 300,
) -> tuple[int, str]:
    """Run a container with real-time log streaming and timeout.

    Args:
        image: Docker image to use.
        command: Command or entrypoint override.
        name: Container name.
        environment: Environment variables dict.
        volumes: Volume bindings dict (e.g., {"/host/path": {"bind": "/container/path", "mode": "rw"}}).
        network: Network to attach the container to.
        timeout: Maximum runtime in seconds.

    Returns:
        Tuple of (exit_code, full_log_output).

    Raises:
        ContainerError: If the container exits with non-zero code.
        APIError: If Docker API operations fail.
    """
    client = docker.from_env()

    container = client.containers.run(
        image=image,
        command=command,
        name=name,
        environment=environment,
        volumes=volumes,
        network=network,
        detach=True,
        auto_remove=False,  # Keep for log retrieval
    )

    # Stream logs in real-time
    full_log: list[str] = []
    try:
        for log_line in container.logs(stream=True, follow=True):
            decoded = log_line.decode("utf-8", errors="replace").rstrip()
            full_log.append(decoded)
            sys.stdout.write(f"{decoded}\n")
    except KeyboardInterrupt:
        container.stop()
        raise

    # Wait for container to finish
    result = container.wait(timeout=timeout)
    exit_code = result.get("StatusCode", -1)

    # Clean up
    container.remove()

    return exit_code, "\n".join(full_log)

Pattern 3: Swarm Service Management

import docker
from docker.errors import APIError
from docker.types.services import ServiceMode, Resources, RestartPolicy


def create_swarm_service(
    service_name: str,
    image: str,
    replicas: int = 3,
    env_vars: dict | None = None,
    publish_port: tuple[int, int] | None = None,
) -> dict:
    """Create or update a Docker Swarm service.

    Args:
        service_name: Name for the service.
        image: Docker image with tag.
        replicas: Desired number of replicas.
        env_vars: Environment variables for the service.
        publish_port: (host_port, container_port) tuple.

    Returns:
        Dict with service ID, mode, and replica count.
    """
    client = docker.from_env()

    # Ensure we're in Swarm mode
    try:
        client.swarm.reload()
    except APIError:
        client.swarm.init()

    env_list = [f"{k}={v}" for k, v in (env_vars or {}).items()]

    service_kwargs = {
        "name": service_name,
        "image": image,
        "env": env_list,
        "mode": ServiceMode("replicated", replicas=replicas),
        "restart_policy": RestartPolicy(
            condition="any",
            max_attempts=3,
        ),
        "resources": Resources(
            cpu_limit=0.5,
            mem_limit="256M",
            cpu_reservation=0.25,
            mem_reservation="128M",
        ),
    }

    if publish_port:
        host_port, container_port = publish_port
        service_kwargs["endpoint_spec"] = docker.types.EndpointSpec(
            mode="vip",
            ports={container_port: host_port},
        )

    try:
        service = client.services.create(**service_kwargs)
    except APIError as exc:
        raise RuntimeError(
            f"Failed to create Swarm service '{service_name}': {exc}"
        ) from exc

    return {
        "service_id": service.id,
        "name": service.name,
        "mode": "replicated",
        "replicas": replicas,
        "image": image,
    }

BAD vs GOOD: Container Cleanup

# ❌ BAD — containers and images pile up, no cleanup
def run_bad(image):
    client = docker.from_env()
    container = client.containers.run(image, detach=True)
    # Never removed — resource leak!

# ✅ GOOD — always clean up with auto_remove and finally blocks
def run_good(image: str, command: str) -> str:
    """Run a command in a container and return output.

    Ensures the container is removed even if an error occurs.
    """
    client = docker.from_env()
    container = None
    try:
        container = client.containers.run(
            image=image,
            command=command,
            detach=False,
            remove=True,  # Auto-remove after exit
        )
        return container.decode("utf-8")
    except APIError as exc:
        raise RuntimeError(
            f"Container execution failed for image '{image}': {exc}"
        ) from exc

MUST DO

• Use docker.from_env() for local socket connections — it reads the DOCKER_HOST and DOCKER_API_VERSION environment variables
• Always set remove=True or auto_remove=True for ephemeral containers to prevent resource leaks
• Use client.containers.list(all=True, filters={}) with filters to avoid fetching all containers
• Pin image tags explicitly (e.g., python:3.12-slim) rather than using latest
• Use forcerm=True when building images to remove intermediate containers immediately
• Handle docker.errors.DockerException (connection errors) separately from APIError (API-level errors)

MUST NOT DO

• Never run containers with privileged=True unless absolutely necessary — it's a security risk
• Do not use the docker-py library to control a remote Docker daemon over an unencrypted TCP socket
• Avoid using auto_remove=False for long-running containers that need cleanup
• Never ignore BuildError when building images — the build log contains essential diagnostics
• Do not hardcode registry credentials in build or push code — use credential helpers or environment variables

Constraints

MUST DO

• Implement structured error responses with consistent format: {error_code, message, details, request_id}
• Add rate limiting per client/API key with configurable burst and sustained limits using a token bucket algorithm
• Validate all incoming requests against a schema before processing — reject malformed input with clear error messages
• Include correlation/request IDs in all log entries for end-to-end request tracing across service boundaries

MUST NOT DO

• Do not expose internal implementation details, stack traces, or database queries in error responses
• Avoid accepting unbounded request bodies — set maximum payload sizes and timeout limits
• Never trust client-supplied authentication tokens without validation (signature verification, expiration check)
• Do not log request/response bodies containing PII, API keys, or other sensitive data

Live References

• docker-py Documentation
• docker-py GitHub Repository
• Docker Engine API Reference
• Docker SDK for Python — Containers
• Docker SDK for Python — Swarm Services
• Dockerfile Reference
• Docker Engine API Version Matrix

Related Skills