diff --git a/src/google/adk/models/anthropic_llm.py b/src/google/adk/models/anthropic_llm.py index f6622628d6..f084c0b12b 100644 --- a/src/google/adk/models/anthropic_llm.py +++ b/src/google/adk/models/anthropic_llm.py @@ -16,6 +16,7 @@ from __future__ import annotations +import asyncio import base64 import copy import dataclasses @@ -46,6 +47,7 @@ from pydantic import BaseModel from pydantic import Field from pydantic import model_validator +from pydantic import PrivateAttr from typing_extensions import override from ..utils import _json_utils @@ -829,6 +831,8 @@ class AnthropicLlm(BaseLlm): ) """An optional pre-configured Anthropic client.""" + _client_init_lock: asyncio.Lock = PrivateAttr(default_factory=asyncio.Lock) + @classmethod @override def supported_models(cls) -> list[str]: @@ -952,15 +956,16 @@ async def generate_content_async( thinking = _build_anthropic_thinking_param(llm_request.config) try: + client = await self._get_anthropic_client() if not stream: kwargs = self._build_anthropic_kwargs( llm_request, messages, tools, tool_choice, thinking ) - message = await self._anthropic_client.messages.create(**kwargs) + message = await client.messages.create(**kwargs) yield message_to_generate_content_response(message) else: async for response in self._generate_content_streaming( - llm_request, messages, tools, tool_choice, thinking + llm_request, messages, tools, tool_choice, thinking, client ): yield response except RateLimitError as rate_limit_error: @@ -978,6 +983,7 @@ async def _generate_content_streaming( anthropic_types.ThinkingConfigAdaptiveParam, NotGiven, ] = NOT_GIVEN, + client: AsyncAnthropic | AsyncAnthropicVertex | None = None, ) -> AsyncGenerator[LlmResponse, None]: """Handles streaming responses from Anthropic models. @@ -995,7 +1001,9 @@ async def _generate_content_streaming( kwargs = self._build_anthropic_kwargs( llm_request, messages, tools, tool_choice, thinking ) - raw_stream = await self._anthropic_client.messages.create( + if client is None: + client = await self._get_anthropic_client() + raw_stream = await client.messages.create( stream=True, **kwargs, ) @@ -1154,6 +1162,27 @@ async def _generate_content_streaming( partial=False, ) + async def _get_anthropic_client( + self, + ) -> AsyncAnthropic | AsyncAnthropicVertex: + """Returns the client without blocking the caller's event loop. + + The Anthropic SDK may perform synchronous credential discovery while + constructing a client. Agent Engine invokes this model from its serving + event loop, so doing that work inline can also block health checks. Keep + construction off the event loop and serialize concurrent first access so + the cached property creates only one client. + """ + cached_client = self.__dict__.get("_anthropic_client") + if cached_client is not None: + return cast(AsyncAnthropic | AsyncAnthropicVertex, cached_client) + + async with self._client_init_lock: + cached_client = self.__dict__.get("_anthropic_client") + if cached_client is not None: + return cast(AsyncAnthropic | AsyncAnthropicVertex, cached_client) + return await asyncio.to_thread(lambda: self._anthropic_client) + @cached_property def _anthropic_client(self) -> AsyncAnthropic | AsyncAnthropicVertex: if self.client: diff --git a/tests/unittests/models/test_anthropic_llm.py b/tests/unittests/models/test_anthropic_llm.py index 1f25a21700..8297462a72 100644 --- a/tests/unittests/models/test_anthropic_llm.py +++ b/tests/unittests/models/test_anthropic_llm.py @@ -12,11 +12,12 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio import base64 import json import os import re -import sys +import threading from unittest import mock from unittest.mock import AsyncMock from unittest.mock import MagicMock @@ -38,7 +39,6 @@ from google.adk.models.llm_request import LlmRequest from google.adk.models.llm_response import LlmResponse from google.genai import types -from google.genai import version as genai_version from google.genai.types import Content from google.genai.types import Part import httpx @@ -151,6 +151,34 @@ def test_claude_anthropic_client_creation_with_full_resource_name(): assert kwargs["region"] == "test-location" +@pytest.mark.asyncio +async def test_claude_anthropic_client_creation_runs_off_event_loop(): + model = Claude( + model="projects/test-project/locations/test-location/publishers/anthropic/models/claude-3-5-sonnet-v2@20241022" + ) + event_loop_thread = threading.get_ident() + construction_threads = [] + client = MagicMock() + + def create_client(**kwargs): + del kwargs + construction_threads.append(threading.get_ident()) + return client + + with mock.patch( + "google.adk.models.anthropic_llm.AsyncAnthropicVertex", + side_effect=create_client, + ) as mock_client_class: + clients = await asyncio.gather( + model._get_anthropic_client(), + model._get_anthropic_client(), + ) + + assert clients == [client, client] + mock_client_class.assert_called_once() + assert construction_threads[0] != event_loop_thread + + def test_supported_models(): models = Claude.supported_models() assert len(models) == 3