ExecutionContainer

DEFAULT_TAG `module-attribute`

DEFAULT_TAG = 'gradion-ai/ipybox'

ExecutionContainer

ExecutionContainer(
    tag: str = DEFAULT_TAG,
    binds: dict[str, str] | None = None,
    env: dict[str, str] | None = None,
    executor_port: int | None = None,
    resource_port: int | None = None,
    port_allocation_timeout: float = 10,
    show_pull_progress: bool = True,
)

Context manager for the lifecycle of a code execution Docker container. A code execution container provides:

a Jupyter Kernel Gateway for stateful code execution in IPython kernels. Clients connect to it via ExecutionClient on the container's executor host port.
a resource server for downloading Python module sources and registering MCP servers. Clients connect to it via ResourceClient on the container's resource host port.
a firewall that can be enabled with init_firewall to restrict network access to allowed domains, IPv4 addresses, or CIDR ranges.

Parameters:

Name	Type	Description	Default
`tag`	`str`	Name and optionally tag of the `ipybox` Docker image to use (format: `name:tag`)	`DEFAULT_TAG`
`binds`	`dict[str, str] \| None`	A dictionary mapping host paths to container paths for bind mounts. Host paths may be relative or absolute. Container paths must be relative and are created as subdirectories of `/app` in the container.	`None`
`env`	`dict[str, str] \| None`	Environment variables to set in the container	`None`
`executor_port`	`int \| None`	Host port for the container's executor port. A random port is allocated if not specified.	`None`
`resource_port`	`int \| None`	Host port for the container's resource port. A random port is allocated if not specified.	`None`
`port_allocation_timeout`	`float`	Maximum time in seconds to wait for port random allocation.	`10`
`show_pull_progress`	`bool`	Whether to show progress when pulling the Docker image.	`True`

Source code in ipybox/container.py

def __init__(
    self,
    tag: str = DEFAULT_TAG,
    binds: dict[str, str] | None = None,
    env: dict[str, str] | None = None,
    executor_port: int | None = None,
    resource_port: int | None = None,
    port_allocation_timeout: float = 10,
    show_pull_progress: bool = True,
):
    self.tag = tag
    self.binds = binds or {}
    self.env = env or {}
    self.show_pull_progress = show_pull_progress

    self._docker = None
    self._container = None
    self._executor_port = executor_port
    self._resource_port = resource_port
    self._port_allocation_timeout = port_allocation_timeout

executor_port `property`

executor_port: int

The host port of the container's executor port. Either an application-defined executor_port via the constructor or a dynamically allocated random port.

Raises:

Type	Description
`RuntimeError`	If the container is not running and an application-defined port was not provided.

resource_port `property`

resource_port: int

The host port of the container's resource port. Either an application-defined resource_port via the constructor or a dynamically allocated random port.

Raises:

Type	Description
`RuntimeError`	If the container is not running and an application-defined port was not provided.

init_firewall `async`

init_firewall(
    allowed_domains: list[str] | None = None,
) -> None

Initialize firewall rules to restrict internet access to a whitelist of allowed domains, IPv4 addresses, or CIDR ranges.

Traffic policy inside the container after initialisation:

DNS resolution (UDP/53) is always permitted so that the script itself can resolve domains and regular runtime code can still perform look-ups.
SSH (TCP/22) is permitted for interaction with the host.
Loopback traffic is unrestricted.
The host network (*/24 derived from the default gateway) is allowed bidirectionally.
Bidirectional traffic on the ipybox executor (8888) and resource (8900) ports is always allowed.
Outbound traffic is allowed only to the specified whitelist entries.

DNS failures when resolving an allowed domain yield a warning but do not stop the firewall initialization.

A firewall can be initialized multiple times per container. Subsequent calls will clear previous firewall rules and enforce the new allowed_domains list.

Parameters:

Name	Type	Description	Default
`allowed_domains`	`list[str] \| None`	List of domains, IP addresses, or CIDR ranges that should be reachable from the container. If None or empty, only essential services are allowed.	`None`

Raises:

Type	Description
`RuntimeError`	If the container is not running, firewall initialization fails, or if the container is running as root (ipybox images built with -r flag).

Source code in ipybox/container.py

async def init_firewall(self, allowed_domains: list[str] | None = None) -> None:
    """Initialize firewall rules to restrict internet access to a whitelist of
    allowed domains, IPv4 addresses, or CIDR ranges.

    Traffic policy inside the container after initialisation:

    - DNS resolution (UDP/53) is always permitted so that the script itself can resolve
      domains and regular runtime code can still perform look-ups.
    - SSH (TCP/22) is permitted for interaction with the host.
    - Loopback traffic is unrestricted.
    - The host network (\\*/24 derived from the default gateway) is allowed bidirectionally.
    - Bidirectional traffic on the ipybox *executor* (8888) and *resource* (8900) ports
      is always allowed.
    - Outbound traffic is allowed only to the specified whitelist entries.

    DNS failures when resolving an allowed domain yield a warning but do not stop
    the firewall initialization.

    A firewall can be initialized multiple times per container. Subsequent calls will
    clear previous firewall rules and enforce the new `allowed_domains` list.

    Args:
        allowed_domains: List of domains, IP addresses, or CIDR ranges that should be
            reachable from the container. If None or empty, only essential services are
            allowed.

    Raises:
        RuntimeError: If the container is not running, firewall initialization fails,
            or if the container is running as root (ipybox images built with -r flag).
    """
    if not self._container:
        raise RuntimeError("Container not running")

    if allowed_domains is None:
        allowed_domains = []

    # Build command arguments
    cmd_args = ["/usr/local/bin/init-firewall.sh"]
    cmd_args.extend(allowed_domains)
    cmd_args.extend(["--executor-port", str(8888), "--resource-port", str(8900)])

    try:
        # Execute firewall initialization script as root
        exec_instance = await self._container.exec(
            cmd=cmd_args,
            stdout=True,
            stderr=True,
            tty=False,
            user="root",
        )

        output_chunks: list[bytes] = []
        async with exec_instance.start(detach=False) as stream:
            while True:
                msg = await stream.read_out()
                if msg is None:
                    break

                # Append both stdout (stream==1) and stderr (stream==2)
                # data so we don't lose any diagnostic information.
                if msg.data:
                    output_chunks.append(msg.data)

        output_text = b"".join(output_chunks).decode(errors="replace")

        # Check the exit status to ensure the firewall script completed successfully.
        # If the script fails, raise an error with the exit code and the output text.
        inspect_data = await exec_instance.inspect()
        exit_code = inspect_data.get("ExitCode")

        if exit_code not in (0, None):
            error_message = f"init script returned exit code {exit_code}."
            error_message = error_message + f"\n{output_text}" if output_text else ""
            raise RuntimeError(error_message)
        else:
            for line in output_text.splitlines():
                logger.info(line)

    except Exception as e:
        raise RuntimeError(f"Failed to initialize firewall: {str(e)}") from e

kill `async`

kill()

Kills and removes the current code execution Docker container.

Source code in ipybox/container.py

async def kill(self):
    """Kills and removes the current code execution Docker container."""
    if self._container:
        await self._container.kill()

    if self._docker:
        await self._docker.close()

run `async`

run()

Creates and starts a code execution Docker container.

Source code in ipybox/container.py

async def run(self):
    """Creates and starts a code execution Docker container."""
    self._docker = Docker()
    await self._run()

ExecutionContainer

DEFAULT_TAG module-attribute