feat: Add Temporal integration and deterministic runtime support

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()

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()

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()

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: