fix(models) initialize Anthropic client off event loop

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:

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