temporalio.contrib.google_adk_agents.TemporalStatefulMcpToolSetProvider

class documentation

class TemporalStatefulMcpToolSetProvider: (source)

Constructor: TemporalStatefulMcpToolSetProvider(name, toolset_factory)

View In Hierarchy

Provider for a stateful, pooled MCP toolset backed by a dedicated worker.

Warning

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

Unlike TemporalMcpToolSetProvider (which builds and closes a fresh McpToolset per activity call), this provider maintains a single McpToolset for the lifetime of a workflow run. The workflow-side TemporalStatefulMcpToolSet starts a dedicated -server-session activity on a task queue scoped to that specific run (name@run_id); that activity builds the toolset once via toolset_factory(factory_argument), holds it open, and runs a nested temporalio.worker.Worker serving the -list-tools/-call-tool activities off the same run-scoped queue. The toolset is closed when the workflow cancels the session activity on cleanup.

Because state lives in a dedicated worker's memory, the caller must handle the case where that worker fails, as Temporal cannot seamlessly recreate the lost toolset. Failure surfaces as an ApplicationError with type="DedicatedWorkerFailure". Prefer the stateless TemporalMcpToolSetProvider unless a persistent connection is genuinely required.

Method __init__ Initializes the stateful toolset provider.
Property name The activity-name prefix (base name with the -stateful suffix).
Method _get_activities Undocumented
Instance Variable _name Undocumented
Instance Variable _toolset_factory Undocumented
Instance Variable _toolsets Undocumented
def __init__(self, name: str, toolset_factory: Callable[[(Any | None)], McpToolset]): (source)

Initializes the stateful toolset provider.

Parameters
name:strName prefix for the generated activities. It is suffixed with -stateful so it never collides with a stateless provider of the same base name registered on the same worker.
toolset_factory:Callable[[(Any | None)], McpToolset]Factory function that creates McpToolset instances. It should return a new toolset each time so that no state is shared between workflow runs.
name: str = (source)

The activity-name prefix (base name with the -stateful suffix).

def _get_activities(self) -> Sequence[Callable]: (source)

Undocumented

_name = (source)

Undocumented

_toolset_factory = (source)

Undocumented

_toolsets: dict[str, McpToolset] = (source)

Undocumented