Ak Dev New Sandbox Provider

Adding a New Sandbox Provider

This guide walks through adding a new sandbox provider to Agent Kernel. Use the shipped local_subprocess (ak-py/src/agentkernel/sandbox/providers/local_subprocess.py) and docker (ak-py/src/agentkernel/sandbox/providers/docker.py) implementations as reference.

Existing Providers

Planned in later iterations: bedrock_agentcore.

Reference implementations by pattern: docker.py (sync SDK via to_thread), e2b.py (native async SDK + native idle timeout passthrough), daytona.py (sync SDK + native auto-stop + configurable base image/snapshot/env_vars + resource mapping), ec2_ssm.py (attach-only provider with user-identity mapping: sts:AssumeRole + run_as), kubernetes.py (sync SDK via to_thread with plain dict manifests, exec over the WebSocket stream API (the caller's RBAC needs BOTH create and get on pods/exec), user identity via cached per-subject RBAC-impersonation clients, and the instance-level capability override, below). Providers with a native auto-stop take the profile's idle_timeout as a second constructor argument, passed by their factory branch; kubernetes takes it too, sizing the pod's activeDeadlineSeconds orphan ceiling.

Attach-only backends (environments the framework connects to but never owns) subclass AttachedEnvironmentProvider instead of SandboxProvider: it fixes create (binds to the config's attach_to, never provisions) and destroy (no-op) once, so only attach is implemented. The handle subclasses AttachedEnvironment (not Sandbox directly), which fixes close() as a no-op and keeps the class name honest (e.g. EC2SSMEnvironment). Declare provisions=False, attaches_external=True — the factory validates the profile's environment: managed | attached mode against these flags at startup (attach-only providers are rejected under managed, and attached requires attaches_external plus an attach_to target), and the worker never self-heal-provisions or disposes an attached environment.

Architecture Overview

The sandbox capability (ak-py/src/agentkernel/sandbox/) is config-driven and pluggable:

• SandboxProvider (base.py) — one long-lived instance per configured profile backend. Implements create()/destroy() (abstract), attach() when capabilities.attach is declared (the base default raises SandboxCapabilityError), and declares a capabilities class attribute.
• Sandbox (base.py) — a handle to one live sandbox. Implements the two abstract methods execute_code() (language="python") and close(), and optionally execute_command(), upload_file(), download_file(), install_packages() (each raises SandboxCapabilityError in the base until overridden).
• SandboxCapabilities (model.py) — the honest declaration of what the provider supports (isolation tier, shell, languages, files, package_install, stateful, attach, principal_user, policy_network/filesystem/resources). The manager/worker consult it before routing an operation; an operation a provider didn't declare raises SandboxCapabilityError.
• SandboxProviderFactory (factory.py) — resolves a profile's type to a provider. Built-in short names are if/elif branches with real lazy imports; anything with a dot is treated as a dotted path to a SandboxProvider subclass (bring-your-own, no code change needed).
• BrokerWorkerCore (broker/worker.py) — enforces principal and policy fail-closed against the declared capabilities, then calls the provider.

Key principle — declare capabilities honestly. Only claim what the backend truly enforces. If you declare policy_network=True but can't actually restrict egress, you've created a false security guarantee. Under-declaring is safe (the operation raises a clear capability error); over-declaring is a security bug.

Instance-level capability override (#503): when a capability's enforcement depends on operator-asserted infrastructure the provider cannot detect, keep the class declaration honest (False) and flip it per instance in __init__ via self.capabilities = type(self).capabilities.model_copy(update=...) behind an explicit config flag. The reference is the kubernetes provider's network_policy: true, which asserts the cluster CNI actually enforces NetworkPolicy; consumers read provider.capabilities on the instance, so attribute shadowing is sufficient.

Step-by-Step

1. Create the Provider File

Create ak-py/src/agentkernel/sandbox/providers/<provider>.py with a Sandbox subclass and a SandboxProvider subclass.

2. Implement the Sandbox Handle

from agentkernel.sandbox import Sandbox
from agentkernel.sandbox.errors import SandboxCapabilityError
from agentkernel.sandbox.model import SandboxResult


class <Provider>Sandbox(Sandbox):
    def __init__(self, sandbox_id: str, ...) -> None:
        self.id = sandbox_id            # provider-scoped id, stable across attach/reconnect

    async def execute_code(self, code: str, language: str = "python", timeout: float | None = None) -> SandboxResult:
        # language="python" is MANDATORY. A failing program (non-zero exit) is a RESULT
        # (SandboxResult with exit_code != 0), NOT an exception. Raise only on machinery failure.
        if language not in <declared languages>:
            raise SandboxCapabilityError(self.__class__.__name__, f"language:{language}")
        ...
        return SandboxResult(stdout=..., stderr=..., exit_code=...)

    # Override ONLY the optional operations your capabilities declare. The base ABC raises
    # SandboxCapabilityError for execute_command / upload_file / download_file / install_packages
    # when not overridden — that is capability honesty in action.

    async def close(self) -> None:
        # Idempotent. For per_session scope this must NOT destroy state needed for a later
        # attach() — only destroy() permanently disposes backend state.
        ...

Sync SDKs must be wrapped in asyncio.to_thread (see docker.py), since the sandbox runs on an event loop. Enforce timeout with asyncio.wait_for when the backend has no native timeout.

3. Implement the Provider

from agentkernel.sandbox import Sandbox, SandboxProvider
from agentkernel.sandbox.errors import SandboxGoneError
from agentkernel.sandbox.model import IsolationTier, SandboxCapabilities, SandboxPolicy, SandboxPrincipal


class <Provider>SandboxProvider(SandboxProvider):
    capabilities = SandboxCapabilities(
        isolation=IsolationTier.CONTAINER,   # declare the REAL boundary
        shell=True,
        languages=["python"],
        files=True,
        package_install=False,
        stateful=False,
        attach=True,
        provisions=True,                     # False for attach-only backends (never create)
        attaches_external=False,             # True only if attach_to can bind to something you didn't create
        principal_user=False,                # True only if you enforce a user identity
        policy_network=False,                # True only if you actually restrict egress
        policy_filesystem=False,
        policy_resources=False,
    )

    def __init__(self, config) -> None:
        super().__init__(config)             # config is the provider's Pydantic config block
        # Create SDK clients lazily (first use), not here — keep import/constructor cheap.

    async def create(self, *, principal: SandboxPrincipal, policy: SandboxPolicy) -> Sandbox:
        # Provision a new sandbox. Map `policy` onto the backend's real controls here; if a
        # declared-enforceable dimension can't be honored for this request, raise SandboxPolicyError.
        ...

    async def attach(self, sandbox_id: str, *, principal: SandboxPrincipal, policy: SandboxPolicy) -> Sandbox:
        # Reconnect to an existing sandbox. Raise SandboxGoneError when the target is gone —
        # that is the self-heal signal the worker uses to recreate under the same session id.
        ...

    async def destroy(self, sandbox_id: str) -> None:
        # Permanently dispose. Idempotent; unknown ids are a no-op.
        ...

Policy mapping and principal mapping (for principal_user=True) belong in create/attach. See docker.py for the policy mapping pattern (network_egress: deny → network_mode: none, cpu/memory → container limits, filesystem → read-only rootfs + writable workdir).

4. Register with the Factory

Add an if/elif branch to SandboxProviderFactory._build in ak-py/src/agentkernel/sandbox/factory.py, and append the short name to _BUILTIN_PROVIDER_NAMES (this is the #541 house pattern: real lazy imports, no registry map):

if type_name == "<provider>":
    config_block = cls._require_block(profile_name, profile, type_name)
    with require_extra("<extra>", "sandbox provider '<provider>'"):   # skip the wrap if stdlib-only
        from .providers.<provider> import <Provider>SandboxProvider
    return <Provider>SandboxProvider(config_block)

_BUILTIN_PROVIDER_NAMES = ["local_subprocess", "docker", "<provider>"]   # add your name

require_extra produces the friendly pip install "agentkernel[<extra>]" message when the SDK is missing. A bring-your-own provider needs none of this — a dotted-path type resolves via resolve_dotted automatically.