From 59d946d8f4f80b6523945b31946e5fb630e54d3f Mon Sep 17 00:00:00 2001 From: Marcus Motill Date: Mon, 15 Dec 2025 18:28:27 +0000 Subject: [PATCH 1/7] feat: Add Temporal integration and deterministic runtime support --- src/google/adk/events/event.py | 5 +- src/google/adk/integrations/__init__.py | 0 src/google/adk/integrations/temporal.py | 151 ++++++++ src/google/adk/runtime.py | 53 +++ .../adk/sessions/in_memory_session_service.py | 5 +- tests/integration/README_TEMPORAL.md | 23 ++ .../manual_test_temporal_integration.py | 321 ++++++++++++++++++ 7 files changed, 554 insertions(+), 4 deletions(-) create mode 100644 src/google/adk/integrations/__init__.py create mode 100644 src/google/adk/integrations/temporal.py create mode 100644 src/google/adk/runtime.py create mode 100644 tests/integration/README_TEMPORAL.md create mode 100644 tests/integration/manual_test_temporal_integration.py diff --git a/src/google/adk/events/event.py b/src/google/adk/events/event.py index cca086430b..faeefc802f 100644 --- a/src/google/adk/events/event.py +++ b/src/google/adk/events/event.py @@ -18,6 +18,7 @@ from typing import Optional import uuid +from google.adk import runtime from google.genai import types from pydantic import alias_generators from pydantic import ConfigDict @@ -70,7 +71,7 @@ class Event(LlmResponse): # Do not assign the ID. It will be assigned by the session. id: str = '' """The unique identifier of the event.""" - timestamp: float = Field(default_factory=lambda: datetime.now().timestamp()) + timestamp: float = Field(default_factory=lambda: runtime.get_time()) """The timestamp of the event.""" def model_post_init(self, __context): @@ -125,4 +126,4 @@ def has_trailing_code_execution_result( @staticmethod def new_id(): - return str(uuid.uuid4()) + return runtime.new_uuid() diff --git a/src/google/adk/integrations/__init__.py b/src/google/adk/integrations/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/google/adk/integrations/temporal.py b/src/google/adk/integrations/temporal.py new file mode 100644 index 0000000000..341e2ab226 --- /dev/null +++ b/src/google/adk/integrations/temporal.py @@ -0,0 +1,151 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Temporal integration helpers for ADK.""" + +import functools +from typing import Any, AsyncGenerator, Callable, Optional, List + +from temporalio import workflow, activity +from google.adk.models import BaseLlm, LlmRequest, LlmResponse, LLMRegistry +from google.genai import types + + +def activity_as_tool( + activity_def: Callable, + **activity_options: Any +) -> Callable: + """Wraps a Temporal Activity Definition into an ADK-compatible tool. + + Args: + activity_def: The Temporal activity definition (decorated with @activity.defn). + **activity_options: Options to pass to workflow.execute_activity + (e.g. start_to_close_timeout, retry_policy). + + Returns: + A callable tool that executes the activity when invoked. + """ + + # We create a wrapper that delegates to workflow.execute_activity + async def tool_wrapper(*args, **kwargs) -> Any: + # Note: ADK tools usually pass args/kwargs strictly matched to signature. + # Activities expect positional args in a list if 'args' is used. + # If the tool signature matches the activity signature, we can pass args. + # It's safer if activity takes Pydantic models or simple types. + + # We assume strict positional argument mapping for now, or simplistic kwargs handling if supported. + # Temporal Python SDK typically invokes activities with `args=[...]`. + + return await workflow.execute_activity( + activity_def, + args=list(args) + list(kwargs.values()) if kwargs else list(args), + **activity_options + ) + + # Copy metadata so ADK can inspect the tool (name, docstring, annotations) + # ADK uses this to generate the tool schema for the LLM. + tool_wrapper.__doc__ = activity_def.__doc__ + tool_wrapper.__name__ = getattr(activity_def, "name", activity_def.__name__) + + # Attempt to copy annotations if they exist + if hasattr(activity_def, "__annotations__"): + tool_wrapper.__annotations__ = activity_def.__annotations__ + + # CRITICAL: Copy signature so FunctionTool can generate correct parameters schema + try: + import inspect + tool_wrapper.__signature__ = inspect.signature(activity_def) + except Exception: + pass # Fallback if signature copy fails (e.g. builtins) + + return tool_wrapper + + +@activity.defn +async def generate_content_activity(request: LlmRequest) -> List[LlmResponse]: + """Generic activity to invoke an LLM via ADK's LLMRegistry. + + The model name is expected to be in `request.model`. + """ + if not request.model: + raise ValueError("LlmRequest.model must be set when using generate_content_activity.") + + llm = LLMRegistry.new_llm(request.model) + return [response async for response in llm.generate_content_async(request)] + + +class TemporalModel(BaseLlm): + """An ADK ModelWrapper that executes content generation as a Temporal Activity. + + This effectively delegates the 'generate_content' call to an external Activity, + ensuring that the network I/O to Vertex/Gemini is recorded in Temporal history. + """ + + activity_def: Callable + activity_options: dict[str, Any] + + def __init__( + self, + model_name: str, + activity_def: Callable = generate_content_activity, + **activity_options: Any + ): + """Initializes the TemporalModel. + + Args: + model_name: The name of the model to report to ADK. + activity_def: The Temporal activity definition to invoke. + Defaults to `generate_content_activity`. + **activity_options: Options for workflow.execute_activity. + """ + super().__init__( + model=model_name, + activity_def=activity_def, + activity_options=activity_options + ) + + async def generate_content_async( + self, + llm_request: LlmRequest, + stream: bool = False + ) -> AsyncGenerator[LlmResponse, None]: + """Generates content by calling the configured Temporal Activity.""" + + # Ensure model name is carried in the request for the generic activity + if not llm_request.model: + llm_request.model = self.model + + # Note: Temporal Activities are not typically streaming in the Python SDK + # in the way python async generators work (streaming back to workflow is complex). + # Standard approach is to return the full response. + # We will assume non-streaming activity execution for now. + + # Execute the activity + responses: List[LlmResponse] = await workflow.execute_activity( + self.activity_def, + args=[llm_request], + **self.activity_options + ) + + # Yield the responses + for response in responses: + yield response + + @classmethod + def default_activities(cls) -> List[Callable]: + """Returns the default activities used by this model wrapper. + + Useful for registering activities with the Temporal Worker. + """ + return [generate_content_activity] diff --git a/src/google/adk/runtime.py b/src/google/adk/runtime.py new file mode 100644 index 0000000000..edd6be9045 --- /dev/null +++ b/src/google/adk/runtime.py @@ -0,0 +1,53 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Runtime module for abstracting system primitives like time and UUIDs.""" + +import time +import uuid +from typing import Callable + +_time_provider: Callable[[], float] = time.time +_id_provider: Callable[[], str] = lambda: str(uuid.uuid4()) + + +def set_time_provider(provider: Callable[[], float]) -> None: + """Sets the provider for the current time. + + Args: + provider: A callable that returns the current time in seconds since the + epoch. + """ + global _time_provider + _time_provider = provider + + +def set_id_provider(provider: Callable[[], str]) -> None: + """Sets the provider for generating unique IDs. + + Args: + provider: A callable that returns a unique ID string. + """ + global _id_provider + _id_provider = provider + + +def get_time() -> float: + """Returns the current time in seconds since the epoch.""" + return _time_provider() + + +def new_uuid() -> str: + """Returns a new unique ID.""" + return _id_provider() diff --git a/src/google/adk/sessions/in_memory_session_service.py b/src/google/adk/sessions/in_memory_session_service.py index 6ba7f0bb01..9df7ca17c9 100644 --- a/src/google/adk/sessions/in_memory_session_service.py +++ b/src/google/adk/sessions/in_memory_session_service.py @@ -22,6 +22,7 @@ from typing_extensions import override +from google.adk import runtime from . import _session_util from ..errors.already_exists_error import AlreadyExistsError from ..events.event import Event @@ -108,14 +109,14 @@ def _create_session_impl( session_id = ( session_id.strip() if session_id and session_id.strip() - else str(uuid.uuid4()) + else runtime.new_uuid() ) session = Session( app_name=app_name, user_id=user_id, id=session_id, state=session_state or {}, - last_update_time=time.time(), + last_update_time=runtime.get_time(), ) if app_name not in self.sessions: diff --git a/tests/integration/README_TEMPORAL.md b/tests/integration/README_TEMPORAL.md new file mode 100644 index 0000000000..31a9c3855e --- /dev/null +++ b/tests/integration/README_TEMPORAL.md @@ -0,0 +1,23 @@ +# Temporal Integration Tests + +The file `manual_test_temporal_integration.py` contains integration tests for ADK's Temporal support. +It is named `manual_test_...` to be excluded from standard CI/test runs because it requires: + +1. **Local Temporal Server**: You must have a Temporal server running locally (e.g., via `temporal server start-dev`). +2. **GCP Credentials**: Environment variables `GOOGLE_CLOUD_PROJECT` and `GOOGLE_CLOUD_LOCATION` must be set. +3. **Local Environment**: It assumes `localhost:7233`. + +## How to Run + +1. Start Temporal Server: + ```bash + temporal server start-dev + ``` + +2. Run the test directly: + ```bash + export GOOGLE_CLOUD_PROJECT="your-project" + export GOOGLE_CLOUD_LOCATION="us-central1" + + uv run pytest tests/integration/manual_test_temporal_integration.py + ``` diff --git a/tests/integration/manual_test_temporal_integration.py b/tests/integration/manual_test_temporal_integration.py new file mode 100644 index 0000000000..8e2af4e91b --- /dev/null +++ b/tests/integration/manual_test_temporal_integration.py @@ -0,0 +1,321 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Integration tests for ADK Temporal support.""" + +import dataclasses +import logging +import uuid +import os +from datetime import timedelta +from typing import AsyncGenerator, List + +import pytest +from temporalio import workflow, activity +from temporalio.client import Client +from temporalio.contrib.pydantic import pydantic_data_converter, PydanticPayloadConverter +from temporalio.converter import DataConverter, DefaultPayloadConverter +from temporalio.plugin import SimplePlugin +from temporalio.worker import Worker, WorkflowRunner +from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner, SandboxRestrictions + + +from google.genai import types +from google.adk import Agent, Runner, runtime +from google.adk.agents import LlmAgent +from google.adk.tools import AgentTool +from google.adk.models import LlmRequest, LlmResponse, LLMRegistry +from google.adk.sessions import InMemorySessionService +from google.adk.utils.context_utils import Aclosing +from google.adk.events import Event +from google.adk.integrations.temporal import activity_as_tool, TemporalModel, generate_content_activity + +# Required Environment Variables for this test: +# - GOOGLE_CLOUD_PROJECT +# - GOOGLE_CLOUD_LOCATION +# - GOOGLE_GENAI_USE_VERTEXAI (optional, defaults to 1 for this test if needed, or set externally) +# - Temporal Server running at localhost:7233 + +logger = logging.getLogger(__name__) + + +@activity.defn +async def get_weather(city: str) -> str: + """Activity that gets weather.""" + return "Warm and sunny. 17 degrees." + +# --- Customized LLM Activities for Better Trace Visibility --- + +@activity.defn(name="coordinator_think") +async def coordinator_think(req: LlmRequest) -> List[LlmResponse]: + """Activity for the Coordinator agent.""" + return await generate_content_activity(req) + +@activity.defn(name="tool_agent_think") +async def tool_agent_think(req: LlmRequest) -> List[LlmResponse]: + """Activity for the Tool Agent.""" + return await generate_content_activity(req) + +@activity.defn(name="specialist_think") +async def specialist_think(req: LlmRequest) -> List[LlmResponse]: + """Activity for the Specialist/Handoff Agent.""" + return await generate_content_activity(req) + + +@workflow.defn +class WeatherAgent: + @workflow.run + async def run(self, prompt: str) -> Event | None: + logger.info("Workflow started.") + + # 1. Configure ADK Runtime to use Temporal Determinism + runtime.set_time_provider(lambda: workflow.now().timestamp()) + runtime.set_id_provider(lambda: str(workflow.uuid4())) + + # 2. Define Agent using Temporal Helpers + # Uses generic 'generate_content_activity' by default + agent_model = TemporalModel( + model_name="gemini-2.5-pro", + start_to_close_timeout=timedelta(minutes=2) + ) + + # Wraps 'get_weather' activity as a Tool + weather_tool = activity_as_tool( + get_weather, + start_to_close_timeout=timedelta(seconds=60) + ) + + agent = Agent( + name='test_agent', + model=agent_model, + tools=[weather_tool] + ) + + # 3. Create Session (uses runtime.new_uuid() -> workflow.uuid4()) + session_service = InMemorySessionService() + logger.info("Create session.") + session = await session_service.create_session(app_name="test_app", user_id="test") + + logger.info(f"Session created with ID: {session.id}") + + # 4. Run Agent + runner = Runner( + agent=agent, + app_name='test_app', + session_service=session_service, + ) + + logger.info("Starting runner.") + last_event = None + async with Aclosing(runner.run_async( + user_id="test", + session_id=session.id, + new_message=types.Content( + role='user', parts=[types.Part(text=prompt)] + ), + )) as agen: + async for event in agen: + logger.info(f"Event: {event}") + last_event = event + + return last_event + +@workflow.defn +class MultiAgentWorkflow: + @workflow.run + async def run(self, prompt: str) -> str: + # 1. Runtime Setup + runtime.set_time_provider(lambda: workflow.now().timestamp()) + runtime.set_id_provider(lambda: str(workflow.uuid4())) + + # 2. Define Distinct Models for Visualization + # We use a separate activity for each agent so they show up distinctly in the Temporal UI. + + coordinator_model = TemporalModel( + model_name="gemini-2.5-pro", + activity_def=coordinator_think, + start_to_close_timeout=timedelta(minutes=2) + ) + + tool_agent_model = TemporalModel( + model_name="gemini-2.5-pro", + activity_def=tool_agent_think, + start_to_close_timeout=timedelta(minutes=2) + ) + + specialist_model = TemporalModel( + model_name="gemini-2.5-pro", + activity_def=specialist_think, + start_to_close_timeout=timedelta(minutes=2) + ) + + # 3. Define Sub-Agents + + # Agent to be used as a Tool + tool_agent = LlmAgent( + name="ToolAgent", + model=tool_agent_model, + instruction="You are a tool agent. You help with specific sub-tasks. Always include 'From ToolAgent:' in your response." + ) + agent_tool = AgentTool(tool_agent) + + # Agent to be transferred to (Handoff) + handoff_agent = LlmAgent( + name="HandoffAgent", + model=specialist_model, + instruction="You are a Specialist Agent. You handle specialized requests. Always include 'From HandoffAgent:' in your response." + ) + + # 4. Define Parent Agent + parent_agent = LlmAgent( + name="Coordinator", + model=coordinator_model, + # Instructions to guide the LLM when to use which + instruction=( + "You are a Coordinator. " + "CRITICAL INSTRUCTION: You MUST NOT answer user queries directly if they related to specific tasks. " + "1. If the user asks for 'help' or 'subtask', you MUST use the 'ToolAgent' tool (AgentTool). " + "2. If the user asks to 'switch' or 'specialist', you MUST transfer to the HandoffAgent using 'transfer_to_agent'. " + "Do not apologize. Do not say you will do it. Just call the function." + ), + tools=[agent_tool], + sub_agents=[handoff_agent] + ) + + # 5. Execute + session_service = InMemorySessionService() + session = await session_service.create_session(app_name="multi_agent_app", user_id="user_MULTI") + + runner = Runner( + agent=parent_agent, + app_name='multi_agent_app', + session_service=session_service, + ) + + # We will run a multi-turn conversation to test both paths + # Turn 1: Trigger Tool + logger.info("--- Turn 1: Trigger Tool ---") + tool_response_text = "" + async with Aclosing(runner.run_async( + user_id=session.user_id, + session_id=session.id, + new_message=types.Content(role='user', parts=[types.Part(text="I need help with a subtask.")]) + )) as agen: + async for event in agen: + logger.info(f"Event Author: {event.author} | Actions: {event.actions}") + if event.content and event.content.parts: + for part in event.content.parts: + if part.text: tool_response_text += part.text + + # Turn 2: Trigger Handoff + logger.info("--- Turn 2: Trigger Handoff ---") + handoff_response_text = "" + last_author = "" + async with Aclosing(runner.run_async( + user_id=session.user_id, + session_id=session.id, + new_message=types.Content(role='user', parts=[types.Part(text="Please switch me to the specialist.")]) + )) as agen: + async for event in agen: + logger.info(f"Event Author: {event.author} | Actions: {event.actions}") + last_author = event.author + if event.content and event.content.parts: + for part in event.content.parts: + if part.text: handoff_response_text += part.text + + logger.info(f"Tool Response: {tool_response_text}") + logger.info(f"Handoff Response: {handoff_response_text}") + + return f"Tool: {tool_response_text} | Handoff: {handoff_response_text}" + + +class ADKPlugin(SimplePlugin): + def __init__(self): + super().__init__( + name="ADKPlugin", + data_converter=_data_converter, + workflow_runner=workflow_runner, + ) + +def workflow_runner(runner: WorkflowRunner | None) -> WorkflowRunner: + if not runner: + raise ValueError("No WorkflowRunner provided to the ADK plugin.") + + # If in sandbox, add additional passthrough + if isinstance(runner, SandboxedWorkflowRunner): + return dataclasses.replace( + runner, + restrictions=runner.restrictions.with_passthrough_modules("google.adk", "google.genai"), + ) + return runner + +def _data_converter(converter: DataConverter | None) -> DataConverter: + if converter is None: + return pydantic_data_converter + elif converter.payload_converter_class is DefaultPayloadConverter: + return dataclasses.replace( + converter, payload_converter_class=PydanticPayloadConverter + ) + elif not isinstance(converter.payload_converter, PydanticPayloadConverter): + raise ValueError( + "The payload converter must be of type PydanticPayloadConverter." + ) + return converter + +@pytest.mark.asyncio +async def test_temporalio_integration(): + """Run full integration test with Temporal Server.""" + + # Normally this should only run if local Temporal server is available + # For now, we assume it is, as per user context. + + # Start client/worker + if "GOOGLE_CLOUD_PROJECT" not in os.environ: + pytest.skip("GOOGLE_CLOUD_PROJECT not set. Skipping integration test.") + + try: + client = await Client.connect("localhost:7233", plugins=[ADKPlugin()]) + except RuntimeError: + pytest.skip("Could not connect to Temporal server. Is it running?") + + async with Worker( + client, + workflows=[WeatherAgent, MultiAgentWorkflow], + activities=TemporalModel.default_activities() + [ + get_weather, + coordinator_think, + tool_agent_think, + specialist_think + ], + task_queue="hello_world_queue", + max_cached_workflows=0, + ) as worker: + print("Worker started.") + # Run Weather Agent + result_weather = await client.execute_workflow( + WeatherAgent.run, + "What is the weather in Tokyo?", + id=str(uuid.uuid4()), + task_queue="hello_world_queue", + ) + print(f"Weather Agent Result: {result_weather}") + + # Run Multi-Agent Workflow + result_multi = await client.execute_workflow( + MultiAgentWorkflow.run, + "start", # Argument ignored in run logic (hardcoded prompts) + id=str(uuid.uuid4()), + task_queue="hello_world_queue", + ) + print(f"Multi-Agent Result: {result_multi}") From 81a5935753338986cc27456a2373fe9fdd25dc08 Mon Sep 17 00:00:00 2001 From: Marcus Motill Date: Mon, 15 Dec 2025 18:40:05 +0000 Subject: [PATCH 2/7] test: Add unit tests for runtime and Temporal integration --- tests/unittests/integrations/test_temporal.py | 143 ++++++++++++++++++ tests/unittests/test_runtime.py | 56 +++++++ 2 files changed, 199 insertions(+) create mode 100644 tests/unittests/integrations/test_temporal.py create mode 100644 tests/unittests/test_runtime.py diff --git a/tests/unittests/integrations/test_temporal.py b/tests/unittests/integrations/test_temporal.py new file mode 100644 index 0000000000..ec7fed9e6a --- /dev/null +++ b/tests/unittests/integrations/test_temporal.py @@ -0,0 +1,143 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for Temporal integration helpers.""" + +import unittest +from unittest.mock import MagicMock, patch, AsyncMock +import sys +import asyncio +from typing import Any + +from google.genai import types + +# Configure Mocks globally +# We create fresh mocks here. +mock_workflow = MagicMock() +mock_activity = MagicMock() +mock_worker = MagicMock() +mock_client = MagicMock() + +# Important: execute_activity must be awaitable +mock_workflow.execute_activity = AsyncMock(return_value="mock_result") + +# Mock the parent package +mock_temporalio = MagicMock() +mock_temporalio.workflow = mock_workflow +mock_temporalio.activity = mock_activity +mock_temporalio.worker = mock_worker +mock_temporalio.client = mock_client + +# Mock sys.modules +with patch.dict(sys.modules, { + "temporalio": mock_temporalio, + "temporalio.workflow": mock_workflow, + "temporalio.activity": mock_activity, + "temporalio.worker": mock_worker, + "temporalio.client": mock_client, +}): + from google.adk.integrations import temporal + from google.adk.models import LlmRequest, LlmResponse + + +class TestTemporalIntegration(unittest.TestCase): + + def test_activity_as_tool_wrapper(self): + # Reset mocks + mock_workflow.reset_mock() + mock_workflow.execute_activity = AsyncMock(return_value="mock_result") + + # Verify mock setup + # If this fails, then 'temporal.workflow' is NOT our 'mock_workflow' + assert temporal.workflow.execute_activity is mock_workflow.execute_activity + + # Define a fake activity + async def fake_activity(arg: str) -> str: + """My Docstring.""" + return f"Hello {arg}" + + fake_activity.name = "fake_activity_name" + + # Create tool + tool = temporal.activity_as_tool( + fake_activity, + start_to_close_timeout=100 + ) + + # Check metadata + self.assertEqual(tool.__name__, "fake_activity_name") + self.assertEqual(tool.__doc__, "My Docstring.") + + # Run tool (wrapper) + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + result = loop.run_until_complete(tool("World")) + finally: + loop.close() + + # Verify call + mock_workflow.execute_activity.assert_called_once() + args, kwargs = mock_workflow.execute_activity.call_args + self.assertEqual(kwargs['args'], ['World']) + self.assertEqual(kwargs['start_to_close_timeout'], 100) + + def test_temporal_model_generate_content(self): + # Reset mocks + mock_workflow.reset_mock() + + # Prepare valid LlmResponse with content + response_content = types.Content(parts=[types.Part(text="test_resp")]) + llm_response = LlmResponse(content=response_content) + + # generate_content_async expects execute_activity to return response list (iterator) + mock_workflow.execute_activity = AsyncMock(return_value=[llm_response]) + + # Mock an activity def + mock_activity_def = MagicMock() + + # Create model + model = temporal.TemporalModel( + model_name="test-model", + activity_def=mock_activity_def, + schedule_to_close_timeout=50 + ) + + # Create request + req = LlmRequest(model="test-model", prompt="hi") + + # Run generate_content_async (it is an async generator) + async def run_gen(): + results = [] + async for r in model.generate_content_async(req): + results.append(r) + return results + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + results = loop.run_until_complete(run_gen()) + finally: + loop.close() + + # Verify execute_activity called + mock_workflow.execute_activity.assert_called_once() + args, kwargs = mock_workflow.execute_activity.call_args + self.assertEqual(kwargs['args'], [req]) + self.assertEqual(kwargs['schedule_to_close_timeout'], 50) + self.assertEqual(len(results), 1) + self.assertEqual(results[0].content.parts[0].text, "test_resp") + diff --git a/tests/unittests/test_runtime.py b/tests/unittests/test_runtime.py new file mode 100644 index 0000000000..cde013631b --- /dev/null +++ b/tests/unittests/test_runtime.py @@ -0,0 +1,56 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for the runtime module.""" + +import time +import uuid +import unittest +from unittest.mock import MagicMock, patch + +from google.adk import runtime + + +class TestRuntime(unittest.TestCase): + + def tearDown(self): + # Reset providers to default after each test + runtime.set_time_provider(time.time) + runtime.set_id_provider(lambda: str(uuid.uuid4())) + + def test_default_time_provider(self): + # Verify it returns a float that is close to now + now = time.time() + rt_time = runtime.get_time() + self.assertIsInstance(rt_time, float) + self.assertAlmostEqual(rt_time, now, delta=1.0) + + def test_default_id_provider(self): + # Verify it returns a string uuid + uid = runtime.new_uuid() + self.assertIsInstance(uid, str) + # Should be parseable as uuid + uuid.UUID(uid) + + def test_custom_time_provider(self): + # Test override + mock_time = 123456789.0 + runtime.set_time_provider(lambda: mock_time) + self.assertEqual(runtime.get_time(), mock_time) + + def test_custom_id_provider(self): + # Test override + mock_id = "test-id-123" + runtime.set_id_provider(lambda: mock_id) + self.assertEqual(runtime.new_uuid(), mock_id) From 1a639853ad4739ba96b7a3f99a15bfd3c897e375 Mon Sep 17 00:00:00 2001 From: Marcus Motill Date: Mon, 15 Dec 2025 18:42:05 +0000 Subject: [PATCH 3/7] fix(temporal): Use robust argument binding and remove unused variable --- src/google/adk/integrations/temporal.py | 24 +++++++++++-------- .../manual_test_temporal_integration.py | 2 -- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/src/google/adk/integrations/temporal.py b/src/google/adk/integrations/temporal.py index 341e2ab226..d05145b5ac 100644 --- a/src/google/adk/integrations/temporal.py +++ b/src/google/adk/integrations/temporal.py @@ -15,6 +15,7 @@ """Temporal integration helpers for ADK.""" import functools +import inspect from typing import Any, AsyncGenerator, Callable, Optional, List from temporalio import workflow, activity @@ -39,17 +40,21 @@ def activity_as_tool( # We create a wrapper that delegates to workflow.execute_activity async def tool_wrapper(*args, **kwargs) -> Any: - # Note: ADK tools usually pass args/kwargs strictly matched to signature. - # Activities expect positional args in a list if 'args' is used. - # If the tool signature matches the activity signature, we can pass args. - # It's safer if activity takes Pydantic models or simple types. - - # We assume strict positional argument mapping for now, or simplistic kwargs handling if supported. - # Temporal Python SDK typically invokes activities with `args=[...]`. - + # Bind arguments to the activity signature to ensure correct order/mapping. + try: + sig = inspect.signature(activity_def) + bound_args = sig.bind(*args, **kwargs) + bound_args.apply_defaults() + activity_args = [bound_args.arguments[p] for p in sig.parameters] + except (TypeError, ValueError): + # Fallback for built-ins or other complex callables where binding may fail + # or if arguments don't match signature (e.g. simpler invocation). + # Temporal Python SDK typically invokes activities with `args=[...]`. + activity_args = list(args) + list(kwargs.values()) if kwargs else list(args) + return await workflow.execute_activity( activity_def, - args=list(args) + list(kwargs.values()) if kwargs else list(args), + args=activity_args, **activity_options ) @@ -64,7 +69,6 @@ async def tool_wrapper(*args, **kwargs) -> Any: # CRITICAL: Copy signature so FunctionTool can generate correct parameters schema try: - import inspect tool_wrapper.__signature__ = inspect.signature(activity_def) except Exception: pass # Fallback if signature copy fails (e.g. builtins) diff --git a/tests/integration/manual_test_temporal_integration.py b/tests/integration/manual_test_temporal_integration.py index 8e2af4e91b..7bc4e3f782 100644 --- a/tests/integration/manual_test_temporal_integration.py +++ b/tests/integration/manual_test_temporal_integration.py @@ -221,7 +221,6 @@ async def run(self, prompt: str) -> str: # Turn 2: Trigger Handoff logger.info("--- Turn 2: Trigger Handoff ---") handoff_response_text = "" - last_author = "" async with Aclosing(runner.run_async( user_id=session.user_id, session_id=session.id, @@ -229,7 +228,6 @@ async def run(self, prompt: str) -> str: )) as agen: async for event in agen: logger.info(f"Event Author: {event.author} | Actions: {event.actions}") - last_author = event.author if event.content and event.content.parts: for part in event.content.parts: if part.text: handoff_response_text += part.text From 389f1169d15ddf6b0cd69a25455dd08858069294 Mon Sep 17 00:00:00 2001 From: Marcus Motill Date: Sat, 17 Jan 2026 19:23:08 +0000 Subject: [PATCH 4/7] update to plugin strategy --- src/google/adk/integrations/temporal.py | 155 ---------- .../adk/integrations/temporal/__init__.py | 284 +++++++++++++++++ tests/integration/README_TEMPORAL.md | 23 -- .../manual_test_temporal_integration.py | 286 ++++++------------ tests/unittests/integrations/test_temporal.py | 123 +++++--- 5 files changed, 455 insertions(+), 416 deletions(-) delete mode 100644 src/google/adk/integrations/temporal.py create mode 100644 src/google/adk/integrations/temporal/__init__.py delete mode 100644 tests/integration/README_TEMPORAL.md diff --git a/src/google/adk/integrations/temporal.py b/src/google/adk/integrations/temporal.py deleted file mode 100644 index d05145b5ac..0000000000 --- a/src/google/adk/integrations/temporal.py +++ /dev/null @@ -1,155 +0,0 @@ -# Copyright 2025 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Temporal integration helpers for ADK.""" - -import functools -import inspect -from typing import Any, AsyncGenerator, Callable, Optional, List - -from temporalio import workflow, activity -from google.adk.models import BaseLlm, LlmRequest, LlmResponse, LLMRegistry -from google.genai import types - - -def activity_as_tool( - activity_def: Callable, - **activity_options: Any -) -> Callable: - """Wraps a Temporal Activity Definition into an ADK-compatible tool. - - Args: - activity_def: The Temporal activity definition (decorated with @activity.defn). - **activity_options: Options to pass to workflow.execute_activity - (e.g. start_to_close_timeout, retry_policy). - - Returns: - A callable tool that executes the activity when invoked. - """ - - # We create a wrapper that delegates to workflow.execute_activity - async def tool_wrapper(*args, **kwargs) -> Any: - # Bind arguments to the activity signature to ensure correct order/mapping. - try: - sig = inspect.signature(activity_def) - bound_args = sig.bind(*args, **kwargs) - bound_args.apply_defaults() - activity_args = [bound_args.arguments[p] for p in sig.parameters] - except (TypeError, ValueError): - # Fallback for built-ins or other complex callables where binding may fail - # or if arguments don't match signature (e.g. simpler invocation). - # Temporal Python SDK typically invokes activities with `args=[...]`. - activity_args = list(args) + list(kwargs.values()) if kwargs else list(args) - - return await workflow.execute_activity( - activity_def, - args=activity_args, - **activity_options - ) - - # Copy metadata so ADK can inspect the tool (name, docstring, annotations) - # ADK uses this to generate the tool schema for the LLM. - tool_wrapper.__doc__ = activity_def.__doc__ - tool_wrapper.__name__ = getattr(activity_def, "name", activity_def.__name__) - - # Attempt to copy annotations if they exist - if hasattr(activity_def, "__annotations__"): - tool_wrapper.__annotations__ = activity_def.__annotations__ - - # CRITICAL: Copy signature so FunctionTool can generate correct parameters schema - try: - tool_wrapper.__signature__ = inspect.signature(activity_def) - except Exception: - pass # Fallback if signature copy fails (e.g. builtins) - - return tool_wrapper - - -@activity.defn -async def generate_content_activity(request: LlmRequest) -> List[LlmResponse]: - """Generic activity to invoke an LLM via ADK's LLMRegistry. - - The model name is expected to be in `request.model`. - """ - if not request.model: - raise ValueError("LlmRequest.model must be set when using generate_content_activity.") - - llm = LLMRegistry.new_llm(request.model) - return [response async for response in llm.generate_content_async(request)] - - -class TemporalModel(BaseLlm): - """An ADK ModelWrapper that executes content generation as a Temporal Activity. - - This effectively delegates the 'generate_content' call to an external Activity, - ensuring that the network I/O to Vertex/Gemini is recorded in Temporal history. - """ - - activity_def: Callable - activity_options: dict[str, Any] - - def __init__( - self, - model_name: str, - activity_def: Callable = generate_content_activity, - **activity_options: Any - ): - """Initializes the TemporalModel. - - Args: - model_name: The name of the model to report to ADK. - activity_def: The Temporal activity definition to invoke. - Defaults to `generate_content_activity`. - **activity_options: Options for workflow.execute_activity. - """ - super().__init__( - model=model_name, - activity_def=activity_def, - activity_options=activity_options - ) - - async def generate_content_async( - self, - llm_request: LlmRequest, - stream: bool = False - ) -> AsyncGenerator[LlmResponse, None]: - """Generates content by calling the configured Temporal Activity.""" - - # Ensure model name is carried in the request for the generic activity - if not llm_request.model: - llm_request.model = self.model - - # Note: Temporal Activities are not typically streaming in the Python SDK - # in the way python async generators work (streaming back to workflow is complex). - # Standard approach is to return the full response. - # We will assume non-streaming activity execution for now. - - # Execute the activity - responses: List[LlmResponse] = await workflow.execute_activity( - self.activity_def, - args=[llm_request], - **self.activity_options - ) - - # Yield the responses - for response in responses: - yield response - - @classmethod - def default_activities(cls) -> List[Callable]: - """Returns the default activities used by this model wrapper. - - Useful for registering activities with the Temporal Worker. - """ - return [generate_content_activity] diff --git a/src/google/adk/integrations/temporal/__init__.py b/src/google/adk/integrations/temporal/__init__.py new file mode 100644 index 0000000000..35ac267860 --- /dev/null +++ b/src/google/adk/integrations/temporal/__init__.py @@ -0,0 +1,284 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Temporal Integration for ADK. + +This module provides the necessary components to run ADK Agents within Temporal Workflows. +""" + +from __future__ import annotations + +import uuid +import dataclasses +import inspect +import functools +import time +import asyncio +from datetime import timedelta +from typing import Callable, Any, Optional, List, AsyncGenerator +from collections.abc import Sequence + +from temporalio import workflow, activity +from temporalio.common import RetryPolicy, RawValue +from temporalio.worker import WorkflowRunner +from temporalio.worker import UnsandboxedWorkflowRunner +from temporalio.converter import DataConverter, DefaultPayloadConverter +from temporalio.plugin import SimplePlugin +from temporalio.contrib.pydantic import PydanticPayloadConverter as _DefaultPydanticPayloadConverter + +from google.adk.plugins import BasePlugin +from google.adk.models import LLMRegistry, BaseLlm, LlmRequest, LlmResponse +from google.adk.agents.invocation_context import InvocationContext +from temporalio.worker import ( + WorkflowInboundInterceptor, + Interceptor, + ExecuteWorkflowInput, + WorkflowInterceptorClassInput +) +from google.adk.agents.callback_context import CallbackContext +from google.genai import types + + + + +def setup_deterministic_runtime(): + """Configures ADK runtime for Temporal determinism. + + This should be called at the start of a Temporal Workflow before any ADK components + (like SessionService) are used, if they rely on runtime.get_time() or runtime.new_uuid(). + """ + try: + from google.adk import runtime + + # Define safer, context-aware providers + def _deterministic_time_provider() -> float: + if workflow.in_workflow(): + return workflow.now().timestamp() + return time.time() # Fallback to system time + + def _deterministic_id_provider() -> str: + if workflow.in_workflow(): + return str(workflow.uuid4()) + return str(uuid.uuid4()) # Fallback to system UUID + + runtime.set_time_provider(_deterministic_time_provider) + runtime.set_id_provider(_deterministic_id_provider) + except ImportError: + pass + except Exception as e: + print(f"Warning: Failed to set deterministic runtime providers: {e}") + +class AdkWorkflowInboundInterceptor(WorkflowInboundInterceptor): + async def execute_workflow(self, input: ExecuteWorkflowInput) -> Any: + # Global runtime setup before ANY user code runs + setup_deterministic_runtime() + return await super().execute_workflow(input) + +class AdkInterceptor(Interceptor): + def workflow_interceptor_class( + self, input: WorkflowInterceptorClassInput + ) -> type[WorkflowInboundInterceptor] | None: + return AdkWorkflowInboundInterceptor + +class TemporalPlugin(BasePlugin): + """ADK Plugin for Temporal integration. + + This plugin automatically configures the ADK runtime to be deterministic when running + inside a Temporal workflow, and intercepts model calls to execute them as Temporal Activities. + """ + + def __init__(self, activity_options: Optional[dict[str, Any]] = None): + """Initializes the Temporal Plugin. + + Args: + activity_options: Default options for model activities (e.g. start_to_close_timeout). + """ + super().__init__(name="temporal_plugin") + self.activity_options = activity_options or {} + + @staticmethod + def activity_tool(activity_def: Callable, **kwargs: Any) -> Callable: + """Decorator/Wrapper to wrap a Temporal Activity as an ADK Tool. + + 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'. + """ + async def wrapper(*args, **kw): + # Inspect signature to bind arguments + sig = inspect.signature(activity_def) + bound = sig.bind(*args, **kw) + bound.apply_defaults() + + # Convert to positional args for Temporal + activity_args = list(bound.arguments.values()) + + # Strategy: Decorator kwargs are defaults. + options = kwargs.copy() + + # Assert workflow import is available or mocked + return await workflow.execute_activity( + activity_def, + *activity_args, + **options + ) + + # Copy metadata + wrapper.__name__ = activity_def.__name__ + wrapper.__doc__ = activity_def.__doc__ + wrapper.__signature__ = inspect.signature(activity_def) + + return wrapper + + @staticmethod + @activity.defn(dynamic=True) + async def dynamic_activity(args: Sequence[RawValue]) -> Any: + """Handles dynamic ADK activities (e.g. 'AgentName.generate_content').""" + activity_type = activity.info().activity_type + + # Check if this is a generate_content call + if activity_type.endswith(".generate_content") or activity_type == "google.adk.generate_content": + return await TemporalPlugin._handle_generate_content(args) + + raise ValueError(f"Unknown dynamic activity: {activity_type}") + + @staticmethod + async def _handle_generate_content(args: List[Any]) -> list[dict[str, Any]]: + """Implementation of content generation.""" + # 1. Decode Arguments + # Dynamic activities receive RawValue wrappers (which host the Payload). + # We must manually decode them using the activity's configured data converter. + converter = activity.payload_converter() + + # We expect a single argument: LlmRequest + if not args: + raise ValueError("Missing llm_request argument for generate_content") + + # Extract payloads from RawValue wrappers + payloads = [arg.payload for arg in args] + + # Decode + # from_payloads returns a list of decoded objects. + # We specify the types we expect for each argument. + try: + decoded_args = converter.from_payloads(payloads, [LlmRequest]) + llm_request: LlmRequest = decoded_args[0] + except Exception as e: + activity.logger.error(f"Failed to decode arguments: {e}") + raise ValueError(f"Argument decoding failed: {e}") from e + + # 3. Model Initialization + llm = LLMRegistry.new_llm(llm_request.model) + if not llm: + raise ValueError(f"Failed to create LLM for model: {llm_request.model}") + + # 4. Execution + responses = [response async for response in llm.generate_content_async(llm_request=llm_request)] + + # 5. Serialization + # Return dicts to avoid Pydantic strictness issues on rehydration + return [ + r.model_dump(mode='json', by_alias=True) + for r in responses + ] + + + async def before_model_callback( + self, *, callback_context: CallbackContext, llm_request: LlmRequest + ) -> LlmResponse | None: + # If already in a workflow, execute the activity + if workflow.in_workflow(): + # Ensure model is set from agent if missing in request + if not llm_request.model: + if isinstance(callback_context.invocation_context.agent.model, str): + llm_request.model = callback_context.invocation_context.agent.model + elif hasattr(callback_context.invocation_context.agent.model, 'model_name'): + llm_request.model = callback_context.invocation_context.agent.model.model_name + + # Default options + options = { + "start_to_close_timeout": timedelta(seconds=60), + "retry_policy": RetryPolicy( + initial_interval=timedelta(seconds=1), + backoff_coefficient=2.0, + maximum_interval=timedelta(seconds=30), + maximum_attempts=5 + ) + } + # Merge with user options + options.update(self.activity_options) + + # Execution + # The activity returns list[dict] to avoid strict Pydantic validation issues. + + # Construct dynamic activity name for visibility + agent_name = callback_context.agent_name + activity_name = f"{agent_name}.generate_content" + + # Debug options + activity.logger.info(f"Executing activity '{activity_name}' with options: {options}") + + # Execute with dynamic name + response_dicts = await workflow.execute_activity( + activity_name, + args=[llm_request], + **options + ) + + # Rehydrate LlmResponse objects safely + responses = [] + for d in response_dicts: + try: + responses.append(LlmResponse.model_validate(d)) + except Exception as e: + raise RuntimeError(f"Failed to deserialized LlmResponse from activity result: {e}") from e + + # Simple consolidation: return the last complete response + return responses[-1] if responses else None + + return None + + + + +class AdkWorkerPlugin(SimplePlugin): + """A Temporal Worker Plugin configured for ADK. + + This plugin configures: + 1. Pydantic Payload Converter (required for ADK objects). + 2. Sandbox Passthrough for `google.adk` and `google.genai`. + """ + def __init__(self): + super().__init__( + name="adk_worker_plugin", + data_converter=self._configure_data_converter, + workflow_runner=self._configure_workflow_runner, + worker_interceptors=[AdkInterceptor()] + ) + + def _configure_data_converter(self, converter: DataConverter | None) -> DataConverter: + if converter is None: + # Create a default converter using our PydanticPayloadConverter + return DataConverter( + payload_converter_class=_DefaultPydanticPayloadConverter + ) + elif converter.payload_converter_class is DefaultPayloadConverter: + return dataclasses.replace( + converter, payload_converter_class=_DefaultPydanticPayloadConverter + ) + return converter + + def _configure_workflow_runner(self, runner: WorkflowRunner | None) -> WorkflowRunner: + from temporalio.worker import UnsandboxedWorkflowRunner + # TODO: Not sure implications here. is this a good default an allow user override? + return UnsandboxedWorkflowRunner() diff --git a/tests/integration/README_TEMPORAL.md b/tests/integration/README_TEMPORAL.md deleted file mode 100644 index 31a9c3855e..0000000000 --- a/tests/integration/README_TEMPORAL.md +++ /dev/null @@ -1,23 +0,0 @@ -# Temporal Integration Tests - -The file `manual_test_temporal_integration.py` contains integration tests for ADK's Temporal support. -It is named `manual_test_...` to be excluded from standard CI/test runs because it requires: - -1. **Local Temporal Server**: You must have a Temporal server running locally (e.g., via `temporal server start-dev`). -2. **GCP Credentials**: Environment variables `GOOGLE_CLOUD_PROJECT` and `GOOGLE_CLOUD_LOCATION` must be set. -3. **Local Environment**: It assumes `localhost:7233`. - -## How to Run - -1. Start Temporal Server: - ```bash - temporal server start-dev - ``` - -2. Run the test directly: - ```bash - export GOOGLE_CLOUD_PROJECT="your-project" - export GOOGLE_CLOUD_LOCATION="us-central1" - - uv run pytest tests/integration/manual_test_temporal_integration.py - ``` diff --git a/tests/integration/manual_test_temporal_integration.py b/tests/integration/manual_test_temporal_integration.py index 7bc4e3f782..273d1fb397 100644 --- a/tests/integration/manual_test_temporal_integration.py +++ b/tests/integration/manual_test_temporal_integration.py @@ -19,7 +19,7 @@ import uuid import os from datetime import timedelta -from typing import AsyncGenerator, List +from typing import AsyncGenerator import pytest from temporalio import workflow, activity @@ -39,81 +39,63 @@ from google.adk.sessions import InMemorySessionService from google.adk.utils.context_utils import Aclosing from google.adk.events import Event -from google.adk.integrations.temporal import activity_as_tool, TemporalModel, generate_content_activity +from google.adk.integrations.temporal import AdkWorkerPlugin, TemporalPlugin # Required Environment Variables for this test: -# - GOOGLE_CLOUD_PROJECT -# - GOOGLE_CLOUD_LOCATION -# - GOOGLE_GENAI_USE_VERTEXAI (optional, defaults to 1 for this test if needed, or set externally) -# - Temporal Server running at localhost:7233 +# in this folder update .env.example to be .env and have the following vars: +# GOOGLE_GENAI_USE_VERTEXAI=1 +# GOOGLE_CLOUD_PROJECT="" +# GOOGLE_CLOUD_LOCATION="" +# TEST_BACKEND=VERTEX_ONLY +# then: +# start temporal: temporal server start-dev +# then: +# uv run pytest tests/integration/manual_test_temporal_integration.py + logger = logging.getLogger(__name__) @activity.defn async def get_weather(city: str) -> str: - """Activity that gets weather.""" + """Activity that gets weather for a given city.""" return "Warm and sunny. 17 degrees." -# --- Customized LLM Activities for Better Trace Visibility --- - -@activity.defn(name="coordinator_think") -async def coordinator_think(req: LlmRequest) -> List[LlmResponse]: - """Activity for the Coordinator agent.""" - return await generate_content_activity(req) - -@activity.defn(name="tool_agent_think") -async def tool_agent_think(req: LlmRequest) -> List[LlmResponse]: - """Activity for the Tool Agent.""" - return await generate_content_activity(req) - -@activity.defn(name="specialist_think") -async def specialist_think(req: LlmRequest) -> List[LlmResponse]: - """Activity for the Specialist/Handoff Agent.""" - return await generate_content_activity(req) - - @workflow.defn class WeatherAgent: @workflow.run async def run(self, prompt: str) -> Event | None: logger.info("Workflow started.") - # 1. Configure ADK Runtime to use Temporal Determinism - runtime.set_time_provider(lambda: workflow.now().timestamp()) - runtime.set_id_provider(lambda: str(workflow.uuid4())) - - # 2. Define Agent using Temporal Helpers - # Uses generic 'generate_content_activity' by default - agent_model = TemporalModel( - model_name="gemini-2.5-pro", - start_to_close_timeout=timedelta(minutes=2) - ) + # 1. Define Agent using Temporal Helpers + # Note: TemporalPlugin in the Runner automatically handles Runtime setup + # and Model Activity interception. We use standard ADK models now. # Wraps 'get_weather' activity as a Tool - weather_tool = activity_as_tool( + weather_tool = TemporalPlugin.activity_tool( get_weather, start_to_close_timeout=timedelta(seconds=60) ) agent = Agent( name='test_agent', - model=agent_model, + model="gemini-2.5-pro", # Standard model string tools=[weather_tool] ) - # 3. Create Session (uses runtime.new_uuid() -> workflow.uuid4()) + # 2. Create Session (uses runtime.new_uuid() -> workflow.uuid4()) session_service = InMemorySessionService() logger.info("Create session.") session = await session_service.create_session(app_name="test_app", user_id="test") logger.info(f"Session created with ID: {session.id}") - # 4. Run Agent + # 3. Run Agent with TemporalPlugin runner = Runner( agent=agent, app_name='test_app', session_service=session_service, + plugins=[TemporalPlugin(activity_options={'start_to_close_timeout': timedelta(minutes=2)})] ) logger.info("Starting runner.") @@ -134,186 +116,108 @@ async def run(self, prompt: str) -> Event | None: @workflow.defn class MultiAgentWorkflow: @workflow.run - async def run(self, prompt: str) -> str: - # 1. Runtime Setup - runtime.set_time_provider(lambda: workflow.now().timestamp()) - runtime.set_id_provider(lambda: str(workflow.uuid4())) - - # 2. Define Distinct Models for Visualization - # We use a separate activity for each agent so they show up distinctly in the Temporal UI. + async def run(self, topic: str) -> str: + # Example of multi-turn/multi-agent orchestration + # This is where Temporal shines - orchestrating complex agent flows - coordinator_model = TemporalModel( - model_name="gemini-2.5-pro", - activity_def=coordinator_think, - start_to_close_timeout=timedelta(minutes=2) - ) - - tool_agent_model = TemporalModel( - model_name="gemini-2.5-pro", - activity_def=tool_agent_think, - start_to_close_timeout=timedelta(minutes=2) - ) + # 0. Deterministic Runtime is now auto-configured by AdkInterceptor! + + # 1. Setup Session Service + session_service = InMemorySessionService() + session = await session_service.create_session(app_name="multi_agent_app", user_id="test_user") - specialist_model = TemporalModel( - model_name="gemini-2.5-pro", - activity_def=specialist_think, - start_to_close_timeout=timedelta(minutes=2) + # 2. Define Agents + # Sub-agent: Researcher + researcher = LlmAgent( + name="researcher", + model="gemini-2.5-pro", + instruction="You are a researcher. Find information about the topic." ) - - # 3. Define Sub-Agents - # Agent to be used as a Tool - tool_agent = LlmAgent( - name="ToolAgent", - model=tool_agent_model, - instruction="You are a tool agent. You help with specific sub-tasks. Always include 'From ToolAgent:' in your response." + # Sub-agent: Writer + writer = LlmAgent( + name="writer", + model="gemini-2.5-pro", + instruction="You are a poet. Write a haiku based on the research." ) - agent_tool = AgentTool(tool_agent) - # Agent to be transferred to (Handoff) - handoff_agent = LlmAgent( - name="HandoffAgent", - model=specialist_model, - instruction="You are a Specialist Agent. You handle specialized requests. Always include 'From HandoffAgent:' in your response." - ) - - # 4. Define Parent Agent - parent_agent = LlmAgent( - name="Coordinator", - model=coordinator_model, - # Instructions to guide the LLM when to use which - instruction=( - "You are a Coordinator. " - "CRITICAL INSTRUCTION: You MUST NOT answer user queries directly if they related to specific tasks. " - "1. If the user asks for 'help' or 'subtask', you MUST use the 'ToolAgent' tool (AgentTool). " - "2. If the user asks to 'switch' or 'specialist', you MUST transfer to the HandoffAgent using 'transfer_to_agent'. " - "Do not apologize. Do not say you will do it. Just call the function." - ), - tools=[agent_tool], - sub_agents=[handoff_agent] + # Root Agent: Coordinator + coordinator = LlmAgent( + name="coordinator", + model="gemini-2.5-pro", + instruction="You are a coordinator. Delegate to researcher then writer.", + sub_agents=[researcher, writer] ) - # 5. Execute - session_service = InMemorySessionService() - session = await session_service.create_session(app_name="multi_agent_app", user_id="user_MULTI") - + # 3. Initialize Runner with required args runner = Runner( - agent=parent_agent, - app_name='multi_agent_app', + agent=coordinator, + app_name="multi_agent_app", session_service=session_service, + plugins=[TemporalPlugin()] ) - - # We will run a multi-turn conversation to test both paths - # Turn 1: Trigger Tool - logger.info("--- Turn 1: Trigger Tool ---") - tool_response_text = "" - async with Aclosing(runner.run_async( - user_id=session.user_id, - session_id=session.id, - new_message=types.Content(role='user', parts=[types.Part(text="I need help with a subtask.")]) - )) as agen: - async for event in agen: - logger.info(f"Event Author: {event.author} | Actions: {event.actions}") - if event.content and event.content.parts: - for part in event.content.parts: - if part.text: tool_response_text += part.text - - # Turn 2: Trigger Handoff - logger.info("--- Turn 2: Trigger Handoff ---") - handoff_response_text = "" - async with Aclosing(runner.run_async( - user_id=session.user_id, - session_id=session.id, - new_message=types.Content(role='user', parts=[types.Part(text="Please switch me to the specialist.")]) - )) as agen: - async for event in agen: - logger.info(f"Event Author: {event.author} | Actions: {event.actions}") - if event.content and event.content.parts: - for part in event.content.parts: - if part.text: handoff_response_text += part.text - - logger.info(f"Tool Response: {tool_response_text}") - logger.info(f"Handoff Response: {handoff_response_text}") - return f"Tool: {tool_response_text} | Handoff: {handoff_response_text}" - - -class ADKPlugin(SimplePlugin): - def __init__(self): - super().__init__( - name="ADKPlugin", - data_converter=_data_converter, - workflow_runner=workflow_runner, - ) + # 4. Run + # Note: In a real temporal app, we might signal the workflow or use queries. + # Here we just run a single turn for the test. + final_content = "" + user_msg = types.Content(role="user", parts=[types.Part(text=f"Write a haiku about {topic}. First research it, then write it.")]) + async for event in runner.run_async( + user_id="test_user", + session_id=session.id, + new_message=user_msg + ): + if event.content and event.content.parts: + final_content = event.content.parts[0].text + + return final_content -def workflow_runner(runner: WorkflowRunner | None) -> WorkflowRunner: - if not runner: - raise ValueError("No WorkflowRunner provided to the ADK plugin.") - - # If in sandbox, add additional passthrough - if isinstance(runner, SandboxedWorkflowRunner): - return dataclasses.replace( - runner, - restrictions=runner.restrictions.with_passthrough_modules("google.adk", "google.genai"), - ) - return runner - -def _data_converter(converter: DataConverter | None) -> DataConverter: - if converter is None: - return pydantic_data_converter - elif converter.payload_converter_class is DefaultPayloadConverter: - return dataclasses.replace( - converter, payload_converter_class=PydanticPayloadConverter - ) - elif not isinstance(converter.payload_converter, PydanticPayloadConverter): - raise ValueError( - "The payload converter must be of type PydanticPayloadConverter." - ) - return converter @pytest.mark.asyncio -async def test_temporalio_integration(): - """Run full integration test with Temporal Server.""" +async def test_temporal_integration(): + """Manual integration test requiring a running Temporal server.""" - # Normally this should only run if local Temporal server is available - # For now, we assume it is, as per user context. + # 1. Start a Worker (in a real app, this would be a separate process) + # We run it here for the test. - # Start client/worker - if "GOOGLE_CLOUD_PROJECT" not in os.environ: - pytest.skip("GOOGLE_CLOUD_PROJECT not set. Skipping integration test.") - try: - client = await Client.connect("localhost:7233", plugins=[ADKPlugin()]) + # Connect to Temporal Server + # We must configure the data converter to handle Pydantic models (like Event) + client = await Client.connect( + "localhost:7233", + data_converter=DataConverter(payload_converter_class=PydanticPayloadConverter) + ) except RuntimeError: pytest.skip("Could not connect to Temporal server. Is it running?") + # Run Worker with the ADK plugin async with Worker( client, - workflows=[WeatherAgent, MultiAgentWorkflow], - activities=TemporalModel.default_activities() + [ + task_queue="adk-task-queue", + activities=[ get_weather, - coordinator_think, - tool_agent_think, - specialist_think + TemporalPlugin.dynamic_activity, + # Note: We can add specific wrapper activities if we want distinct names in UI, + # but generate_content is the generic one used by TemporalPlugin. ], - task_queue="hello_world_queue", - max_cached_workflows=0, - ) as worker: + workflows=[WeatherAgent, MultiAgentWorkflow], + plugins=[AdkWorkerPlugin()] # <--- Use the class based plugin + ): print("Worker started.") - # Run Weather Agent - result_weather = await client.execute_workflow( + # Test Weather Agent + result = await client.execute_workflow( WeatherAgent.run, - "What is the weather in Tokyo?", - id=str(uuid.uuid4()), - task_queue="hello_world_queue", + "What is the weather in New York?", + id=f"weather-agent-workflow-{uuid.uuid4()}", + task_queue="adk-task-queue", ) - print(f"Weather Agent Result: {result_weather}") - - # Run Multi-Agent Workflow + print(f"Workflow result: {result}") + + # Test Multi Agent result_multi = await client.execute_workflow( MultiAgentWorkflow.run, - "start", # Argument ignored in run logic (hardcoded prompts) - id=str(uuid.uuid4()), - task_queue="hello_world_queue", + "Run mult-agent flow", + id=f"multi-agent-workflow-{uuid.uuid4()}", + task_queue="adk-task-queue", ) - print(f"Multi-Agent Result: {result_multi}") + print(f"Multi-Agent Workflow result: {result_multi}") diff --git a/tests/unittests/integrations/test_temporal.py b/tests/unittests/integrations/test_temporal.py index ec7fed9e6a..8508b92e67 100644 --- a/tests/unittests/integrations/test_temporal.py +++ b/tests/unittests/integrations/test_temporal.py @@ -28,9 +28,13 @@ mock_activity = MagicMock() mock_worker = MagicMock() mock_client = MagicMock() +mock_converter = MagicMock() # Important: execute_activity must be awaitable mock_workflow.execute_activity = AsyncMock(return_value="mock_result") +mock_workflow.in_workflow = MagicMock(return_value=False) +mock_workflow.now = MagicMock() +mock_workflow.uuid4 = MagicMock() # Mock the parent package mock_temporalio = MagicMock() @@ -38,18 +42,43 @@ mock_temporalio.activity = mock_activity mock_temporalio.worker = mock_worker mock_temporalio.client = mock_client +mock_temporalio.converter = mock_converter +class FakeSimplePlugin: + def __init__(self, **kwargs): + pass +mock_temporalio.plugin = MagicMock() +mock_temporalio.plugin.SimplePlugin = FakeSimplePlugin +mock_temporalio.worker.workflow_sandbox = MagicMock() +mock_temporalio.contrib = MagicMock() +mock_temporalio.contrib.pydantic = MagicMock() # Mock sys.modules +# Mock sys.modules +# We must ensure we get a fresh import of 'google.adk.integrations.temporal' +# that uses our MOCKED 'temporalio'. +# If it was already loaded, we remove it. +for mod in list(sys.modules.keys()): + if mod.startswith("google.adk") or mod == "temporalio": + del sys.modules[mod] + with patch.dict(sys.modules, { "temporalio": mock_temporalio, "temporalio.workflow": mock_workflow, "temporalio.activity": mock_activity, "temporalio.worker": mock_worker, "temporalio.client": mock_client, + "temporalio.converter": mock_converter, + "temporalio.common": MagicMock(), + "temporalio.plugin": mock_temporalio.plugin, + "temporalio.worker.workflow_sandbox": mock_temporalio.worker.workflow_sandbox, + "temporalio.contrib": mock_temporalio.contrib, + "temporalio.contrib.pydantic": mock_temporalio.contrib.pydantic, }): from google.adk.integrations import temporal from google.adk.models import LlmRequest, LlmResponse - + from google.adk.agents.invocation_context import InvocationContext + from google.adk.agents.callback_context import CallbackContext + from google.adk import runtime class TestTemporalIntegration(unittest.TestCase): @@ -59,85 +88,85 @@ def test_activity_as_tool_wrapper(self): mock_workflow.execute_activity = AsyncMock(return_value="mock_result") # Verify mock setup - # If this fails, then 'temporal.workflow' is NOT our 'mock_workflow' assert temporal.workflow.execute_activity is mock_workflow.execute_activity # Define a fake activity - async def fake_activity(arg: str) -> str: + async def my_activity(arg: str) -> str: """My Docstring.""" return f"Hello {arg}" - fake_activity.name = "fake_activity_name" - - # Create tool - tool = temporal.activity_as_tool( - fake_activity, - start_to_close_timeout=100 + # Wrap it + tool = temporal.TemporalPlugin.activity_tool( + my_activity, + start_to_close_timeout=100 ) # Check metadata - self.assertEqual(tool.__name__, "fake_activity_name") + self.assertEqual(tool.__name__, "my_activity") # Matches function name self.assertEqual(tool.__doc__, "My Docstring.") # Run tool (wrapper) loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - result = loop.run_until_complete(tool("World")) + asyncio.set_event_loop(loop) + loop.run_until_complete(tool(arg="World")) finally: loop.close() # Verify call mock_workflow.execute_activity.assert_called_once() args, kwargs = mock_workflow.execute_activity.call_args - self.assertEqual(kwargs['args'], ['World']) + self.assertEqual(args[1], 'World') self.assertEqual(kwargs['start_to_close_timeout'], 100) - def test_temporal_model_generate_content(self): - # Reset mocks - mock_workflow.reset_mock() + + + def test_temporal_plugin_before_model(self): + plugin = temporal.TemporalPlugin(activity_options={"start_to_close_timeout": 60}) - # Prepare valid LlmResponse with content - response_content = types.Content(parts=[types.Part(text="test_resp")]) + # Setup mocks + mock_workflow.reset_mock() + mock_workflow.in_workflow.return_value = True + response_content = types.Content(parts=[types.Part(text="plugin_resp")]) llm_response = LlmResponse(content=response_content) + # The plugin now expects the activity to return dicts (model_dump(mode='json')) + # to ensure safe deserialization across process boundaries. + response_dict = llm_response.model_dump(mode='json', by_alias=True) + # Ensure 'content' key is present and correct (pydantic dump might be complex) + # For the test simple case, the dump is sufficient. - # generate_content_async expects execute_activity to return response list (iterator) - mock_workflow.execute_activity = AsyncMock(return_value=[llm_response]) + mock_workflow.execute_activity = AsyncMock(return_value=[response_dict]) - # Mock an activity def - mock_activity_def = MagicMock() + # callback_context = MagicMock(spec=CallbackContext) + # Using spec might hide dynamic attributes or properties if not fully mocked + callback_context = MagicMock() + callback_context.agent_name = "test-agent" + callback_context.invocation_context.agent.model = "test-agent-model" - # Create model - model = temporal.TemporalModel( - model_name="test-model", - activity_def=mock_activity_def, - schedule_to_close_timeout=50 - ) - - # Create request - req = LlmRequest(model="test-model", prompt="hi") + llm_request = LlmRequest(model=None, prompt="hi") - # Run generate_content_async (it is an async generator) - async def run_gen(): - results = [] - async for r in model.generate_content_async(req): - results.append(r) - return results - + # Run callback loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) - try: - results = loop.run_until_complete(run_gen()) + result = loop.run_until_complete(plugin.before_model_callback( + callback_context=callback_context, + llm_request=llm_request + )) finally: loop.close() - - # Verify execute_activity called + + # Verify execution mock_workflow.execute_activity.assert_called_once() args, kwargs = mock_workflow.execute_activity.call_args - self.assertEqual(kwargs['args'], [req]) - self.assertEqual(kwargs['schedule_to_close_timeout'], 50) - self.assertEqual(len(results), 1) - self.assertEqual(results[0].content.parts[0].text, "test_resp") + self.assertEqual(kwargs['start_to_close_timeout'], 60) + + # Check dynamic activity name + self.assertEqual(args[0], "test-agent.generate_content") + self.assertEqual(kwargs['args'][0].model, "test-agent-model") + + # Verify result merge + self.assertIsNotNone(result) + # Result is re-hydrated LlmResponse + self.assertEqual(result.content.parts[0].text, "plugin_resp") From b90ee97cfee6aa5af816b9bf1fc0c2e017b72544 Mon Sep 17 00:00:00 2001 From: Marcus Motill Date: Sun, 18 Jan 2026 01:09:35 +0000 Subject: [PATCH 5/7] add readme and updates --- .../adk/integrations/temporal/README.md | 81 +++++++++++++++++++ .../adk/integrations/temporal/__init__.py | 78 ++++++------------ 2 files changed, 104 insertions(+), 55 deletions(-) create mode 100644 src/google/adk/integrations/temporal/README.md diff --git a/src/google/adk/integrations/temporal/README.md b/src/google/adk/integrations/temporal/README.md new file mode 100644 index 0000000000..56a18f2955 --- /dev/null +++ b/src/google/adk/integrations/temporal/README.md @@ -0,0 +1,81 @@ +# ADK Temporal Integration Internals + +This package provides the integration layer between the Google ADK and Temporal. It allows ADK Agents to run reliably within Temporal Workflows by ensuring determinism and correctly routing external calls (network I/O) through Temporal Activities. + +## Core Concepts + +### 1. Interception Flow (`TemporalPlugin`) + +The `TemporalPlugin` acts as a middleware that intercepts model calls (e.g., `agent.generate_content`) *before* they execute. + +**Workflow Interception:** +1. **Intercept**: The ADK invokes `before_model_callback` when an agent attempts to call a model. +2. **Delegate**: The plugin calls `workflow.execute_activity()`, routing the request to Temporal for execution. +3. **Return**: The plugin awaits the activity result and returns it immediately. The ADK stops its own request processing, using the activity result as the final response. + +This ensures that all model interactions are recorded in the Temporal Workflow history, enabling reliable replay and determinism. + +### 2. Dynamic Activity Registration + +To provide visibility in the Temporal UI, activities are dynamically named after the calling agent (e.g., `MyAgent.generate_content`). Since agent names are not known at startup, the integration uses Temporal's dynamic activity registration. + +```python +@activity.defn(dynamic=True) +async def dynamic_activity(args: Sequence[RawValue]) -> Any: + ... +``` + +When the workflow executes an activity with an unknown name (e.g., `MyAgent.generate_content`), the worker routes the call to `dynamic_activity`. This handler inspects the `activity_type` and delegates execution to the appropriate internal logic (`_handle_generate_content`), enabling arbitrary activity names without explicit registration. + +### 3. Usage & Configuration + +The integration requires setup on both the Agent (Workflow) side and the Worker side. + +#### Agent Setup (Workflow Side) +Attach the `TemporalPlugin` to your ADK agent. This safely routes model calls through Temporal activities. You **must** provide activity options (e.g., timeouts) as there are no defaults. + +```python +from datetime import timedelta +from temporalio.common import RetryPolicy +from google.adk.integrations.temporal import TemporalPlugin + +# 1. Define Temporal Activity Options +activity_options = { + "start_to_close_timeout": timedelta(minutes=1), + "retry_policy": RetryPolicy(maximum_attempts=3) +} + +# 2. Add Plugin to Agent +agent = Agent( + model="gemini-2.5-pro", + plugins=[ + # Routes model calls to Temporal Activities + TemporalPlugin(activity_options=activity_options) + ] +) + +# 3. Use Agent in Workflow +# When agent.generate_content() is called, it will execute as a Temporal Activity. +``` + +#### Worker Setup +Install the `AdkWorkerPlugin` on your Temporal Worker. This handles serialization and runtime determinism. + +```python +from temporalio.worker import Worker +from google.adk.integrations.temporal import AdkWorkerPlugin + +async def main(): + worker = Worker( + client, + task_queue="my-queue", + # Configures ADK Runtime & Pydantic Support + plugins=[AdkWorkerPlugin()] + ) + await worker.run() +``` + +**What `AdkWorkerPlugin` Does:** +* **Data Converter**: Enables Pydantic serialization for ADK objects. +* **Interceptors**: Sets up specific ADK runtime hooks for determinism (replacing `time.time`, `uuid.uuid4`) before workflow execution. +* TODO: is this enough . **Unsandboxed Workflow Runner**: Configures the worker to use the `UnsandboxedWorkflowRunner`, allowing standard imports in ADK agents. diff --git a/src/google/adk/integrations/temporal/__init__.py b/src/google/adk/integrations/temporal/__init__.py index 35ac267860..2baf56c96c 100644 --- a/src/google/adk/integrations/temporal/__init__.py +++ b/src/google/adk/integrations/temporal/__init__.py @@ -65,12 +65,12 @@ def setup_deterministic_runtime(): def _deterministic_time_provider() -> float: if workflow.in_workflow(): return workflow.now().timestamp() - return time.time() # Fallback to system time + return time.time() def _deterministic_id_provider() -> str: if workflow.in_workflow(): return str(workflow.uuid4()) - return str(uuid.uuid4()) # Fallback to system UUID + return str(uuid.uuid4()) runtime.set_time_provider(_deterministic_time_provider) runtime.set_id_provider(_deterministic_id_provider) @@ -123,10 +123,9 @@ async def wrapper(*args, **kw): # Convert to positional args for Temporal activity_args = list(bound.arguments.values()) - # Strategy: Decorator kwargs are defaults. + # Decorator kwargs are defaults. options = kwargs.copy() - # Assert workflow import is available or mocked return await workflow.execute_activity( activity_def, *activity_args, @@ -196,57 +195,27 @@ async def _handle_generate_content(args: List[Any]) -> list[dict[str, Any]]: async def before_model_callback( self, *, callback_context: CallbackContext, llm_request: LlmRequest ) -> LlmResponse | None: - # If already in a workflow, execute the activity - if workflow.in_workflow(): - # Ensure model is set from agent if missing in request - if not llm_request.model: - if isinstance(callback_context.invocation_context.agent.model, str): - llm_request.model = callback_context.invocation_context.agent.model - elif hasattr(callback_context.invocation_context.agent.model, 'model_name'): - llm_request.model = callback_context.invocation_context.agent.model.model_name - - # Default options - options = { - "start_to_close_timeout": timedelta(seconds=60), - "retry_policy": RetryPolicy( - initial_interval=timedelta(seconds=1), - backoff_coefficient=2.0, - maximum_interval=timedelta(seconds=30), - maximum_attempts=5 - ) - } - # Merge with user options - options.update(self.activity_options) - - # Execution - # The activity returns list[dict] to avoid strict Pydantic validation issues. - - # Construct dynamic activity name for visibility - agent_name = callback_context.agent_name - activity_name = f"{agent_name}.generate_content" - - # Debug options - activity.logger.info(f"Executing activity '{activity_name}' with options: {options}") - - # Execute with dynamic name - response_dicts = await workflow.execute_activity( - activity_name, - args=[llm_request], - **options - ) - - # Rehydrate LlmResponse objects safely - responses = [] - for d in response_dicts: - try: - responses.append(LlmResponse.model_validate(d)) - except Exception as e: - raise RuntimeError(f"Failed to deserialized LlmResponse from activity result: {e}") from e - - # Simple consolidation: return the last complete response - return responses[-1] if responses else None + # Construct dynamic activity name for visibility + agent_name = callback_context.agent_name + activity_name = f"{agent_name}.generate_content" + + # Execute with dynamic name + response_dicts = await workflow.execute_activity( + activity_name, + args=[llm_request], + **self.activity_options + ) + + # Rehydrate LlmResponse objects safely + responses = [] + for d in response_dicts: + try: + responses.append(LlmResponse.model_validate(d)) + except Exception as e: + raise RuntimeError(f"Failed to deserialized LlmResponse from activity result: {e}") from e - return None + # Simple consolidation: return the last complete response + return responses[-1] if responses else None @@ -268,7 +237,6 @@ def __init__(self): def _configure_data_converter(self, converter: DataConverter | None) -> DataConverter: if converter is None: - # Create a default converter using our PydanticPayloadConverter return DataConverter( payload_converter_class=_DefaultPydanticPayloadConverter ) From abea6e6e183956ddebbc160cc5f08fa96c1167d7 Mon Sep 17 00:00:00 2001 From: Marcus Motill Date: Sun, 18 Jan 2026 02:23:15 +0000 Subject: [PATCH 6/7] move activity registration --- .../adk/integrations/temporal/__init__.py | 93 ++++++++++--------- .../manual_test_temporal_integration.py | 7 +- tests/unittests/integrations/test_temporal.py | 2 +- 3 files changed, 50 insertions(+), 52 deletions(-) diff --git a/src/google/adk/integrations/temporal/__init__.py b/src/google/adk/integrations/temporal/__init__.py index 2baf56c96c..89de98fe5e 100644 --- a/src/google/adk/integrations/temporal/__init__.py +++ b/src/google/adk/integrations/temporal/__init__.py @@ -139,6 +139,51 @@ async def wrapper(*args, **kw): return wrapper + async def before_model_callback( + self, *, callback_context: CallbackContext, llm_request: LlmRequest + ) -> LlmResponse | None: + # Construct dynamic activity name for visibility + agent_name = callback_context.agent_name + activity_name = f"{agent_name}.generate_content" + + # Execute with dynamic name + response_dicts = await workflow.execute_activity( + activity_name, + args=[llm_request], + **self.activity_options + ) + + # Rehydrate LlmResponse objects safely + responses = [] + for d in response_dicts: + try: + responses.append(LlmResponse.model_validate(d)) + except Exception as e: + raise RuntimeError(f"Failed to deserialized LlmResponse from activity result: {e}") from e + + # Simple consolidation: return the last complete response + return responses[-1] if responses else None + + + + +class AdkWorkerPlugin(SimplePlugin): + """A Temporal Worker Plugin configured for ADK. + + This plugin configures: + 1. Pydantic Payload Converter (required for ADK objects). + 2. Sandbox Passthrough for `google.adk` and `google.genai`. + """ + + def __init__(self): + super().__init__( + name="adk_worker_plugin", + data_converter=self._configure_data_converter, + workflow_runner=self._configure_workflow_runner, + activities=[self.dynamic_activity], + worker_interceptors=[AdkInterceptor()] + ) + @staticmethod @activity.defn(dynamic=True) async def dynamic_activity(args: Sequence[RawValue]) -> Any: @@ -147,8 +192,8 @@ async def dynamic_activity(args: Sequence[RawValue]) -> Any: # Check if this is a generate_content call if activity_type.endswith(".generate_content") or activity_type == "google.adk.generate_content": - return await TemporalPlugin._handle_generate_content(args) - + return await AdkWorkerPlugin._handle_generate_content(args) + raise ValueError(f"Unknown dynamic activity: {activity_type}") @staticmethod @@ -191,50 +236,6 @@ async def _handle_generate_content(args: List[Any]) -> list[dict[str, Any]]: for r in responses ] - - async def before_model_callback( - self, *, callback_context: CallbackContext, llm_request: LlmRequest - ) -> LlmResponse | None: - # Construct dynamic activity name for visibility - agent_name = callback_context.agent_name - activity_name = f"{agent_name}.generate_content" - - # Execute with dynamic name - response_dicts = await workflow.execute_activity( - activity_name, - args=[llm_request], - **self.activity_options - ) - - # Rehydrate LlmResponse objects safely - responses = [] - for d in response_dicts: - try: - responses.append(LlmResponse.model_validate(d)) - except Exception as e: - raise RuntimeError(f"Failed to deserialized LlmResponse from activity result: {e}") from e - - # Simple consolidation: return the last complete response - return responses[-1] if responses else None - - - - -class AdkWorkerPlugin(SimplePlugin): - """A Temporal Worker Plugin configured for ADK. - - This plugin configures: - 1. Pydantic Payload Converter (required for ADK objects). - 2. Sandbox Passthrough for `google.adk` and `google.genai`. - """ - def __init__(self): - super().__init__( - name="adk_worker_plugin", - data_converter=self._configure_data_converter, - workflow_runner=self._configure_workflow_runner, - worker_interceptors=[AdkInterceptor()] - ) - def _configure_data_converter(self, converter: DataConverter | None) -> DataConverter: if converter is None: return DataConverter( diff --git a/tests/integration/manual_test_temporal_integration.py b/tests/integration/manual_test_temporal_integration.py index 273d1fb397..f823810428 100644 --- a/tests/integration/manual_test_temporal_integration.py +++ b/tests/integration/manual_test_temporal_integration.py @@ -154,7 +154,7 @@ async def run(self, topic: str) -> str: agent=coordinator, app_name="multi_agent_app", session_service=session_service, - plugins=[TemporalPlugin()] + plugins=[TemporalPlugin(activity_options={'start_to_close_timeout': timedelta(minutes=2)})] ) # 4. Run @@ -196,12 +196,9 @@ async def test_temporal_integration(): task_queue="adk-task-queue", activities=[ get_weather, - TemporalPlugin.dynamic_activity, - # Note: We can add specific wrapper activities if we want distinct names in UI, - # but generate_content is the generic one used by TemporalPlugin. ], workflows=[WeatherAgent, MultiAgentWorkflow], - plugins=[AdkWorkerPlugin()] # <--- Use the class based plugin + plugins=[AdkWorkerPlugin()] ): print("Worker started.") # Test Weather Agent diff --git a/tests/unittests/integrations/test_temporal.py b/tests/unittests/integrations/test_temporal.py index 8508b92e67..a5a0ce30b0 100644 --- a/tests/unittests/integrations/test_temporal.py +++ b/tests/unittests/integrations/test_temporal.py @@ -143,7 +143,7 @@ def test_temporal_plugin_before_model(self): callback_context.agent_name = "test-agent" callback_context.invocation_context.agent.model = "test-agent-model" - llm_request = LlmRequest(model=None, prompt="hi") + llm_request = LlmRequest(model="test-agent-model", prompt="hi") # Run callback loop = asyncio.new_event_loop() From 6b83d77bfacc6e7d3eee171c61323d64ef8713a0 Mon Sep 17 00:00:00 2001 From: Marcus Motill Date: Sun, 18 Jan 2026 02:43:43 +0000 Subject: [PATCH 7/7] better naming --- src/google/adk/integrations/temporal/README.md | 18 +++++++++--------- .../adk/integrations/temporal/__init__.py | 6 +++--- .../manual_test_temporal_integration.py | 14 +++++++------- tests/unittests/integrations/test_temporal.py | 4 ++-- 4 files changed, 21 insertions(+), 21 deletions(-) diff --git a/src/google/adk/integrations/temporal/README.md b/src/google/adk/integrations/temporal/README.md index 56a18f2955..ffa2f69c65 100644 --- a/src/google/adk/integrations/temporal/README.md +++ b/src/google/adk/integrations/temporal/README.md @@ -4,9 +4,9 @@ This package provides the integration layer between the Google ADK and Temporal. ## Core Concepts -### 1. Interception Flow (`TemporalPlugin`) +### 1. Interception Flow (`AgentPlugin`) -The `TemporalPlugin` acts as a middleware that intercepts model calls (e.g., `agent.generate_content`) *before* they execute. +The `AgentPlugin` acts as a middleware that intercepts model calls (e.g., `agent.generate_content`) *before* they execute. **Workflow Interception:** 1. **Intercept**: The ADK invokes `before_model_callback` when an agent attempts to call a model. @@ -32,12 +32,12 @@ When the workflow executes an activity with an unknown name (e.g., `MyAgent.gene The integration requires setup on both the Agent (Workflow) side and the Worker side. #### Agent Setup (Workflow Side) -Attach the `TemporalPlugin` to your ADK agent. This safely routes model calls through Temporal activities. You **must** provide activity options (e.g., timeouts) as there are no defaults. +Attach the `AgentPlugin` to your ADK agent. This safely routes model calls through Temporal activities. You **must** provide activity options (e.g., timeouts) as there are no defaults. ```python from datetime import timedelta from temporalio.common import RetryPolicy -from google.adk.integrations.temporal import TemporalPlugin +from google.adk.integrations.temporal import AgentPlugin # 1. Define Temporal Activity Options activity_options = { @@ -50,7 +50,7 @@ agent = Agent( model="gemini-2.5-pro", plugins=[ # Routes model calls to Temporal Activities - TemporalPlugin(activity_options=activity_options) + AgentPlugin(activity_options=activity_options) ] ) @@ -59,23 +59,23 @@ agent = Agent( ``` #### Worker Setup -Install the `AdkWorkerPlugin` on your Temporal Worker. This handles serialization and runtime determinism. +Install the `WorkerPlugin` on your Temporal Worker. This handles serialization and runtime determinism. ```python from temporalio.worker import Worker -from google.adk.integrations.temporal import AdkWorkerPlugin +from google.adk.integrations.temporal import WorkerPlugin async def main(): worker = Worker( client, task_queue="my-queue", # Configures ADK Runtime & Pydantic Support - plugins=[AdkWorkerPlugin()] + plugins=[WorkerPlugin()] ) await worker.run() ``` -**What `AdkWorkerPlugin` Does:** +**What `WorkerPlugin` Does:** * **Data Converter**: Enables Pydantic serialization for ADK objects. * **Interceptors**: Sets up specific ADK runtime hooks for determinism (replacing `time.time`, `uuid.uuid4`) before workflow execution. * TODO: is this enough . **Unsandboxed Workflow Runner**: Configures the worker to use the `UnsandboxedWorkflowRunner`, allowing standard imports in ADK agents. diff --git a/src/google/adk/integrations/temporal/__init__.py b/src/google/adk/integrations/temporal/__init__.py index 89de98fe5e..21419a054c 100644 --- a/src/google/adk/integrations/temporal/__init__.py +++ b/src/google/adk/integrations/temporal/__init__.py @@ -91,7 +91,7 @@ def workflow_interceptor_class( ) -> type[WorkflowInboundInterceptor] | None: return AdkWorkflowInboundInterceptor -class TemporalPlugin(BasePlugin): +class AgentPlugin(BasePlugin): """ADK Plugin for Temporal integration. This plugin automatically configures the ADK runtime to be deterministic when running @@ -167,7 +167,7 @@ async def before_model_callback( -class AdkWorkerPlugin(SimplePlugin): +class WorkerPlugin(SimplePlugin): """A Temporal Worker Plugin configured for ADK. This plugin configures: @@ -192,7 +192,7 @@ async def dynamic_activity(args: Sequence[RawValue]) -> Any: # Check if this is a generate_content call if activity_type.endswith(".generate_content") or activity_type == "google.adk.generate_content": - return await AdkWorkerPlugin._handle_generate_content(args) + return await WorkerPlugin._handle_generate_content(args) raise ValueError(f"Unknown dynamic activity: {activity_type}") diff --git a/tests/integration/manual_test_temporal_integration.py b/tests/integration/manual_test_temporal_integration.py index f823810428..e6394df078 100644 --- a/tests/integration/manual_test_temporal_integration.py +++ b/tests/integration/manual_test_temporal_integration.py @@ -39,7 +39,7 @@ from google.adk.sessions import InMemorySessionService from google.adk.utils.context_utils import Aclosing from google.adk.events import Event -from google.adk.integrations.temporal import AdkWorkerPlugin, TemporalPlugin +from google.adk.integrations.temporal import WorkerPlugin, AgentPlugin # Required Environment Variables for this test: # in this folder update .env.example to be .env and have the following vars: @@ -68,11 +68,11 @@ async def run(self, prompt: str) -> Event | None: logger.info("Workflow started.") # 1. Define Agent using Temporal Helpers - # Note: TemporalPlugin in the Runner automatically handles Runtime setup + # Note: AgentPlugin in the Runner automatically handles Runtime setup # and Model Activity interception. We use standard ADK models now. # Wraps 'get_weather' activity as a Tool - weather_tool = TemporalPlugin.activity_tool( + weather_tool = AgentPlugin.activity_tool( get_weather, start_to_close_timeout=timedelta(seconds=60) ) @@ -90,12 +90,12 @@ async def run(self, prompt: str) -> Event | None: logger.info(f"Session created with ID: {session.id}") - # 3. Run Agent with TemporalPlugin + # 3. Run Agent with AgentPlugin runner = Runner( agent=agent, app_name='test_app', session_service=session_service, - plugins=[TemporalPlugin(activity_options={'start_to_close_timeout': timedelta(minutes=2)})] + plugins=[AgentPlugin(activity_options={'start_to_close_timeout': timedelta(minutes=2)})] ) logger.info("Starting runner.") @@ -154,7 +154,7 @@ async def run(self, topic: str) -> str: agent=coordinator, app_name="multi_agent_app", session_service=session_service, - plugins=[TemporalPlugin(activity_options={'start_to_close_timeout': timedelta(minutes=2)})] + plugins=[AgentPlugin(activity_options={'start_to_close_timeout': timedelta(minutes=2)})] ) # 4. Run @@ -198,7 +198,7 @@ async def test_temporal_integration(): get_weather, ], workflows=[WeatherAgent, MultiAgentWorkflow], - plugins=[AdkWorkerPlugin()] + plugins=[WorkerPlugin()] ): print("Worker started.") # Test Weather Agent diff --git a/tests/unittests/integrations/test_temporal.py b/tests/unittests/integrations/test_temporal.py index a5a0ce30b0..9efa38a988 100644 --- a/tests/unittests/integrations/test_temporal.py +++ b/tests/unittests/integrations/test_temporal.py @@ -96,7 +96,7 @@ async def my_activity(arg: str) -> str: return f"Hello {arg}" # Wrap it - tool = temporal.TemporalPlugin.activity_tool( + tool = temporal.AgentPlugin.activity_tool( my_activity, start_to_close_timeout=100 ) @@ -122,7 +122,7 @@ async def my_activity(arg: str) -> str: def test_temporal_plugin_before_model(self): - plugin = temporal.TemporalPlugin(activity_options={"start_to_close_timeout": 60}) + plugin = temporal.AgentPlugin(activity_options={"start_to_close_timeout": 60}) # Setup mocks mock_workflow.reset_mock()