temporalio.contrib.strands.TemporalSandbox

class documentation

class TemporalSandbox(Sandbox): (source)

Constructor: TemporalSandbox(name, task_queue, schedule_to_close_timeout, schedule_to_start_timeout, ...)

View In Hierarchy

Workflow-side sandbox that dispatches operations as Temporal activities.

Method __init__ Configure a registered sandbox name and its activity options.
Async Method execute_code_streaming Execute code in the registered worker-side sandbox.
Async Method execute_streaming Execute a command in the registered worker-side sandbox.
Method get_tools Vend Strands' standard shell and file-editor sandbox tools.
Async Method list_files List a directory in the registered worker-side sandbox.
Async Method read_file Read bytes from the registered worker-side sandbox.
Async Method remove_file Remove a file from the registered worker-side sandbox.
Async Method write_file Write bytes to the registered worker-side sandbox.
Async Method _execute Undocumented
Instance Variable _name Undocumented
Instance Variable _options Undocumented
Instance Variable _streaming_batch_interval Undocumented
Instance Variable _streaming_topic Undocumented
def __init__(self, name: str, *, task_queue: str | None = None, schedule_to_close_timeout: timedelta | None = None, schedule_to_start_timeout: timedelta | None = None, start_to_close_timeout: timedelta | None = None, heartbeat_timeout: timedelta | None = None, retry_policy: RetryPolicy | None = None, cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL, versioning_intent: VersioningIntent | None = None, summary: str | None = None, priority: Priority = Priority.default, streaming_topic: str | None = None, streaming_batch_interval: timedelta = timedelta(milliseconds=100)): (source)

Configure a registered sandbox name and its activity options.

async def execute_code_streaming(self, code: str, language: str, *, timeout: float | None = None, cwd: str | None = None, env: dict[str, str] | None = None, **kwargs: Any) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: (source)

Execute code in the registered worker-side sandbox.

async def execute_streaming(self, command: str, *, timeout: float | None = None, cwd: str | None = None, env: dict[str, str] | None = None, **kwargs: Any) -> AsyncGenerator[StreamChunk | ExecutionResult, None]: (source)

Execute a command in the registered worker-side sandbox.

def get_tools(self) -> list[AgentTool]: (source)

Vend Strands' standard shell and file-editor sandbox tools.

async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]: (source)

List a directory in the registered worker-side sandbox.

async def read_file(self, path: str, **kwargs: Any) -> bytes: (source)

Read bytes from the registered worker-side sandbox.

async def remove_file(self, path: str, **kwargs: Any): (source)

Remove a file from the registered worker-side sandbox.

async def write_file(self, path: str, content: bytes, **kwargs: Any): (source)

Write bytes to the registered worker-side sandbox.

async def _execute(self, operation: str, input: Any, *, result_type: type | None = None) -> Any: (source)

Undocumented

_name = (source)

Undocumented

_options: dict[str, Any] = (source)

Undocumented

_streaming_batch_interval = (source)

Undocumented

_streaming_topic = (source)

Undocumented