temporalio.contrib.openai_agents.sandbox._temporal_sandbox_client.TemporalSandboxClient

class documentation

class TemporalSandboxClient(BaseSandboxClient[BaseSandboxClientOptions]): (source)

Constructor: TemporalSandboxClient(name, config)

View In Hierarchy

Stateless client that dispatches all lifecycle operations as Temporal activities.

No inner client is needed -- session creation, resumption, and deletion are all handled by activities whose names are prefixed with the provider name (e.g. "daytona-sandbox_create_session"). The real BaseSandboxClient lives inside SandboxClientProvider on the worker.

Users should never need to instantiate this directly -- use temporalio.contrib.openai_agents.workflow.temporal_sandbox_client instead.

Parameters
nameThe name of the SandboxClientProvider registered on the worker. Used as an activity-name prefix so that the correct sandbox backend is targeted.
configOptional activity configuration for controlling timeouts, retries, etc. Defaults to a 5-minute start_to_close_timeout.
Method __init__ Initialize the client.
Async Method create Create a new sandbox session via activity.
Async Method delete Delete a sandbox session via activity.
Method deserialize_session_state Deserialize a session state from a dict.
Async Method resume Resume an existing sandbox session via activity.
Instance Variable backend_id Undocumented
Instance Variable _config Undocumented
Instance Variable _name Undocumented
def __init__(self, name: str, config: ActivityConfig | None = None): (source)

Initialize the client.

async def create(self, *, snapshot: SnapshotSpec | SnapshotBase | None = None, manifest: Manifest | None = None, options: BaseSandboxClientOptions) -> SandboxSession: (source)

Create a new sandbox session via activity.

async def delete(self, session: TemporalSandboxSession) -> TemporalSandboxSession: (source)

Delete a sandbox session via activity.

def deserialize_session_state(self, payload: dict[str, Any]) -> SandboxSessionState: (source)

Deserialize a session state from a dict.

async def resume(self, state: SandboxSessionState) -> SandboxSession: (source)

Resume an existing sandbox session via activity.

backend_id = (source)

Undocumented

_config: ActivityConfig = (source)

Undocumented

_name = (source)

Undocumented