temporalio.contrib.google_genai.testing.GeminiTestServer

class documentation

class GeminiTestServer: (source)

Constructor: GeminiTestServer(responses)

View In Hierarchy

Scripts Gemini model responses so workflows run without real API calls.

Pass canned response bodies built with text_response / function_call_response. Each model call — including each turn of an automatic-function-calling loop and each generate_content_stream call — consumes the next response in order. Build a plugin with plugin and register it on your worker; inspect requests afterwards to assert exactly what the integration sent.

Only model calls (client.models) are scripted. File, interaction, and agent operations are not; mock those on a genai.Client directly if a test needs them. MCP servers can be registered with plugin(mcp_servers=...) and run for real.

Method __init__ Initialize with the response bodies to serve, in order.
Method plugin Return a GoogleGenAIPlugin whose model calls serve the script.
Instance Variable requests Undocumented
Method _next Undocumented
Instance Variable _index Undocumented
Instance Variable _responses Undocumented
def __init__(self, responses: Sequence[str]): (source)

Initialize with the response bodies to serve, in order.

def plugin(self, *, mcp_servers: dict[str, McpSessionFactory] | None = None, mcp_connection_idle_timeout: timedelta | None = None) -> GoogleGenAIPlugin: (source)

Return a GoogleGenAIPlugin whose model calls serve the script.

The real plugin activities run; only the underlying HTTP layer is replaced, so request formatting and the AFC loop are exercised exactly as in production.

Parameters
mcp_servers:dict[str, McpSessionFactory] | NoneMCP servers to expose to workflows, as on temporalio.contrib.google_genai.GoogleGenAIPlugin. These are not scripted — list_tools / call_tool run for real against the given sessions — so a test can drive an MCP tool call with a scripted function_call_response and assert on what the server actually returned.
mcp_connection_idle_timeout:timedelta | NoneHow long an idle worker-side MCP connection stays open, as on temporalio.contrib.google_genai.GoogleGenAIPlugin.
Returns
GoogleGenAIPluginUndocumented
requests: list[dict[str, Any]] = (source)

Undocumented

def _next(self) -> str: (source)

Undocumented

_index: int = (source)

Undocumented

_responses = (source)

Undocumented