temporalio.contrib.google_genai._temporal_interactions.TemporalAsyncInteractions

class documentation

class TemporalAsyncInteractions: (source)

Constructor: TemporalAsyncInteractions(activity_config)

View In Hierarchy

Interactions resource shim that routes calls through activities.

Methods accept the same keyword arguments as the real resource and forward them verbatim — the SDK validates them on the worker side, so a bad argument surfaces as an activity failure (retried per the activity's retry policy) rather than a workflow-side error.

with_raw_response / with_streaming_response are not supported in workflows.

Method __init__ Initialize with activity config for interaction operation timeouts.
Async Method cancel Cancel an interaction via a Temporal activity.
Async Method create Create an interaction via a Temporal activity.
Async Method delete Delete an interaction via a Temporal activity.
Async Method get Get an interaction via a Temporal activity.
Property with_raw_response Raise — raw responses are not available in workflows.
Property with_streaming_response Raise — streaming responses are not available in workflows.
Method _config Undocumented
Instance Variable _activity_config Undocumented
def __init__(self, activity_config: ActivityConfig | None = None): (source)

Initialize with activity config for interaction operation timeouts.

async def cancel(self, id: str, **kwargs: Any) -> Interaction: (source)

Cancel an interaction via a Temporal activity.

async def create(self, *, stream: bool = False, **kwargs: Any) -> Interaction | _TemporalInteractionAsyncStream: (source)

Create an interaction via a Temporal activity.

kwargs is forwarded verbatim to client.aio.interactions.create on the worker. With stream=True the activity drains the SSE stream and returns all events batched; the returned object supports async for / async with like the SDK's streaming response.

async def delete(self, id: str, **kwargs: Any) -> object: (source)

Delete an interaction via a Temporal activity.

async def get(self, id: str, *, stream: bool = False, **kwargs: Any) -> Interaction | _TemporalInteractionAsyncStream: (source)

Get an interaction via a Temporal activity.

Supports stream=True (with the SDK's last_event_id kwarg for resumption); events come back batched like create.

with_raw_response: Any = (source)

Raise — raw responses are not available in workflows.

with_streaming_response: Any = (source)

Raise — streaming responses are not available in workflows.

def _config(self, summary: str, params: dict[str, Any]) -> ActivityConfig: (source)

Undocumented

_activity_config = (source)

Undocumented