temporalio.contrib.google_adk

module documentation

(source)

Workflow utilities for Google ADK agents integration with Temporal.

Class	`ToolContextSnapshot`	Serializable snapshot of the ADK `ToolContext` for activity-backed tools.
Function	`activity_as_tool`	Decorator/Wrapper to wrap a Temporal Activity as an ADK Tool.
Function	`activity_node`	Wraps a Temporal Activity as a node for an ADK workflow graph.
Function	`_adk_context_error`	Builds the error for a parameter annotated with a live ADK context.
Function	`_annotation_display`	Renders annotation members for error messages.
Function	`_annotation_members`	Returns a union annotation's members, or the annotation itself.
Function	`_bind_node_input`	Maps a dict node input onto the activity's positional parameters by name.
Function	`_snapshot_tool_context`	Builds the serializable snapshot from a live ADK `ToolContext`.
Function	`_tool_context_parameter`	Validates context-related parameters and returns the `tool_context` one.
Function	`_validated_tool_context_parameter`	Returns the `tool_context` parameter after validating its annotation.
Constant	`_TOOL_CONTEXT_PARAM`	Undocumented

def activity_as_tool(activity_def: Callable, **kwargs: Any) -> Callable: (source) ¶

Decorator/Wrapper to wrap a Temporal Activity as an ADK Tool.

Warning

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

This ensures the activity's signature is preserved for ADK's tool schema generation while marking it as a tool that executes via 'workflow.execute_activity'.

If the activity declares a parameter named tool_context, it must be annotated with ToolContextSnapshot (or ToolContextSnapshot | None). ADK excludes the parameter from the tool schema and injects the live ToolContext into the wrapper, which passes the activity a serializable snapshot of it (session state and function-call id) in that parameter's position. Annotating any parameter with a live ADK context type raises ValueError at wrap time, since ADK would inject the non-serializable context into it regardless of its name.

def activity_node(activity_def: Callable, *, name: str | None = None, rerun_on_resume: bool = False, **kwargs: Any) -> FunctionNode: (source) ¶

Wraps a Temporal Activity as a node for an ADK workflow graph.

Warning

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

The returned google.adk.workflow.FunctionNode can be placed in a Workflow(edges=[...]) graph or invoked from a dynamic node via ctx.run_node(...). The previous node's output (node_input) is passed to the activity: directly for a single-parameter activity, or bound by parameter name for a multi-parameter activity (node_input must then be a dict). Outside a workflow (local ADK runs) the activity function is invoked directly.

Parameters
activity_def:`Callable`	The `@activity.defn` function to run.
name:`str \| None`	Node name; defaults to the activity function's name.
rerun_on_resume:`bool`	Passed through to `FunctionNode`. Keep the default `False` so that on a human-in-the-loop resume the node is fast-forwarded from the session instead of re-executing the activity.
**kwargs:`Any`	Activity execution options for `workflow.execute_activity` (e.g. `start_to_close_timeout`). Prefer configuring retries here via `retry_policy` rather than wrapping the node with an ADK `RetryConfig`, which would retry on top of Temporal's own activity retries.

Returns
`FunctionNode`	Undocumented

def _adk_context_error(activity_def: Callable, name: str, annotation_name: str) -> ValueError: (source) ¶

Builds the error for a parameter annotated with a live ADK context.

def _annotation_display(members: tuple[Any, ...]) -> str: (source) ¶

Renders annotation members for error messages.

def _annotation_members(annotation: Any) -> tuple[Any, ...]: (source) ¶

Returns a union annotation's members, or the annotation itself.

def _bind_node_input(activity_def: Callable, params: list[inspect.Parameter], node_input: dict[str, Any]) -> list[Any]: (source) ¶

Maps a dict node input onto the activity's positional parameters by name.

def _snapshot_tool_context(tool_context: Any) -> ToolContextSnapshot: (source) ¶

Builds the serializable snapshot from a live ADK ToolContext.

def _tool_context_parameter(activity_def: Callable) -> inspect.Parameter | None: (source) ¶

Validates context-related parameters and returns the tool_context one.

ADK injects the live context into the first parameter annotated with an ADK context type (regardless of name), or failing that into one named tool_context, and excludes that parameter from the LLM-facing tool schema. Live context objects cannot cross the activity boundary, so the only supported declaration is a parameter named tool_context annotated with ToolContextSnapshot (or ToolContextSnapshot | None); anything else ADK would treat as a context parameter is rejected at wrap time, as is a misplaced ToolContextSnapshot annotation that would leak into the tool schema.

def _validated_tool_context_parameter(activity_def: Callable, parameter: inspect.Parameter, members: tuple[Any, ...]) -> inspect.Parameter: (source) ¶

Returns the tool_context parameter after validating its annotation.

The parameter must be annotated with ToolContextSnapshot or ToolContextSnapshot | None.

_TOOL_CONTEXT_PARAM: str = (source) ¶

Undocumented

Value

'tool_context'

temporalio.contrib.google_adk_agents.workflow

`temporalio.contrib.google_adk_agents.workflow`