temporalio.contrib.google_genai.testing

module documentation
(source)

Testing utilities for the Google Gemini SDK Temporal integration.

These let you exercise workflows that use temporalio.contrib.google_genai.TemporalAsyncClient without making real Gemini API calls. Script the model's responses with text_response / function_call_response, build a plugin with GeminiTestServer, and register it on your worker like the real temporalio.contrib.google_genai.GoogleGenAIPlugin.

Example:

server = GeminiTestServer(
    [
        function_call_response("get_weather", {"city": "Tokyo"}),
        text_response("It's sunny in Tokyo."),
    ]
)
async with Worker(
    client,
    task_queue="test",
    workflows=[MyAgentWorkflow],
    activities=[get_weather],
    plugins=[server.plugin()],
):
    ...
assert len(server.requests) == 2  # one per model turn
Class GeminiTestServer Scripts Gemini model responses so workflows run without real API calls.
Function function_call_response Build a generate_content response body with a single function call.
Function text_response Build a generate_content response body with a single text part.
def function_call_response(name: str, args: dict[str, Any]) -> str: (source)

Build a generate_content response body with a single function call.

The Gemini SDK's automatic function calling loop will invoke the matching tool, then request another response — so pair each function-call response with a following text_response (or further calls).

def text_response(text: str) -> str: (source)

Build a generate_content response body with a single text part.