CodeAdapter(sandbox, *, languages=('python', 'bash'), runners=None, timeout=None)
One :class:~autogen.beta.tools.code.CodeEnvironment implementation that works on every :class:Sandbox.
The language matrix (python/bash → inline; js/ts → file) lives here once. Backends only need to implement :meth:Sandbox.exec (always) plus :meth:Sandbox.put_file (only for file-mode languages).
| PARAMETER | DESCRIPTION |
sandbox | Either a long-lived :class:Sandbox or a :class:SandboxFactory (opened per :meth:run). TYPE: Sandbox | SandboxFactory |
languages | Languages this adapter is willing to accept; surfaced as :attr:supported_languages. TYPE: tuple[CodeLanguage, ...] DEFAULT: ('python', 'bash') |
runners | Override or extend :data:DEFAULT_RUNNERS. Use this hook to add new languages or tweak interpreters (ts-node vs tsx, etc.). TYPE: dict[CodeLanguage, LanguageRunner] | None DEFAULT: None |
timeout | Per-call timeout in seconds. None lets the backend pick its default. TYPE: float | None DEFAULT: None |
Source code in autogen/beta/tools/sandbox/adapter/code.py
| def __init__(
self,
sandbox: "Sandbox | SandboxFactory",
*,
languages: tuple[CodeLanguage, ...] = ("python", "bash"),
runners: dict[CodeLanguage, LanguageRunner] | None = None,
timeout: float | None = None,
) -> None:
merged = dict(DEFAULT_RUNNERS)
if runners:
merged.update(runners)
for lang in languages:
if lang not in merged:
raise ValueError(f"No LanguageRunner registered for language {lang!r}.")
self._factory: SandboxFactory = sandbox if isinstance(sandbox, SandboxFactory) else SingletonFactory(sandbox)
self._languages: tuple[CodeLanguage, ...] = tuple(languages)
self._runners = merged
self._timeout = timeout
|
run(code, language, *, context=None)
Source code in autogen/beta/tools/sandbox/adapter/code.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,
)
runner = self._runners[language]
async with self._open(context) as sandbox:
if runner.inline_argv is not None:
result = await sandbox.exec(
[*runner.inline_argv, code],
timeout=self._timeout,
)
return CodeRunResult(output=result.output, exit_code=result.exit_code)
assert runner.file_extension is not None and runner.file_runner_argv is not None
filename = f"ag2_{uuid.uuid4().hex}.{runner.file_extension}"
path = PurePosixPath(filename)
await sandbox.put_file(path, code.encode("utf-8"))
target = sandbox.workdir / path
try:
result = await sandbox.exec(
[*runner.file_runner_argv, str(target)],
timeout=self._timeout,
)
finally:
# Best-effort cleanup so persistent sandboxes (e.g. a
# reused container or a LocalSandbox over a fixed workdir)
# do not accumulate temp scripts. Never fails the run.
with suppress(Exception):
await sandbox.remove_file(path)
return CodeRunResult(output=result.output, exit_code=result.exit_code)
|