temporalio.contrib.google_genai.TemporalAsyncClient

class documentation

class TemporalAsyncClient(AsyncClient): (source)

Constructor: TemporalAsyncClient(vertexai, project, location, activity_config, ...)

An AsyncClient whose API calls run as Temporal activities.

Warning

This API is experimental and may change in future versions. Use with caution in production environments.

Builds a private BaseApiClient that dispatches HTTP calls through workflow.execute_activity and wraps it, so the SDK's request-formatting code (including the AFC loop) runs in the workflow while only the actual API calls cross into activities. Credentials are never fetched or stored in the workflow — the activity worker handles authentication independently.

AsyncFiles and AsyncFileSearchStores are replaced with shims that run upload/download as activities; interactions and agents — which bypass BaseApiClient via a vendored HTTP client — are likewise replaced with activity-backed shims; webhooks is not supported in workflows and raises. Other modules (models, tunings, caches, batches, live, tokens, operations) are inherited unchanged and work through the private api client's activity-backed HTTP methods.

Construct it from within a workflow run method:

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, query: str) -> str:
        client = TemporalAsyncClient()
        response = await client.models.generate_content(
            model="gemini-2.0-flash",
            contents=query,
            config=GenerateContentConfig(
                tools=[
                    activity_as_tool(
                        my_tool,
                        activity_config=ActivityConfig(
                            start_to_close_timeout=timedelta(seconds=30),
                        ),
                    ),
                ],
            ),
        )
        return response.text

Method	`__init__`	Initialize a Temporal-aware client.
Property	`agents`	Temporal-aware agents resource; operations run as activities.
Property	`interactions`	Temporal-aware interactions resource; operations run as activities.
Property	`webhooks`	Raise — webhooks are not supported in Temporal workflows.
Instance Variable	`_file_search_stores`	Undocumented
Instance Variable	`_files`	Undocumented
Instance Variable	`_models`	Undocumented
Instance Variable	`_temporal_agents`	Undocumented
Instance Variable	`_temporal_interactions`	Undocumented

def __init__(self, *, vertexai: bool = False, project: str | None = None, location: str | None = None, activity_config: ActivityConfig | None = None, streaming_topic: str | None = None, streaming_batch_interval: timedelta = timedelta(milliseconds=100)): (source) ¶

Initialize a Temporal-aware client.

Parameters
vertexai:`bool`	Whether to use Vertex AI API endpoints. Must match the `GoogleGenAIPlugin` configuration on the worker side. Defaults to `False` (Gemini Developer API).
project:`str \| None`	Google Cloud project ID. Only needed when `vertexai=True` and the SDK's request formatting requires it (e.g., cache operations).
location:`str \| None`	Google Cloud location. Same conditions as `project`.
activity_config:`ActivityConfig \| None`	Override the default activity configuration (timeouts, retry policy, etc.) for Gemini API call activities. When not provided, every operation (model calls, files, interactions, managed agents) defaults to a 60-second `start_to_close_timeout` and Temporal's default retry policy.
streaming_topic:`str \| None`	When set, `generate_content_stream` publishes each streamed `GenerateContentResponse` to this `temporalio.contrib.workflow_streams.WorkflowStream` topic as it arrives, so external consumers can observe the model output in real time. The workflow must construct a `WorkflowStream` in `@workflow.init`; otherwise the call raises. The workflow's own stream iteration is unchanged.
streaming_batch_interval:`timedelta`	How often the streaming activity flushes published chunks to the workflow stream. Defaults to 100ms.