google_adk_agents: graph workflows, dynamic workflows, and durable HITL (ADK v2)

pyproject.toml

@@ -29,7 +29,7 @@ grpc = ["grpcio>=1.48.2,<2"]
 opentelemetry = ["opentelemetry-api>=1.11.1,<2", "opentelemetry-sdk>=1.11.1,<2"]
 pydantic = ["pydantic>=2.0.0,<3"]
 openai-agents = ["openai-agents>=0.17.5", "mcp>=1.9.4, <2"]
-google-adk = ["google-adk>=2.2.0,<3"]
+google-adk = ["google-adk>=2.5.0,<3"]
 langgraph = ["langgraph>=1.1.0"]
 langsmith = ["langsmith>=0.7.34,<0.9"]
 lambda-worker-otel = [
@@ -267,3 +267,5 @@ exclude = ["temporalio/bridge/target/**/*", "temporalio/bridge/sdk-core/.git"]
 # Prevent uv commands from building the package by default
 package = false
 exclude-newer = "2 weeks"
+# TODO: remove once google-adk 2.5.0 (published 2026-07-16) passes the cooldown
+exclude-newer-package = { google-adk = false }

temporalio/contrib/google_adk_agents/README.md

@@ -195,6 +195,163 @@ agent = Agent(
 )
 ```
 
+## Graph Workflows (ADK v2)
+
+ADK v2's graph runtime (`google.adk.workflow`) runs inside Temporal workflows:
+the scheduler is pure asyncio and executes deterministically on Temporal's
+workflow event loop, while LLM calls (`TemporalModel`), MCP tools
+(`TemporalMcpToolSet`), and activity-backed nodes leave the workflow as
+activities.
+
+Use `activity_node(...)` to run a graph node as a Temporal activity. The
+previous node's output is passed to the activity — directly for a
+single-parameter activity, bound by name (from a dict) for multi-parameter
+activities:
+
+```python
+from google.adk.workflow import JoinNode, Workflow
+from temporalio.contrib.google_adk_agents.workflow import activity_node
+
+fetch = activity_node(fetch_data, start_to_close_timeout=timedelta(seconds=30))
+
+def summarize(node_input):  # plain nodes run in-workflow: keep deterministic
+    return f"{node_input} summarized"
+
+graph = Workflow(name="pipeline", edges=[("START", fetch, summarize)])
+```
+
+Conditional routing (`(router, {"KEY": handler, ...})`, `DEFAULT_ROUTE`),
+parallel fan-out with `JoinNode`, and `LlmAgent` nodes (with
+`mode="task"`/`"single_turn"`) all work — agent nodes route their model calls
+through `TemporalModel` as usual.
+
+## Dynamic Workflows
+
+Dynamic nodes (`await ctx.run_node(...)` with loops, branches, and
+`asyncio.gather`) work in-workflow; child-run caching reads only the
+in-memory session, so re-entry after a HITL resume replays deterministically.
+
+```python
+from google.adk.workflow import node
+
+@node(rerun_on_resume=True)
+async def pipeline(ctx):
+    data = await ctx.run_node(fetch, "query")          # activity_node child
+    results = await asyncio.gather(
+        *(ctx.run_node(worker, item) for item in data)  # parallel children
+    )
+    return results
+```
+
+On a HITL resume, a `rerun_on_resume=True` dynamic node re-executes its body
+while completed children are skipped from the session cache. Place activity
+invocations in child nodes (`activity_node`, `activity_tool`) rather than
+inline in the dynamic node body, or make them idempotent — inline calls run
+again on re-entry (ADK's documented at-least-once semantics).
+
+A `Workflow` with an `input_schema` can also be passed in an agent's
+`tools=[...]` list (Workflow-as-Tool), letting the model invoke whole graphs
+as tools.
+
+## Durable Human-in-the-Loop
+
+ADK pauses a run for human input (a node yielding `RequestInput`) or tool
+confirmation (`FunctionTool(..., require_confirmation=True)`); in a Temporal
+workflow that pause becomes a durable wait. The
+`pending_hitl_requests` / `hitl_input_response` / `hitl_confirmation_response`
+helpers cover the wire format; the wait itself is ordinary workflow code:
+
+```python
+from temporalio.contrib.google_adk_agents import (
+    HitlRequest,
+    hitl_input_response,
+    pending_hitl_requests,
+)
+
+@workflow.defn
+class ApprovalWorkflow:
+    def __init__(self) -> None:
+        self._pending: dict[str, HitlRequest] = {}
+        self._responses: dict[str, Any] = {}
+
+    @workflow.query
+    def pending_requests(self) -> list[HitlRequest]:
+        return list(self._pending.values())
+
+    @workflow.update
+    def respond(self, interrupt_id: str, response: Any) -> None:
+        self._responses[interrupt_id] = response
+
+    @workflow.run
+    async def run(self, prompt: str) -> str:
+        runner = Runner(
+            app_name="app", node=graph, session_service=InMemorySessionService()
+        )
+        session = await runner.session_service.create_session(
+            app_name="app", user_id="user"
+        )
+        message = types.Content(role="user", parts=[types.Part(text=prompt)])
+        result = ""
+        while True:
+            async for event in runner.run_async(
+                user_id="user", session_id=session.id, new_message=message
+            ):
+                for request in pending_hitl_requests(event):
+                    self._pending[request.interrupt_id] = request
+                if event.content and event.content.parts and event.content.parts[0].text:
+                    result = event.content.parts[0].text
+            if not self._pending:
+                return result
+            await workflow.wait_condition(
+                lambda: any(i in self._responses for i in self._pending)
+            )
+            parts = [
+                hitl_input_response(i, self._responses.pop(i))
+                for i in list(self._pending)
+                if i in self._responses
+            ]
+            for part in parts:
+                self._pending.pop(part.function_response.id)
+            message = types.Content(role="user", parts=parts)
+```
+
+Tool confirmation composes with `activity_tool` with no extra plumbing —
+`FunctionTool(func=activity_tool(risky_activity, ...), require_confirmation=True)`
+never schedules the activity until the human approves (answer with
+`hitl_confirmation_response(interrupt_id, confirmed=True)`). MCP tools
+requesting confirmation via `tool_context.request_confirmation(...)` flow
+through the same loop. Partial responses are fine: unanswered requests stay
+pending across `run_async` turns.
+
+> **Replay-safety note:** HITL resume matches recorded human responses against
+> generated interrupt/function-call ids, so those ids must regenerate
+> identically on replay. The plugin installs ADK's platform time/uuid/random
+> providers as process-wide defaults to guarantee this. On google-adk versions
+> where `RequestInput` ids bypass the platform seam, pass an explicit
+> `interrupt_id` to `RequestInput(...)` (as the examples here do).
+
+## Determinism Notes
+
+- The plugin patches ADK's `google.adk.platform` time, uuid, and (on ADK
+  versions that expose it) random providers to `workflow.now()`,
+  `workflow.uuid4()`, and `workflow.random()` inside workflows.
+- ADK node `timeout=`/`RetryConfig` map onto durable timers
+  (`asyncio.wait_for`/`asyncio.sleep`). For activity-backed nodes, prefer
+  Temporal activity timeouts and `retry_policy` via `activity_node(...)`
+  options; an ADK `RetryConfig` on top would retry on top of Temporal's own
+  activity retries, and an ADK node timeout cancels the in-flight activity.
+- Never set `RunConfig.tool_thread_pool_config` inside a workflow — it runs
+  tools on threads, which breaks workflow determinism. Live/BIDI mode is
+  likewise unsupported in workflows.
+- ADK resume is at-least-once: on a HITL resume, completed nodes fast-forward
+  from the in-memory session, but `rerun_on_resume=True` node bodies
+  re-execute. This is deterministic under Temporal replay; schedule side
+  effects through activities (retried/tracked by Temporal) or make them
+  idempotent.
+- Very long HITL conversations grow the workflow history with each turn;
+  consider `continue-as-new` boundaries between `run_async` turns for
+  long-running chats.
+
 ## Integration Points
 
 This integration provides comprehensive support for running Google ADK Agents within Temporal workflows while maintaining:

temporalio/contrib/google_adk_agents/__init__.py

@@ -3,6 +3,12 @@
 This module provides the necessary components to run ADK Agents within Temporal Workflows.
 """
 
+from temporalio.contrib.google_adk_agents._hitl import (
+    HitlRequest,
+    hitl_confirmation_response,
+    hitl_input_response,
+    pending_hitl_requests,
+)
 from temporalio.contrib.google_adk_agents._mcp import (
     TemporalMcpToolSet,
     TemporalMcpToolSetProvider,
@@ -14,7 +20,11 @@
 
 __all__ = [
     "GoogleAdkPlugin",
+    "HitlRequest",
     "TemporalMcpToolSet",
     "TemporalMcpToolSetProvider",
     "TemporalModel",
+    "hitl_confirmation_response",
+    "hitl_input_response",
+    "pending_hitl_requests",
 ]

temporalio/contrib/google_adk_agents/_hitl.py

@@ -0,0 +1,193 @@
+"""Human-in-the-loop helpers for Google ADK agents running in Temporal workflows.
+
+ADK pauses a run by emitting an event that carries a special function call
+(``adk_request_input`` for human-input nodes, ``adk_request_confirmation`` for
+tool confirmation, ``adk_request_credential`` for auth) and resumes when a
+later user message answers it with a matching ``FunctionResponse``. Inside a
+Temporal workflow the pause maps naturally onto a durable wait: collect the
+pending requests from the events yielded by ``runner.run_async``, expose them
+via a query, wait for responses via ``workflow.wait_condition`` on a signal or
+update handler, then call ``runner.run_async`` again with the response parts.
+
+These helpers cover the wire format only; the wait topology stays ordinary
+Temporal workflow code.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Literal, Mapping, Optional
+
+from google.adk.events import Event
+from google.adk.tools.tool_confirmation import ToolConfirmation
+from google.genai import types
+
+# The function-call names ADK uses on the wire for HITL pauses.
+_REQUEST_INPUT_FUNCTION_CALL_NAME = "adk_request_input"
+_REQUEST_CONFIRMATION_FUNCTION_CALL_NAME = "adk_request_confirmation"
+_REQUEST_CREDENTIAL_FUNCTION_CALL_NAME = "adk_request_credential"
+
+_KIND_BY_FUNCTION_CALL_NAME: dict[
+    str, Literal["input", "tool_confirmation", "credential"]
+] = {
+    _REQUEST_INPUT_FUNCTION_CALL_NAME: "input",
+    _REQUEST_CONFIRMATION_FUNCTION_CALL_NAME: "tool_confirmation",
+    _REQUEST_CREDENTIAL_FUNCTION_CALL_NAME: "credential",
+}
+
+
+@dataclass(frozen=True)
+class HitlRequest:
+    """A pending human-in-the-loop request extracted from an ADK event.
+
+    .. warning::
+        This class is experimental and may change in future versions.
+        Use with caution in production environments.
+
+    Attributes:
+        kind: ``"input"`` for a human-input node's ``RequestInput``,
+            ``"tool_confirmation"`` for a tool confirmation request,
+            ``"credential"`` for an auth request.
+        interrupt_id: The id a response must reference. Pass it to
+            :func:`hitl_input_response` or :func:`hitl_confirmation_response`.
+        invocation_id: The ADK invocation that is paused on this request.
+        author: The agent/node that raised the request.
+        message: Human-readable prompt (``RequestInput.message`` or the
+            confirmation hint), if any.
+        payload: Custom payload attached to the request, if any.
+        response_schema: JSON schema the response must satisfy (input
+            requests only), if any.
+        original_function_call: For tool confirmations, the gated tool call
+            as ``{"name": ..., "args": ..., "id": ...}`` — useful for
+            displaying what is being approved.
+    """
+
+    kind: Literal["input", "tool_confirmation", "credential"]
+    interrupt_id: str
+    invocation_id: Optional[str] = None
+    author: Optional[str] = None
+    message: Optional[str] = None
+    payload: Optional[Any] = None
+    response_schema: Optional[dict[str, Any]] = None
+    original_function_call: Optional[dict[str, Any]] = None
+
+
+def pending_hitl_requests(event: Event) -> list[HitlRequest]:
+    """Extracts pending human-in-the-loop requests from an ADK event.
+
+    .. warning::
+        This function is experimental and may change in future versions.
+        Use with caution in production environments.
+
+    Call this on each event yielded by ``runner.run_async``. A non-empty
+    result means the run is pausing for the returned requests; once
+    ``run_async`` completes, resume by sending a new user message whose parts
+    answer them (see :func:`hitl_input_response` and
+    :func:`hitl_confirmation_response`).
+
+    Args:
+        event: An event yielded by ``runner.run_async``.
+
+    Returns:
+        The requests carried by this event; empty for ordinary events.
+    """
+    if not event.long_running_tool_ids:
+        return []
+    if not event.content or not event.content.parts:
+        return []
+    requests: list[HitlRequest] = []
+    for part in event.content.parts:
+        function_call = part.function_call
+        if not function_call or not function_call.id:
+            continue
+        kind = _KIND_BY_FUNCTION_CALL_NAME.get(function_call.name or "")
+        if kind is None:
+            continue
+        args = function_call.args or {}
+        message: Optional[str] = None
+        payload: Optional[Any] = None
+        response_schema: Optional[dict[str, Any]] = None
+        original_function_call: Optional[dict[str, Any]] = None
+        if kind == "input":
+            message = args.get("message")
+            payload = args.get("payload")
+            response_schema = args.get("response_schema")
+        elif kind == "tool_confirmation":
+            confirmation = args.get("toolConfirmation") or {}
+            message = confirmation.get("hint")
+            payload = confirmation.get("payload")
+            original_function_call = args.get("originalFunctionCall")
+        else:
+            payload = args
+        requests.append(
+            HitlRequest(
+                kind=kind,
+                interrupt_id=function_call.id,
+                invocation_id=event.invocation_id,
+                author=event.author,
+                message=message,
+                payload=payload,
+                response_schema=response_schema,
+                original_function_call=original_function_call,
+            )
+        )
+    return requests
+
+
+def hitl_input_response(interrupt_id: str, response: Any) -> types.Part:
+    """Builds the message part answering a human-input (``RequestInput``) request.
+
+    .. warning::
+        This function is experimental and may change in future versions.
+        Use with caution in production environments.
+
+    Compose one or more parts into ``types.Content(role="user", parts=[...])``
+    and pass it as ``new_message`` to ``runner.run_async`` to resume the
+    paused run. Non-mapping values are wrapped as ``{"result": value}`` per
+    ADK's convention and unwrapped on delivery to the node.
+
+    Args:
+        interrupt_id: The :attr:`HitlRequest.interrupt_id` being answered.
+        response: The human's response value.
+    """
+    if isinstance(response, Mapping):
+        response_dict = dict(response)
+    else:
+        response_dict = {"result": response}
+    return types.Part(
+        function_response=types.FunctionResponse(
+            id=interrupt_id,
+            name=_REQUEST_INPUT_FUNCTION_CALL_NAME,
+            response=response_dict,
+        )
+    )
+
+
+def hitl_confirmation_response(
+    interrupt_id: str, *, confirmed: bool, payload: Optional[Any] = None
+) -> types.Part:
+    """Builds the message part answering a tool-confirmation request.
+
+    .. warning::
+        This function is experimental and may change in future versions.
+        Use with caution in production environments.
+
+    Compose one or more parts into ``types.Content(role="user", parts=[...])``
+    and pass it as ``new_message`` to ``runner.run_async`` to resume the
+    paused run. If ``confirmed`` is false the gated tool is not executed and
+    the model receives a rejection response instead.
+
+    Args:
+        interrupt_id: The :attr:`HitlRequest.interrupt_id` being answered.
+        confirmed: Whether the human approved running the tool.
+        payload: Optional custom payload made available to the tool via
+            ``tool_context.tool_confirmation.payload``.
+    """
+    confirmation = ToolConfirmation(confirmed=confirmed, payload=payload)
+    return types.Part(
+        function_response=types.FunctionResponse(
+            id=interrupt_id,
+            name=_REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
+            response=confirmation.model_dump(mode="json"),
+        )
+    )