DockerCodeEnvironment(*, image='python:3.12-slim', env_vars=None, timeout=60, network_mode='none', mem_limit='512m', cpu_quota=None, user=None, auto_remove=True, languages=('python', 'bash'))
Bases: CodeEnvironment
:class:~autogen.beta.tools.code.CodeEnvironment backed by a local Docker container.
A long-lived container is created lazily on the first :meth:run call (running sleep infinity so it stays up between calls) and reused for the lifetime of the environment. Each run_code invocation issues a docker exec against that container.
Safety defaults:
network_mode="none" — no network access. Set to "bridge" to opt in.mem_limit="512m" — bound runaway processes.auto_remove=True and explicit aclose() ensure the container is removed when no longer needed.
All sandbox-shaping parameters (image, env_vars, network_mode) accept a :class:~autogen.beta.annotations.Variable for deferred resolution from context.variables.
| PARAMETER | DESCRIPTION |
image | Docker image. Default "python:3.12-slim" (has python and bash). TYPE: str | Variable DEFAULT: 'python:3.12-slim' |
env_vars | Environment variables passed into the container. TYPE: dict[str, str] | Variable | None DEFAULT: None |
timeout | Per-execution timeout in seconds. Default: 60. TYPE: int DEFAULT: 60 |
network_mode | Container network mode. Default "none" (no network); "bridge" for default network, "host" to share host networking. TYPE: str | Variable DEFAULT: 'none' |
mem_limit | Memory limit (Docker syntax, e.g. "512m", "1g"). Default: "512m". None disables. TYPE: str | None DEFAULT: '512m' |
cpu_quota | CPU quota in microseconds per 100ms period. None disables. TYPE: int | None DEFAULT: None |
user | User to run as inside the container. None (default) uses the image's default user. Recommended override: "nobody" for images that ship that user. TYPE: str | None DEFAULT: None |
auto_remove | Whether Docker should auto-remove the container on stop. Default: True. TYPE: bool DEFAULT: True |
languages | Languages this environment will accept. Default ("python", "bash") — these come for free with python:3.12-slim. Add "javascript" / "typescript" only if your image has node / ts-node. TYPE: tuple[CodeLanguage, ...] DEFAULT: ('python', 'bash') |
Source code in autogen/beta/extensions/docker/environment.py
| def __init__(
self,
*,
image: "str | Variable" = "python:3.12-slim",
env_vars: "dict[str, str] | Variable | None" = None,
timeout: int = 60,
network_mode: "str | Variable" = "none",
mem_limit: str | None = "512m",
cpu_quota: int | None = None,
user: str | None = None,
auto_remove: bool = True,
languages: tuple[CodeLanguage, ...] = ("python", "bash"),
) -> None:
if timeout < 1:
raise ValueError("`timeout` must be >= 1 second.")
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._languages: tuple[CodeLanguage, ...] = tuple(languages)
self._client: docker.DockerClient | None = None
self._container: Any = None
self._lock = asyncio.Lock()
self._closed = False
|
property
async
run(code, language, *, context=None)
Source code in autogen/beta/extensions/docker/environment.py
| async def run(
self,
code: str,
language: CodeLanguage,
*,
context: "ConversationContext | None" = None,
) -> CodeRunResult:
if language not in self._languages:
return CodeRunResult(
output=f"Language {language!r} is not enabled. Available: {list(self._languages)}",
exit_code=2,
)
container = await self._ensure_container(context)
if language in _LANG_RUN_CMD:
cmd: list[str] = [*_LANG_RUN_CMD[language], code]
else:
# File-based execution for js/ts: base64-decode the snippet
# into a temp file inside /workspace, exec the runner, clean up.
ext = _LANG_FILE_EXT[language]
runner = _FILE_RUN_CMD[language]
script_path = f"/workspace/ag2_{uuid.uuid4().hex}.{ext}"
encoded = base64.b64encode(code.encode("utf-8")).decode("ascii")
cmd = [
"sh",
"-c",
f"echo {encoded} | base64 -d > {script_path} && {runner} {script_path}; "
f"rc=$?; rm -f {script_path}; exit $rc",
]
try:
result = await asyncio.wait_for(
asyncio.to_thread(container.exec_run, cmd, stderr=True, stdout=True, workdir="/workspace"),
timeout=self._timeout,
)
except asyncio.TimeoutError:
# docker exec has no clean cancel API — recycle the container so a runaway
# process doesn't keep consuming resources.
await self._restart_container()
return CodeRunResult(
output=f"Execution timed out after {self._timeout}s",
exit_code=124,
)
except APIError as e:
return CodeRunResult(output=f"Docker API error: {e}", exit_code=1)
output_bytes = result.output if isinstance(result.output, bytes) else b"".join(result.output or [])
return CodeRunResult(
output=output_bytes.decode(errors="replace"),
exit_code=result.exit_code or 0,
)
|
async
Stop and remove the container. Safe to call multiple times.
Source code in autogen/beta/extensions/docker/environment.py
| async def aclose(self) -> None:
"""Stop and remove the container. Safe to call multiple times."""
atexit.unregister(self._atexit_close)
self._closed = True
if self._container is not None:
try:
await asyncio.to_thread(self._container.stop, timeout=1)
except NotFound:
pass
except Exception as e:
logger.debug("Suppressed exception during container stop: %s", e)
if not self._auto_remove:
try:
await asyncio.to_thread(self._container.remove, force=True)
except NotFound:
pass
except Exception as e:
logger.debug("Suppressed exception during container remove: %s", e)
self._container = None
if self._client is not None:
try:
await asyncio.to_thread(self._client.close)
except Exception as e:
logger.debug("Suppressed exception during client close: %s", e)
self._client = None
|