DaytonaCodeEnvironment(*, api_key=None, api_url=None, target=None, snapshot=None, image=None, env_vars=None, resources=None, timeout=60, languages=('python', 'bash', 'javascript', 'typescript'))
Bases: CodeEnvironment
:class:~autogen.beta.tools.code.CodeEnvironment backed by a Daytona sandbox.
The sandbox is created lazily on the first :meth:run call and reused for the lifetime of the environment. Cleanup is registered via :func:atexit so the sandbox is released even if the user forgets to call :meth:aclose — for tighter scoping use async with DaytonaCodeEnvironment(...) as env.
All sandbox-shaping parameters (api_key, api_url, target, snapshot, image, env_vars) accept a :class:~autogen.beta.annotations.Variable for deferred resolution from context.variables — useful for per-tenant credentials or A/B-tested images. Variables are resolved on the first :meth:run call, when the sandbox is created.
| PARAMETER | DESCRIPTION |
api_key | Daytona API key. Falls back to DAYTONA_API_KEY. TYPE: str | Variable | None DEFAULT: None |
api_url | Daytona API URL. Falls back to DAYTONA_API_URL. TYPE: str | Variable | None DEFAULT: None |
target | Daytona target region (e.g. "us", "eu"). Falls back to DAYTONA_TARGET. TYPE: str | Variable | None DEFAULT: None |
snapshot | Snapshot name. Mutually exclusive with image. TYPE: str | Variable | None DEFAULT: None |
image | Custom Docker image. Mutually exclusive with snapshot. TYPE: str | Image | Variable | None DEFAULT: None |
env_vars | Environment variables passed into the sandbox. TYPE: dict[str, str] | Variable | None DEFAULT: None |
resources | Resource limits. Only applied with image. TYPE: DaytonaResources | None DEFAULT: None |
timeout | Per-execution timeout in seconds. Default: 60. TYPE: int DEFAULT: 60 |
languages | Languages this environment will accept. Defaults to all four supported by Daytona. TYPE: tuple[CodeLanguage, ...] DEFAULT: ('python', 'bash', 'javascript', 'typescript') |
Source code in autogen/beta/extensions/daytona/environment.py
| def __init__(
self,
*,
api_key: str | Variable | None = None,
api_url: str | Variable | None = None,
target: str | Variable | None = None,
snapshot: str | Variable | None = None,
image: "str | Image | Variable | None" = None,
env_vars: dict[str, str] | Variable | None = None,
resources: DaytonaResources | None = None,
timeout: int = 60,
languages: tuple[CodeLanguage, ...] = ("python", "bash", "javascript", "typescript"),
) -> None:
if (
snapshot is not None
and image is not None
and not isinstance(snapshot, Variable)
and not isinstance(image, Variable)
):
raise ValueError("Specify either `snapshot` or `image`, not both.")
if timeout < 1:
raise ValueError("`timeout` must be >= 1 second.")
self._api_key = api_key
self._api_url = api_url
self._target = target
self._snapshot = snapshot
self._image = image
self._env_vars = env_vars
self._resources = resources
self._timeout = timeout
self._languages: tuple[CodeLanguage, ...] = tuple(languages)
self._client: AsyncDaytona | None = None
self._sandbox: Any = None
self._lock = asyncio.Lock()
self._closed = False
|
property
async
run(code, language, *, context=None)
Source code in autogen/beta/extensions/daytona/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,
)
sandbox = await self._ensure_sandbox(context)
try:
if language == "python":
response = await sandbox.process.code_run(code, timeout=self._timeout)
else:
ext = _LANG_FILE_EXT[language]
script_path = f"/tmp/ag2_{uuid.uuid4().hex}.{ext}"
await sandbox.fs.upload_file(code.encode("utf-8"), script_path)
cmd = f"{_LANG_RUN_CMD[language]} {script_path}"
response = await sandbox.process.exec(cmd, timeout=self._timeout)
try:
await sandbox.fs.delete_file(script_path)
except DaytonaNotFoundError:
pass
except Exception as e:
logger.debug("Failed to delete temp script %s: %s", script_path, e)
except DaytonaTimeoutError as e:
return CodeRunResult(output=f"Execution timed out: {e}", exit_code=124)
except DaytonaRateLimitError as e:
return CodeRunResult(output=f"Daytona rate limit exceeded: {e}", exit_code=1)
except DaytonaError as e:
return CodeRunResult(output=f"Daytona error: {e}", exit_code=1)
return CodeRunResult(output=response.result or "", exit_code=response.exit_code or 0)
|
async
Delete the sandbox and close the client. Safe to call multiple times.
Source code in autogen/beta/extensions/daytona/environment.py
| async def aclose(self) -> None:
"""Delete the sandbox and close the client. Safe to call multiple times."""
atexit.unregister(self._atexit_close)
self._closed = True
if self._sandbox is not None:
try:
await self._sandbox.delete()
except DaytonaNotFoundError:
pass
except Exception as e:
logger.debug("Suppressed exception during sandbox deletion: %s", e)
self._sandbox = None
if self._client is not None:
try:
await self._client.close()
except Exception as e:
logger.debug("Suppressed exception during client close: %s", e)
self._client = None
|