From b862834830c62c0c5072ef471100911d116a49e5 Mon Sep 17 00:00:00 2001 From: Marcus Motill Date: Mon, 19 Jan 2026 02:20:32 +0000 Subject: [PATCH] add google adk plugins --- .../contrib/google_adk_agents/README.md | 81 ++++++ .../contrib/google_adk_agents/__init__.py | 261 ++++++++++++++++++ .../test_temporal_integration.py | 245 ++++++++++++++++ .../google_adk_agents/test_temporal_unit.py | 177 ++++++++++++ 4 files changed, 764 insertions(+) create mode 100644 temporalio/contrib/google_adk_agents/README.md create mode 100644 temporalio/contrib/google_adk_agents/__init__.py create mode 100644 tests/contrib/google_adk_agents/test_temporal_integration.py create mode 100644 tests/contrib/google_adk_agents/test_temporal_unit.py diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md new file mode 100644 index 000000000..632e92c4e --- /dev/null +++ b/temporalio/contrib/google_adk_agents/README.md @@ -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. diff --git a/temporalio/contrib/google_adk_agents/__init__.py b/temporalio/contrib/google_adk_agents/__init__.py new file mode 100644 index 000000000..123aac4bd --- /dev/null +++ b/temporalio/contrib/google_adk_agents/__init__.py @@ -0,0 +1,261 @@ +# 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 asyncio +import dataclasses +import functools +import inspect +import time +import uuid +from collections.abc import Sequence +from datetime import timedelta +from typing import Any, AsyncGenerator, Callable, List, Optional + +from google.adk.agents.callback_context import CallbackContext +from google.adk.agents.invocation_context import InvocationContext +from google.adk.models import BaseLlm, LLMRegistry, LlmRequest, LlmResponse +from google.adk.plugins import BasePlugin +from google.genai import types + +from temporalio import activity, workflow +from temporalio.common import RawValue, RetryPolicy +from temporalio.contrib.pydantic import ( + PydanticPayloadConverter as _DefaultPydanticPayloadConverter, +) +from temporalio.converter import DataConverter, DefaultPayloadConverter +from temporalio.plugin import SimplePlugin +from temporalio.worker import ( + ExecuteWorkflowInput, + Interceptor, + UnsandboxedWorkflowRunner, + WorkflowInboundInterceptor, + WorkflowInterceptorClassInput, + WorkflowRunner, +) + + +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 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 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__ + wrapper.__signature__ = inspect.signature(activity_def) + + 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 WorkerPlugin(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: + """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 WorkerPlugin._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] + + 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 + + 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/contrib/google_adk_agents/test_temporal_integration.py b/tests/contrib/google_adk_agents/test_temporal_integration.py new file mode 100644 index 000000000..8ea2eba8f --- /dev/null +++ b/tests/contrib/google_adk_agents/test_temporal_integration.py @@ -0,0 +1,245 @@ +# 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 os +import uuid +from datetime import timedelta +from typing import AsyncGenerator + +import pytest +from google.adk import Agent, Runner, runtime +from google.adk.agents import LlmAgent +from google.adk.events import Event +from google.adk.models import LLMRegistry, LlmRequest, LlmResponse +from google.adk.sessions import InMemorySessionService +from google.adk.tools import AgentTool +from google.adk.utils.context_utils import Aclosing +from google.genai import types + +from temporalio import activity, workflow +from temporalio.client import Client +from temporalio.contrib.google_adk_agents import AgentPlugin, WorkerPlugin +from temporalio.contrib.pydantic import ( + PydanticPayloadConverter, + pydantic_data_converter, +) +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, +) + +# Required Environment Variables for this test: +# 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 for a given city.""" + return "Warm and sunny. 17 degrees." + + +@workflow.defn +class WeatherAgent: + @workflow.run + async def run(self, prompt: str) -> Event | None: + logger.info("Workflow started.") + + # 1. Define Agent using Temporal Helpers + # 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 = AgentPlugin.activity_tool( + get_weather, start_to_close_timeout=timedelta(seconds=60) + ) + + agent = Agent( + name="test_agent", + model="gemini-2.5-pro", # Standard model string + tools=[weather_tool], + ) + + # 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}") + + # 3. Run Agent with AgentPlugin + runner = Runner( + agent=agent, + app_name="test_app", + session_service=session_service, + plugins=[ + AgentPlugin( + activity_options={"start_to_close_timeout": timedelta(minutes=2)} + ) + ], + ) + + 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, topic: str) -> str: + # Example of multi-turn/multi-agent orchestration + # This is where Temporal shines - orchestrating complex agent flows + + # 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" + ) + + # 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.", + ) + + # Sub-agent: Writer + writer = LlmAgent( + name="writer", + model="gemini-2.5-pro", + instruction="You are a poet. Write a haiku based on the research.", + ) + + # 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], + ) + + # 3. Initialize Runner with required args + runner = Runner( + agent=coordinator, + app_name="multi_agent_app", + session_service=session_service, + plugins=[ + AgentPlugin( + activity_options={"start_to_close_timeout": timedelta(minutes=2)} + ) + ], + ) + + # 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 + + +@pytest.mark.asyncio +async def test_temporal_integration(): + """Manual integration test requiring a running Temporal server.""" + + # 1. Start a Worker (in a real app, this would be a separate process) + # We run it here for the test. + + try: + # 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, + task_queue="adk-task-queue", + activities=[ + get_weather, + ], + workflows=[WeatherAgent, MultiAgentWorkflow], + plugins=[WorkerPlugin()], + ): + print("Worker started.") + # Test Weather Agent + result = await client.execute_workflow( + WeatherAgent.run, + "What is the weather in New York?", + id=f"weather-agent-workflow-{uuid.uuid4()}", + task_queue="adk-task-queue", + ) + print(f"Workflow result: {result}") + + # Test Multi Agent + result_multi = await client.execute_workflow( + MultiAgentWorkflow.run, + "Run mult-agent flow", + id=f"multi-agent-workflow-{uuid.uuid4()}", + task_queue="adk-task-queue", + ) + print(f"Multi-Agent Workflow result: {result_multi}") diff --git a/tests/contrib/google_adk_agents/test_temporal_unit.py b/tests/contrib/google_adk_agents/test_temporal_unit.py new file mode 100644 index 000000000..41447b09b --- /dev/null +++ b/tests/contrib/google_adk_agents/test_temporal_unit.py @@ -0,0 +1,177 @@ +# 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 asyncio +import sys +import unittest +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +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() +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() +mock_temporalio.workflow = mock_workflow +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 import runtime + from google.adk.agents.callback_context import CallbackContext + from google.adk.agents.invocation_context import InvocationContext + from google.adk.models import LlmRequest, LlmResponse + + from temporalio.contrib import google_adk_agents as temporal + + +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 + assert temporal.workflow.execute_activity is mock_workflow.execute_activity + + # Define a fake activity + async def my_activity(arg: str) -> str: + """My Docstring.""" + return f"Hello {arg}" + + # Wrap it + tool = temporal.AgentPlugin.activity_tool( + my_activity, start_to_close_timeout=100 + ) + + # Check metadata + self.assertEqual(tool.__name__, "my_activity") # Matches function name + self.assertEqual(tool.__doc__, "My Docstring.") + + # Run tool (wrapper) + loop = asyncio.new_event_loop() + try: + 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(args[1], "World") + self.assertEqual(kwargs["start_to_close_timeout"], 100) + + def test_temporal_plugin_before_model(self): + plugin = temporal.AgentPlugin(activity_options={"start_to_close_timeout": 60}) + + # 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. + + mock_workflow.execute_activity = AsyncMock(return_value=[response_dict]) + + # 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" + + llm_request = LlmRequest(model="test-agent-model", prompt="hi") + + # Run callback + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + result = loop.run_until_complete( + plugin.before_model_callback( + callback_context=callback_context, llm_request=llm_request + ) + ) + finally: + loop.close() + + # Verify execution + mock_workflow.execute_activity.assert_called_once() + args, kwargs = mock_workflow.execute_activity.call_args + 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")