temporalio.contrib.google_genai._temporal_interactions

module documentation
(source)

Temporal-aware interactions resource shim.

TemporalAsyncInteractions exposes the same surface as google-genai's AsyncClient.interactions resource, but each operation is dispatched through a Temporal activity holding the real genai.Client on the worker. The Interactions API does not go through BaseApiClient — it uses a vendored, Stainless-generated HTTP client that the TemporalApiClient shim never sees — so each operation is routed as a whole through an activity instead.

The shim depends only on the public google.genai.interactions surface (types plus client.aio.interactions on the worker), not on google-genai internals, so it is unaffected by regeneration of the vendored client.

Class TemporalAsyncInteractions Interactions resource shim that routes calls through activities.
Class _TemporalInteractionAsyncStream Async stream over interaction events already drained in an activity.
Function _deserialize Rehydrate a dict returned by an activity into its public genai model.
Function _pop_timeout Pop a per-call timeout kwarg and apply it to the activity config.
Constant _DEFAULT_INTERACTION_TIMEOUT Undocumented
Constant _SSE_EVENT_ADAPTER Undocumented
def _deserialize(value: dict[str, Any], type_: Any) -> Any: (source)

Rehydrate a dict returned by an activity into its public genai model.

InteractionSSEEvent is deserialized through its TypeAdapter, which dispatches the discriminated union (and nested unions) and tolerates the sparse nested payloads the API legitimately emits (e.g. an interaction.created event carrying an Interaction with just id and object). Plain models use model_validate, which recurses nested models (e.g. AgentListResponse.agents into Agent) and resolves aliases; the SDK's optional fields keep it tolerant of the minimal objects the API returns. Both paths are pure functions, safe to run in the workflow on every replay.

def _pop_timeout(params: dict[str, Any], config: ActivityConfig): (source)

Pop a per-call timeout kwarg and apply it to the activity config.

The Interactions API expresses timeouts in seconds. Temporal owns timeouts/retries, so the value maps to start_to_close_timeout rather than being forwarded to the underlying HTTP client.

_DEFAULT_INTERACTION_TIMEOUT = (source)

Undocumented

Value
timedelta(seconds=60)
_SSE_EVENT_ADAPTER: pydantic.TypeAdapter[Any] = (source)

Undocumented

Value
pydantic.TypeAdapter(InteractionSSEEvent)