"""Temporal Sandbox agent example. Runs a SandboxAgent as a durable Temporal workflow. The workflow is long-lived and conversational: after processing each turn it idles waiting for the next user message. Workflows persist indefinitely in Temporal. A separate session manager workflow (``temporal_session_manager.py``) orchestrates session creation, destruction, and discovery. Usage ----- Install the Temporal extra first:: uv sync --extra temporal --extra daytona Start a local Temporal server (requires the Temporal CLI):: temporal server start-dev In one terminal, start the worker:: python examples/sandbox/extensions/temporal_sandbox_agent.py worker In another terminal, start the TUI:: python examples/sandbox/extensions/temporal_sandbox_agent.py run """ from __future__ import annotations import argparse import asyncio import json import os as _os import sys from datetime import timedelta from enum import Enum from pathlib import Path from typing import Any, Literal, cast from pydantic import BaseModel, SerializeAsAny, field_validator, model_serializer from temporalio import workflow from temporalio.client import Client from temporalio.contrib.openai_agents.workflow import temporal_sandbox_client from temporalio.worker import Worker from temporalio.worker.workflow_sandbox import ( SandboxedWorkflowRunner, SandboxRestrictions, ) from agents import ModelSettings, Runner from agents.agent import Agent from agents.extensions.sandbox import ( DaytonaSandboxClientOptions, DaytonaSandboxSessionState, E2BSandboxClientOptions, E2BSandboxSessionState, ) from agents.items import ( MessageOutputItem, RunItem, ToolApprovalItem, ToolCallItem, TResponseInputItem, ) from agents.lifecycle import RunHooksBase from agents.run import RunConfig from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig from agents.sandbox.sandboxes import ( DockerSandboxClientOptions, DockerSandboxSessionState, UnixLocalSandboxClientOptions, UnixLocalSandboxSessionState, ) from agents.sandbox.session.sandbox_session_state import SandboxSessionState from agents.sandbox.snapshot import SnapshotBase # Allow sibling and repo-root imports. _THIS_DIR = _os.path.dirname(_os.path.abspath(__file__)) _REPO_ROOT = _os.path.abspath(_os.path.join(_THIS_DIR, "..", "..", "..", "..")) for _p in (_THIS_DIR, _REPO_ROOT): if _p not in sys.path: sys.path.insert(0, _p) from examples.sandbox.misc.workspace_shell import WorkspaceShellCapability # noqa: E402 class SandboxBackend(str, Enum): DAYTONA = "daytona" DOCKER = "docker" E2B = "e2b" LOCAL = "local" DEFAULT_BACKEND = SandboxBackend.DAYTONA TASK_QUEUE = "sandbox-agent-queue" class _AlwaysSerializeType(BaseModel): """Base that ensures the ``type`` discriminator survives ``exclude_unset`` round-trips.""" @model_serializer(mode="wrap") def _serialize_always_include_type(self, handler: Any) -> dict[str, Any]: data: dict[str, Any] = handler(self) data["type"] = self.type # type: ignore[attr-defined] return data class SwitchToLocalBackend(_AlwaysSerializeType): """Switch target for the local unix sandbox backend.""" type: Literal["local"] = "local" workspace_root: str = "/workspace" class SwitchBackendSignal(BaseModel): """Payload for the ``switch_backend`` signal.""" target: Literal["daytona", "docker", "e2b"] | SwitchToLocalBackend # --------------------------------------------------------------------------- # Workflow input / output types # --------------------------------------------------------------------------- class _HasSnapshot(BaseModel): @field_validator("snapshot", mode="before", check_fields=False) @classmethod def _parse_snapshot(cls, v: object) -> SnapshotBase | None: if v is None or isinstance(v, SnapshotBase): return v return SnapshotBase.parse(v) class WorkflowSnapshot(_HasSnapshot): """Atomic snapshot of an agent workflow's forkable state.""" sandbox_session_state: ( DaytonaSandboxSessionState | DockerSandboxSessionState | E2BSandboxSessionState | UnixLocalSandboxSessionState | None ) = None snapshot: SerializeAsAny[SnapshotBase] | None = ( None # serialized SnapshotBase for cross-backend creation ) previous_response_id: str | None = None history: list[dict[str, Any]] = [] class AgentRequest(_HasSnapshot): messages: list[dict[str, Any]] cwd: str = "" backend: str = "daytona" # SandboxBackend value — determines client options sandbox_session_state: ( DaytonaSandboxSessionState | DockerSandboxSessionState | E2BSandboxSessionState | UnixLocalSandboxSessionState | None ) = None snapshot: SerializeAsAny[SnapshotBase] | None = ( None # serialized SnapshotBase for cross-backend creation ) previous_response_id: str | None = None history: list[dict[str, Any]] = [] # conversation history to seed (e.g. when forking) manifest: Manifest | None = None # per-session manifest override class AgentResponse(BaseModel): """Returned when the workflow is destroyed.""" pass class ToolCallRecord(BaseModel): """A single tool call with its input and output for TUI display.""" tool_name: str description: str arguments_json: str output: str | None = None requires_approval: bool = False approved: bool | None = None class ChatResponse(BaseModel): """Structured response from chat() replacing the plain string.""" text: str | None = None tool_calls: list[ToolCallRecord] = [] approval_request: ToolCallRecord | None = None class LiveToolCall(BaseModel): """A tool call visible to the TUI during an active turn.""" call_id: str tool_name: str arguments: str status: str = "pending" # pending | running | completed output: str | None = None class TurnState(BaseModel): """Everything the TUI needs — returned by a single query during polling.""" # idle | thinking | awaiting_approval | complete status: str = "idle" tool_calls: list[LiveToolCall] = [] response_text: str | None = None approval_request: ToolCallRecord | None = None turn_id: int = 0 # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _format_approval_item(item: ToolApprovalItem) -> str: """Return a human-readable summary of a tool approval request.""" raw = item.raw_item name = getattr(raw, "name", None) or item.tool_name or "unknown" # Try to extract arguments for shell commands args_str = getattr(raw, "arguments", None) if args_str and isinstance(args_str, str): try: parsed = json.loads(args_str) if name == "shell" and "commands" in parsed: cmds = parsed["commands"] return f"shell: {'; '.join(cmds)}" except (json.JSONDecodeError, TypeError): pass return f"{name}: {args_str or '(no args)'}" def _extract_text_from_items(items: list[RunItem]) -> str | None: """Pull the last assistant text from generated run items.""" for item in reversed(items): if isinstance(item, MessageOutputItem): raw = item.raw_item content = getattr(raw, "content", []) if isinstance(content, list): for block in content: text = getattr(block, "text", None) if isinstance(text, str): return text return None def _tool_call_records_from_items(items: list[RunItem]) -> list[ToolCallRecord]: """Build ToolCallRecord list from generated RunItems.""" records: list[ToolCallRecord] = [] for item in items: if isinstance(item, ToolCallItem): raw = item.raw_item name = getattr(raw, "name", None) or "unknown" args = getattr(raw, "arguments", "{}") records.append( ToolCallRecord( tool_name=name, description=f"{name}: {args}", arguments_json=args if isinstance(args, str) else json.dumps(args), ) ) return records # --------------------------------------------------------------------------- # Workflow definition # --------------------------------------------------------------------------- class _LiveStateHooks(RunHooksBase[Any, Agent[Any]]): """RunHooks that update workflow-queryable state for live TUI polling.""" def __init__(self, wf: AgentWorkflow) -> None: self._wf = wf async def on_llm_end(self, context, agent, response): """Extract tool calls from the model response and register them.""" for item in response.output: call_id = getattr(item, "call_id", None) if not call_id: continue # Standard function calls have name + arguments name = getattr(item, "name", None) if name: self._wf._live_tool_calls.append( LiveToolCall( call_id=call_id, tool_name=name, arguments=getattr(item, "arguments", None) or "{}", status="pending", ) ) continue # Shell tool calls have action.commands / action.command action = getattr(item, "action", None) if action: cmds = getattr(action, "commands", None) or getattr(action, "command", None) if isinstance(cmds, list): args = json.dumps({"commands": cmds}) elif isinstance(cmds, str): args = json.dumps({"command": cmds}) else: args = "{}" tool_name = getattr(item, "type", None) or "shell" self._wf._live_tool_calls.append( LiveToolCall( call_id=call_id, tool_name=tool_name, arguments=args, status="pending", ) ) async def on_tool_start(self, context, agent, tool): # Match first pending tool call (tools execute in order) for tc in self._wf._live_tool_calls: if tc.status == "pending": tc.status = "running" break async def on_tool_end(self, context, agent, tool, result): # Match first running tool call for tc in self._wf._live_tool_calls: if tc.status == "running": tc.status = "completed" tc.output = result[:4000] if result else None break @workflow.defn class AgentWorkflow: """A long-lived conversational agent workflow. The workflow persists indefinitely in Temporal, idling between TUI sessions. It only terminates when explicitly destroyed via the ``destroy`` signal (sent by the session manager). """ def __init__(self) -> None: self._pending_messages: list[str] = [] self._done = False self._conversation_history: list[dict[str, Any]] = [] self._sandbox_session_state: ( DaytonaSandboxSessionState | DockerSandboxSessionState | E2BSandboxSessionState | UnixLocalSandboxSessionState | None ) = None self._previous_response_id: str | None = None self._paused: bool = False self._pause_requested = False self._turn_tool_calls: list[ToolCallRecord] = [] self._manifest_override: Manifest | None = None self._backend: SandboxBackend = DEFAULT_BACKEND self._snapshot: SnapshotBase | None = None self._live_tool_calls: list[LiveToolCall] = [] # Turn state — queried by the TUI polling loop self._turn_status: str = "idle" self._turn_id: int = 0 self._last_response_text: str | None = None self._pending_approval: ToolCallRecord | None = None @workflow.query def is_paused(self) -> bool: return self._paused @workflow.signal async def send_message(self, msg: str) -> None: """Enqueue a user message. The TUI drives everything via get_turn_state polling.""" self._pending_messages.append(msg) self._conversation_history.append({"role": "user", "content": msg}) @workflow.query def get_history(self) -> list[dict[str, Any]]: """Return conversation history for TUI replay on reconnect.""" return self._conversation_history @workflow.query def get_snapshot_id(self) -> str | None: """Return just the current snapshot ID (lightweight).""" if self._sandbox_session_state: return self._sandbox_session_state.snapshot.id return None @workflow.query def get_snapshot(self) -> WorkflowSnapshot: """Return an atomic snapshot of run state and conversation history.""" # Prefer the live session snapshot, but fall back to self._snapshot # so workspace state survives a backend switch (which clears # _sandbox_session_state) until the next turn recreates a session. snapshot = self._snapshot if self._sandbox_session_state: snapshot = self._sandbox_session_state.snapshot return WorkflowSnapshot( sandbox_session_state=self._sandbox_session_state, snapshot=snapshot, previous_response_id=self._previous_response_id, history=self._conversation_history, ) @workflow.query def get_turn_state(self) -> TurnState: """Single query that returns everything the TUI needs.""" return TurnState( status=self._turn_status, tool_calls=list(self._live_tool_calls), response_text=self._last_response_text, approval_request=self._pending_approval, turn_id=self._turn_id, ) @workflow.update async def pause(self) -> None: """Request the workflow to pause.""" if self._paused: return self._pause_requested = True await workflow.wait_condition(lambda: self._paused) @workflow.update async def switch_backend(self, args: SwitchBackendSignal) -> None: """Switch to a different sandbox backend for subsequent turns. Clears the backend-specific session state so the next turn creates a fresh session on the new backend. The portable snapshot is preserved so the workspace filesystem can be carried over. """ match args.target: case "daytona": self._backend = SandboxBackend.DAYTONA self._manifest_override = Manifest(root="/home/daytona/workspace") case "docker": self._backend = SandboxBackend.DOCKER self._manifest_override = Manifest(root="/workspace") case "e2b": self._backend = SandboxBackend.E2B self._manifest_override = Manifest() # E2B resolves relative to sandbox home case SwitchToLocalBackend(workspace_root=root): self._backend = SandboxBackend.LOCAL self._manifest_override = Manifest(root=root) self._sandbox_session_state = None @workflow.signal async def destroy(self) -> None: """Terminate the workflow permanently.""" self._done = True def _resolve_sandbox_options( self, ) -> ( DaytonaSandboxClientOptions | DockerSandboxClientOptions | E2BSandboxClientOptions | UnixLocalSandboxClientOptions ): match self._backend: case SandboxBackend.DAYTONA: return DaytonaSandboxClientOptions(pause_on_exit=False) case SandboxBackend.DOCKER: return DockerSandboxClientOptions(image="python:3.14") case SandboxBackend.E2B: return E2BSandboxClientOptions(sandbox_type="e2b") case SandboxBackend.LOCAL: return UnixLocalSandboxClientOptions() def _resolve_manifest(self) -> Manifest: match self._backend: case SandboxBackend.DAYTONA: return Manifest(root="/home/daytona/workspace") case SandboxBackend.DOCKER: return Manifest(root="/workspace") case SandboxBackend.E2B: return Manifest() # E2B resolves workspace root relative to the sandbox home case SandboxBackend.LOCAL: return Manifest(root="/workspace") @workflow.run async def run(self, request: AgentRequest) -> AgentResponse: self._backend = SandboxBackend(request.backend) self._snapshot = request.snapshot if request.history: self._conversation_history = list(request.history) if request.sandbox_session_state: self._sandbox_session_state = request.sandbox_session_state if request.previous_response_id: self._previous_response_id = request.previous_response_id self._manifest_override = request.manifest while not self._done: await workflow.wait_condition( lambda: (len(self._pending_messages) > 0 or self._pause_requested or self._done), ) if self._pause_requested: # Let the caller (e.g. SessionManagerWorkflow.fork_session) know # no turn is in progress so it can safely snapshot state. self._paused = True self._pause_requested = False await workflow.wait_condition(lambda: len(self._pending_messages) > 0 or self._done) self._paused = False if self._done: break user_messages = list(self._pending_messages) self._pending_messages.clear() self._turn_id += 1 self._turn_status = "thinking" self._live_tool_calls = [] self._pending_approval = None self._last_response_text = None try: manifest = self._manifest_override or self._resolve_manifest() agent = self._build_agent(manifest) await self._run_turn(agent, user_messages) self._last_response_text = self._last_text if self._last_text: self._conversation_history.append( {"role": "assistant", "content": self._last_text} ) except Exception as e: self._last_response_text = f"Error: {e}" finally: self._turn_status = "complete" return AgentResponse() def _build_agent(self, manifest: Manifest, model: str = "gpt-5.6-sol") -> SandboxAgent: """Construct the SandboxAgent used by the workflow.""" return SandboxAgent( name="Temporal Sandbox Agent", model=model, instructions=( "You are a helpful coding assistant. Inspect the workspace and answer " "questions. Use the shell tool to run commands. " "Do not invent files or statuses that are not present in the workspace. " "Cite the file names you inspected." ), default_manifest=manifest, capabilities=[WorkspaceShellCapability()], model_settings=ModelSettings(tool_choice="auto"), ) async def _run_turn( self, agent: SandboxAgent, user_messages: list[str], ) -> None: self._turn_tool_calls = [] self._last_text: str | None = None hooks = _LiveStateHooks(self) # Always pass fresh input — previous_response_id gives the API # conversation context. Sandbox session state is carried via # run_config.sandbox.session_state to preserve the sandbox across turns. if len(user_messages) == 1: input_arg: str | list[TResponseInputItem] = user_messages[0] else: input_arg = [{"role": "user", "content": m} for m in user_messages] run_config = RunConfig( sandbox=SandboxRunConfig( client=temporal_sandbox_client(self._backend.value), options=self._resolve_sandbox_options(), # Restore sandbox session state from the previous turn if available. session_state=self._sandbox_session_state, snapshot=self._snapshot, ), workflow_name="Temporal Sandbox workflow", ) # Run the agent -- loops internally handling tool calls result = await Runner.run( agent, input_arg, run_config=run_config, hooks=hooks, previous_response_id=self._previous_response_id, ) # Extract results self._turn_tool_calls.extend(_tool_call_records_from_items(result.new_items)) self._last_text = _extract_text_from_items(result.new_items) # Track response ID for conversation continuity and save state # to preserve sandbox session across turns. self._previous_response_id = result.last_response_id # Persist sandbox session state for the next turn. try: state = result.to_state() sandbox_data = state.to_json().get("sandbox", {}) session_state_data = sandbox_data.get("session_state") if session_state_data: self._sandbox_session_state = cast( DaytonaSandboxSessionState | UnixLocalSandboxSessionState, SandboxSessionState.parse(session_state_data), ) # Keep the portable snapshot up to date so it can seed a # fresh session after a backend switch. self._snapshot = self._sandbox_session_state.snapshot except Exception: pass # --------------------------------------------------------------------------- # Worker entrypoint # --------------------------------------------------------------------------- async def run_worker() -> None: # Imported here to avoid unnecessary passthroughs in the workflow sandbox. import docker # type: ignore[import-untyped] from _worker_setup import print_backend_warnings # type: ignore[import-not-found] from temporal_session_manager import ( # type: ignore[import-not-found] SessionManagerWorkflow, pause_workflow, query_workflow_snapshot, switch_workflow_backend, ) from temporalio.contrib.openai_agents import ( ModelActivityParameters, OpenAIAgentsPlugin, SandboxClientProvider, ) from agents.extensions.sandbox import DaytonaSandboxClient, E2BSandboxClient from agents.sandbox.sandboxes import DockerSandboxClient, UnixLocalSandboxClient sandbox_clients: list[SandboxClientProvider] = [ SandboxClientProvider("local", UnixLocalSandboxClient()), ] if _os.environ.get("DAYTONA_API_KEY"): sandbox_clients.append(SandboxClientProvider("daytona", DaytonaSandboxClient())) if _os.environ.get("E2B_API_KEY"): sandbox_clients.append(SandboxClientProvider("e2b", E2BSandboxClient())) try: sandbox_clients.append( SandboxClientProvider("docker", DockerSandboxClient(docker.from_env())) ) except docker.errors.DockerException: pass plugin = OpenAIAgentsPlugin( model_params=ModelActivityParameters( start_to_close_timeout=timedelta(seconds=120), ), sandbox_clients=sandbox_clients, ) temporal_client = await Client.connect("localhost:7233", plugins=[plugin]) worker = Worker( temporal_client, task_queue=TASK_QUEUE, workflows=[AgentWorkflow, SessionManagerWorkflow], activities=[pause_workflow, query_workflow_snapshot, switch_workflow_backend], workflow_runner=SandboxedWorkflowRunner( restrictions=SandboxRestrictions.default.with_passthrough_modules( "pydantic_core", ), ), ) print_backend_warnings({p.name for p in sandbox_clients}) print(f"Worker started on task queue '{TASK_QUEUE}'. Press Ctrl-C to stop.") await worker.run() # --------------------------------------------------------------------------- # CLI entrypoints # --------------------------------------------------------------------------- async def run_conversation() -> None: """Start the TUI -- sessions are managed entirely via Temporal.""" from temporal_sandbox_tui import ConversationApp # type: ignore[import-not-found] app = ConversationApp( workflow_cls=AgentWorkflow, task_queue=TASK_QUEUE, cwd=str(Path.cwd()), ) await app.run_async() # --------------------------------------------------------------------------- # Argument parsing # --------------------------------------------------------------------------- def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Run the Sandbox agent as a multi-turn Temporal workflow." ) sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("worker", help="Start the Temporal worker process.") sub.add_parser("run", help="Start an interactive agent conversation.") return parser.parse_args() if __name__ == "__main__": args = parse_args() if args.command == "worker": asyncio.run(run_worker()) else: asyncio.run(run_conversation())