temporalio.contrib.google_genai.TemporalMcpClientSession

class documentation

class TemporalMcpClientSession(ClientSession): (source)

Constructor: TemporalMcpClientSession(server_name, cache_tools, activity_config)

View In Hierarchy

mcp.ClientSession whose tool discovery and calls run as activities.

Warning

This API is experimental and may change in future versions.

Construct inside a workflow and pass it in the tools list of a generate_content call. The matching server name must be registered on the worker via GoogleGenAIPlugin(mcp_servers={name: factory}).

cache_tools controls how often tools are listed. When False (the default) the {server}-list-tools activity runs each time the SDK discovers tools (i.e. per generate_content call), so a server whose tools changed mid-workflow is picked up. When True the first listing is cached on this instance and reused for its lifetime (replay-safe in-workflow state).

Parameters
server_nameName selecting the worker-side factory; also the activity prefix ({server_name}-list-tools / {server_name}-call-tool).
cache_toolsCache the tool listing after the first call.
activity_configActivity configuration (timeouts, retry policy, etc.) for the MCP activities. Defaults to a 60-second start_to_close_timeout.
Method __init__ Initialize without calling super (no real streams exist here).
Async Method call_tool Call a tool via the {server}-call-tool activity.
Async Method list_tools List the server's tools via the {server}-list-tools activity.
Method _config Undocumented
Instance Variable _activity_config Undocumented
Instance Variable _cache_tools Undocumented
Instance Variable _cached_tools Undocumented
Instance Variable _server_name Undocumented
def __init__(self, server_name: str, *, cache_tools: bool = False, activity_config: ActivityConfig | None = None): (source)

Initialize without calling super (no real streams exist here).

async def call_tool(self, name: str, arguments: dict[str, Any] | None = None, read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, *, meta: dict[str, Any] | None = None) -> CallToolResult: (source)

Call a tool via the {server}-call-tool activity.

async def list_tools(self, cursor: str | None = None, *, params: PaginatedRequestParams | None = None) -> ListToolsResult: (source)

List the server's tools via the {server}-list-tools activity.

def _config(self, summary: str) -> ActivityConfig: (source)

Undocumented

_activity_config: ActivityConfig = (source)

Undocumented

_cache_tools = (source)

Undocumented

_cached_tools: ListToolsResult | None = (source)

Undocumented

_server_name = (source)

Undocumented