Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ Documentation = "https://docs.temporal.io/docs/python"
dev = [
"basedpyright==1.34.0",
"cibuildwheel>=2.22.0,<3",
"google-adk @ git+https://github.com/marcusmotill/adk-python-temporal.git@motill/durable-support",
"grpcio-tools>=1.48.2,<2",
"mypy==1.18.2",
"mypy-protobuf>=3.3.0,<4",
Expand Down
81 changes: 81 additions & 0 deletions temporalio/contrib/google_adk_agents/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
# Google ADK Agents SDK Integration for Temporal

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

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.
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 `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 AgentPlugin

# 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
AgentPlugin(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 `WorkerPlugin` on your Temporal Worker. This handles serialization and runtime determinism.

```python
from temporalio.worker import Worker
from google.adk.integrations.temporal import WorkerPlugin

async def main():
worker = Worker(
client,
task_queue="my-queue",
# Configures ADK Runtime & Pydantic Support
plugins=[WorkerPlugin()]
)
await worker.run()
```

**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.
198 changes: 198 additions & 0 deletions temporalio/contrib/google_adk_agents/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,198 @@
# 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 dataclasses
import inspect
import time
import uuid
from contextlib import asynccontextmanager
from typing import Any, AsyncIterator, Callable, Optional

from google.adk.agents.callback_context import CallbackContext
from google.adk.models import LLMRegistry
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.adk.plugins import BasePlugin

from temporalio import activity, workflow
from temporalio.contrib.pydantic import (
PydanticPayloadConverter as _DefaultPydanticPayloadConverter,
)
from temporalio.converter import DataConverter, DefaultPayloadConverter
from temporalio.plugin import SimplePlugin
from temporalio.worker import (
WorkflowRunner,
)
from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner


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

def _deterministic_id_provider() -> str:
if workflow.in_workflow():
return str(workflow.uuid4())
return str(uuid.uuid4())

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

# Decorator kwargs are defaults.
options = kwargs.copy()

return await workflow.execute_activity(
activity_def, *activity_args, **options
)

# Copy metadata
wrapper.__name__ = activity_def.__name__
wrapper.__doc__ = activity_def.__doc__
setattr(wrapper, "__signature__", inspect.signature(activity_def))

return wrapper

async def before_model_callback(
self, *, callback_context: CallbackContext, llm_request: LlmRequest
) -> LlmResponse | None:
responses = await workflow.execute_activity(
invoke_model,
args=[llm_request],
summary=callback_context.agent_name,
**self.activity_options,
)

# Simple consolidation: return the last complete response
return responses[-1] if responses else None


@activity.defn
async def invoke_model(llm_request: LlmRequest) -> list[LlmResponse]:
if llm_request.model is None:
raise ValueError(f"No model name provided, could not create LLM.")

llm = LLMRegistry.new_llm(llm_request.model)
if not llm:
raise ValueError(f"Failed to create LLM for model: {llm_request.model}")

return [
response
async for response in llm.generate_content_async(llm_request=llm_request)
]


class GoogleAdkPlugin(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):
@asynccontextmanager
async def run_context() -> AsyncIterator[None]:
setup_deterministic_runtime()
yield

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

super().__init__(
name="google_adk_plugin",
data_converter=self._configure_data_converter,
activities=[invoke_model],
run_context=lambda: run_context(),
workflow_runner=workflow_runner,
)

def _configure_data_converter(
self, converter: DataConverter | None
) -> DataConverter:
if converter is None:
return DataConverter(
payload_converter_class=_DefaultPydanticPayloadConverter
)
elif converter.payload_converter_class is DefaultPayloadConverter:
return dataclasses.replace(
converter, payload_converter_class=_DefaultPydanticPayloadConverter
)
return converter
Loading