Skip to content

DockerEnvironment

autogen.beta.extensions.docker.environment.DockerEnvironment #

DockerEnvironment(*, image='python:3.12-slim', env_vars=None, timeout=60, network_mode='none', mem_limit='512m', cpu_quota=None, user=None, auto_remove=True, host_path=None, workdir='/workspace')

:class:~autogen.beta.tools.sandbox.SandboxFactory for :class:DockerSandbox.

This is the backend object you hand to a tool::

env = DockerEnvironment(image="python:3.12-slim", network_mode="none")
shell = SandboxShellTool(env, allowed=["git"])
code = SandboxCodeTool(env, languages=("python",))

All :class:~autogen.beta.annotations.Variable-capable parameters (image, env_vars, network_mode) are resolved on :meth:open against the active :class:~autogen.beta.context.ConversationContext. The resulting :class:DockerSandbox receives only concrete values, so the backend never has to know about Variables or Context.

The factory owns the sandbox lifecycle and caches by resolved parameters: the first :meth:open for a given set of resolved values starts a container; subsequent opens with the same values reuse it, so state (created files, installed packages) persists across tool calls. Different resolved values (e.g. a per-tenant image=Variable(...)) get distinct containers. Cached containers live until :meth:aclose (or process-exit atexit, registered per container).

Source code in autogen/beta/extensions/docker/environment.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    *,
    image: "str | Variable" = "python:3.12-slim",
    env_vars: "dict[str, str] | Variable | None" = None,
    timeout: float = 60,
    network_mode: "str | Variable" = "none",
    mem_limit: str | None = "512m",
    cpu_quota: int | None = None,
    user: str | None = None,
    auto_remove: bool = True,
    host_path: str | os.PathLike[str] | None = None,
    workdir: str = "/workspace",
) -> None:
    self._image = image
    self._env_vars = env_vars
    self._timeout = timeout
    self._network_mode = network_mode
    self._mem_limit = mem_limit
    self._cpu_quota = cpu_quota
    self._user = user
    self._auto_remove = auto_remove
    self._host_path = host_path
    self._workdir = workdir

    self._cache: dict[Hashable, DockerSandbox] = {}
    self._cache_lock = threading.Lock()

open async #

open(context=None)
Source code in autogen/beta/extensions/docker/environment.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@asynccontextmanager
async def open(
    self,
    context: "ConversationContext | None" = None,
) -> AsyncIterator[DockerSandbox]:
    image = resolve_variable(self._image, context, param_name="image") if context else self._image
    env_vars = (
        resolve_variable(self._env_vars, context, param_name="env_vars") if context else self._env_vars
    ) or {}
    network_mode = (
        resolve_variable(self._network_mode, context, param_name="network_mode") if context else self._network_mode
    )

    for value, name in ((image, "image"), (network_mode, "network_mode")):
        if isinstance(value, Variable):
            raise RuntimeError(
                f"Docker `{name}` given as Variable but no Context available to resolve it. "
                "Variables are only resolvable when the sandbox is driven from an Agent "
                "(SandboxCodeTool / SandboxShellTool wrappers forward the active Context)."
            )

    assert isinstance(image, str)
    assert isinstance(network_mode, str)
    assert isinstance(env_vars, dict)

    key: Hashable = (
        image,
        network_mode,
        tuple(sorted(env_vars.items())),
        self._mem_limit,
        self._cpu_quota,
        self._user,
        self._auto_remove,
        str(self._host_path) if self._host_path is not None else None,
        self._workdir,
        self._timeout,
    )

    # Reserving the cache slot is loop-agnostic (plain threading.Lock,
    # no await) so concurrent openers across throw-away event loops agree
    # on one sandbox object. The container itself starts in __aenter__,
    # which is internally guarded and idempotent.
    with self._cache_lock:
        sandbox = self._cache.get(key)
        if sandbox is None or sandbox.closed:
            sandbox = DockerSandbox(
                image=image,
                env_vars=dict(env_vars),
                timeout=self._timeout,
                network_mode=network_mode,
                mem_limit=self._mem_limit,
                cpu_quota=self._cpu_quota,
                user=self._user,
                auto_remove=self._auto_remove,
                host_path=self._host_path,
                workdir=self._workdir,
            )
            self._cache[key] = sandbox

    await sandbox.__aenter__()
    # Do NOT close on scope exit — the factory owns the lifecycle so the
    # container is reused by the next open() with the same key.
    yield sandbox

aclose async #

aclose()

Tear down every cached container. Safe to call multiple times.

Source code in autogen/beta/extensions/docker/environment.py 
138
139
140
141
142
143
144
async def aclose(self) -> None:
    """Tear down every cached container. Safe to call multiple times."""
    with self._cache_lock:
        sandboxes = list(self._cache.values())
        self._cache.clear()
    for sandbox in sandboxes:
        await sandbox.aclose()