temporalio.contrib.openai_agents.sandbox._temporal_sandbox_session.TemporalSandboxSession

class documentation

class TemporalSandboxSession(BaseSandboxSession): (source)

Constructor: TemporalSandboxSession(name, config, state, supports_pty_flag)

View In Hierarchy

A BaseSandboxSession that routes all I/O through Temporal activities.

This class is fully stateless with respect to the physical sandbox -- it holds only the serializable SandboxSessionState and a supports_pty flag (both provided by the worker-side SessionResult).

Activity names are prefixed with the provider name so that dispatches reach the correct sandbox backend's activities on the worker.

Each activity receives a single Pydantic model instance. Because the Temporal client is configured with pydantic_data_converter, all fields are serialized and deserialized automatically.

Method __init__ Initialize the session.
Async Method exec Execute a command in the sandbox via activity.
Async Method hydrate_workspace Hydrate the workspace via activity.
Async Method persist_workspace Persist the workspace via activity.
Async Method pty_exec_start Start a PTY exec via activity.
Async Method pty_write_stdin Write to PTY stdin via activity.
Async Method read Read a file from the sandbox via activity.
Async Method running Check if the sandbox is running via activity.
Async Method shutdown Shut down the sandbox via activity.
Async Method start Start the sandbox session via activity.
Method state.setter Undocumented
Async Method stop Stop the sandbox session via activity.
Method supports_pty Whether this session supports PTY operations.
Async Method write Write a file to the sandbox via activity.
Property state The current session state.
Async Method _exec_internal Undocumented
Instance Variable _config Undocumented
Instance Variable _name Undocumented
Instance Variable _state Undocumented
Instance Variable _supports_pty Undocumented
def __init__(self, name: str, config: ActivityConfig, state: SandboxSessionState, supports_pty_flag: bool = True): (source)

Initialize the session.

async def exec(self, *command: str | Path, timeout: float | None = None, shell: bool | list[str] = True, user: str | User | None = None) -> ExecResult: (source)

Execute a command in the sandbox via activity.

async def hydrate_workspace(self, data: io.IOBase): (source)

Hydrate the workspace via activity.

async def persist_workspace(self) -> io.IOBase: (source)

Persist the workspace via activity.

async def pty_exec_start(self, *command: str | Path, timeout: float | None = None, shell: bool | list[str] = True, user: str | User | None = None, tty: bool = False, yield_time_s: float | None = None, max_output_tokens: int | None = None) -> PtyExecUpdate: (source)

Start a PTY exec via activity.

async def pty_write_stdin(self, *, session_id: int, chars: str, yield_time_s: float | None = None, max_output_tokens: int | None = None) -> PtyExecUpdate: (source)

Write to PTY stdin via activity.

async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase: (source)

Read a file from the sandbox via activity.

async def running(self) -> bool: (source)

Check if the sandbox is running via activity.

async def shutdown(self): (source)

Shut down the sandbox via activity.

async def start(self): (source)

Start the sandbox session via activity.

def state(self, value: SandboxSessionState): (source)

Undocumented

async def stop(self): (source)

Stop the sandbox session via activity.

def supports_pty(self) -> bool: (source)

Whether this session supports PTY operations.

async def write(self, path: Path, data: io.IOBase, *, user: str | User | None = None): (source)

Write a file to the sandbox via activity.

state: SandboxSessionState = (source)

The current session state.

async def _exec_internal(self, *command: str | Path, timeout: float | None = None) -> ExecResult: (source)

Undocumented

_config = (source)

Undocumented

_name = (source)

Undocumented

_state = (source)

Undocumented

_supports_pty = (source)

Undocumented