feat(models): adding python support for Azure foundry models (OpenAI + Anthropic types)

python/packages/kagent-adk/pyproject.toml

@@ -41,6 +41,7 @@ dependencies = [
   "boto3>=1.28.57",
   "ollama >=0.3.6",  # Ollama SDK
   "numpy>=2.2.6",
+  "azure-identity>=1.19.0",  # Azure Workload Identity for Azure OpenAI / Foundry
 ]
 
 [tool.uv.sources]

python/packages/kagent-adk/src/kagent/adk/models/__init__.py

@@ -4,11 +4,13 @@
 from ._gemini import KAgentGeminiLlm, KAgentGeminiVertexAILlm
 from ._ollama import KAgentOllamaLlm
 from ._openai import AzureOpenAI, OpenAI
+from ._openai import FoundryOpenAI as Foundry
 from ._sap_ai_core import KAgentSAPAICoreLlm
 
 __all__ = [
     "OpenAI",
     "AzureOpenAI",
+    "Foundry",
     "KAgentAnthropicLlm",
     "KAgentBedrockLlm",
     "KAgentGeminiLlm",

python/packages/kagent-adk/src/kagent/adk/models/_anthropic.py

@@ -10,6 +10,11 @@
 from anthropic import AsyncAnthropic
 from google.adk.models.anthropic_llm import AnthropicLlm
 
+from ._azure import (
+    build_foundry_anthropic_client,
+    resolve_azure_api_key,
+    resolve_foundry_endpoint_deployment,
+)
 from ._ssl import KAgentTLSMixin
 
 logger = logging.getLogger(__name__)
@@ -28,10 +33,10 @@ class KAgentAnthropicLlm(KAgentTLSMixin, AnthropicLlm):
 
     def set_passthrough_key(self, token: str) -> None:
         """Forward the Bearer token from the incoming A2A request as the Anthropic API key."""
-        self._api_key = token
-        # Invalidate cached clients so they're recreated with the new key
-        self.__dict__.pop("_anthropic_client", None)
-        self.__dict__.pop("_http_client", None)
+        if self._api_key != token:
+            self._api_key = token
+            # The SDK client captures auth at construction, so rebuild it only when the token changes.
+            self.__dict__.pop("_anthropic_client", None)
 
     def _create_http_client(self):
         """Create HTTP client with custom SSL context using Anthropic SDK defaults.
@@ -58,3 +63,31 @@ def _anthropic_client(self) -> AsyncAnthropic:
             kwargs["http_client"] = http_client
 
         return AsyncAnthropic(**kwargs)
+
+
+class FoundryAnthropic(KAgentAnthropicLlm):
+    """Claude on Azure AI Foundry's Anthropic Messages API."""
+
+    endpoint: Optional[str] = None
+    deployment: Optional[str] = None
+
+    def _resolve_model_name(self, model: Optional[str]) -> str:
+        del model
+        _, deployment = resolve_foundry_endpoint_deployment(self.endpoint, self.deployment)
+        return deployment
+
+    @cached_property
+    def _anthropic_client(self) -> AsyncAnthropic:
+        endpoint, _ = resolve_foundry_endpoint_deployment(self.endpoint, self.deployment)
+        api_key = resolve_azure_api_key(
+            self._api_key,
+            api_key_passthrough=self.api_key_passthrough,
+            environment_variable="FOUNDRY_API_KEY",
+        )
+        return build_foundry_anthropic_client(
+            endpoint=endpoint,
+            api_key=api_key,
+            api_key_passthrough=self.api_key_passthrough,
+            default_headers=self.extra_headers,
+            http_client=self._create_http_client(),
+        )

python/packages/kagent-adk/src/kagent/adk/models/_azure.py

@@ -0,0 +1,179 @@
+"""Shared configuration and client helpers for Azure AI providers."""
+
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, Optional
+
+import httpx
+from anthropic import AsyncAnthropic
+from openai import AsyncAzureOpenAI
+from openai.lib.azure import API_KEY_SENTINEL
+
+if TYPE_CHECKING:
+    from anthropic.lib.credentials import AccessToken, AccessTokenProvider
+    from azure.identity import DefaultAzureCredential
+
+COGNITIVE_SERVICES_SCOPE = "https://cognitiveservices.azure.com/.default"
+AI_FOUNDRY_SCOPE = "https://ai.azure.com/.default"
+
+AZURE_OPENAI_DEFAULT_API_VERSION = "2024-02-15-preview"
+FOUNDRY_DEFAULT_API_VERSION = "2024-10-21"
+
+_AUTH_HEADER_NAMES = {"authorization", "api-key", "x-api-key"}
+
+AsyncTokenProvider = Callable[[], Awaitable[str]]
+
+
+def azure_ad_token_provider(scope: str = COGNITIVE_SERVICES_SCOPE) -> AsyncTokenProvider:
+    """Return an async bearer-token provider backed by ``DefaultAzureCredential``."""
+    from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider
+
+    return get_bearer_token_provider(DefaultAzureCredential(), scope)
+
+
+class _AzureAccessTokenProvider:
+    """Adapt Azure Identity to Anthropic's synchronous access-token provider."""
+
+    def __init__(self, scope: str) -> None:
+        from azure.identity import DefaultAzureCredential
+
+        self._credential: DefaultAzureCredential = DefaultAzureCredential()
+        self._scope = scope
+
+    def __call__(self, *, force_refresh: bool = False) -> "AccessToken":
+        del force_refresh
+
+        from anthropic.lib.credentials import AccessToken
+
+        token = self._credential.get_token(self._scope)
+        return AccessToken(token=token.token, expires_at=token.expires_on)
+
+    def close(self) -> None:
+        self._credential.close()
+
+
+def azure_access_token_provider(scope: str = AI_FOUNDRY_SCOPE) -> "AccessTokenProvider":
+    """Return an Azure provider compatible with Anthropic's token cache."""
+    return _AzureAccessTokenProvider(scope)
+
+
+def resolve_azure_api_key(
+    api_key: Optional[str],
+    *,
+    api_key_passthrough: Optional[bool],
+    environment_variable: str,
+) -> Optional[str]:
+    """Resolve an Azure API key without bypassing passthrough mode."""
+    if api_key_passthrough:
+        return api_key
+    return api_key or os.environ.get(environment_variable)
+
+
+def resolve_azure_openai_config(endpoint: Optional[str], api_version: Optional[str]) -> tuple[str, str]:
+    """Resolve Azure OpenAI endpoint and API version configuration."""
+    resolved_endpoint = endpoint or os.environ.get("AZURE_OPENAI_ENDPOINT")
+    if not resolved_endpoint:
+        raise ValueError(
+            "Azure endpoint must be provided either via azure_endpoint parameter or "
+            "AZURE_OPENAI_ENDPOINT environment variable"
+        )
+
+    resolved_api_version = api_version or os.environ.get("OPENAI_API_VERSION") or AZURE_OPENAI_DEFAULT_API_VERSION
+    return resolved_endpoint, resolved_api_version
+
+
+def resolve_foundry_endpoint_deployment(endpoint: Optional[str], deployment: Optional[str]) -> tuple[str, str]:
+    """Resolve the Foundry endpoint and deployment."""
+    resolved_endpoint = endpoint or os.environ.get("FOUNDRY_ENDPOINT")
+    if not resolved_endpoint:
+        raise ValueError(
+            "Foundry endpoint must be provided either via endpoint parameter or FOUNDRY_ENDPOINT environment variable"
+        )
+
+    resolved_deployment = deployment or os.environ.get("FOUNDRY_DEPLOYMENT")
+    if not resolved_deployment:
+        raise ValueError(
+            "Foundry deployment must be provided either via deployment parameter or "
+            "FOUNDRY_DEPLOYMENT environment variable"
+        )
+
+    return resolved_endpoint, resolved_deployment
+
+
+def resolve_foundry_config(
+    endpoint: Optional[str], deployment: Optional[str], api_version: Optional[str]
+) -> tuple[str, str, str]:
+    """Resolve Foundry OpenAI-compatible data-plane configuration."""
+    resolved_endpoint, resolved_deployment = resolve_foundry_endpoint_deployment(endpoint, deployment)
+
+    resolved_api_version = api_version or os.environ.get("FOUNDRY_API_VERSION") or FOUNDRY_DEFAULT_API_VERSION
+    return resolved_endpoint, resolved_deployment, resolved_api_version
+
+
+def sanitize_azure_auth_headers(headers: Optional[dict[str, str]]) -> Optional[dict[str, str]]:
+    """Remove configured headers that could conflict with resolved Azure auth."""
+    if not headers:
+        return None
+    sanitized = {name: value for name, value in headers.items() if name.lower() not in _AUTH_HEADER_NAMES}
+    return sanitized or None
+
+
+def build_azure_openai_client(
+    *,
+    api_version: str,
+    azure_endpoint: str,
+    azure_deployment: Optional[str],
+    api_key: Optional[str],
+    api_key_passthrough: Optional[bool],
+    default_headers: Optional[dict[str, str]],
+    http_client: Optional[httpx.AsyncClient],
+    missing_credential_hint: str,
+) -> AsyncAzureOpenAI:
+    """Build an Azure OpenAI client using key, passthrough, or Workload Identity auth."""
+    token_provider = None
+    if not api_key:
+        if api_key_passthrough:
+            raise ValueError(missing_credential_hint)
+        token_provider = azure_ad_token_provider()
+
+    return AsyncAzureOpenAI(
+        # The sentinel prevents environment-key fallback while the token
+        # provider authenticates each request.
+        api_key=API_KEY_SENTINEL if token_provider is not None else api_key,
+        azure_ad_token_provider=token_provider,
+        api_version=api_version,
+        azure_endpoint=azure_endpoint,
+        azure_deployment=azure_deployment,
+        default_headers=sanitize_azure_auth_headers(default_headers),
+        http_client=http_client,
+    )
+
+
+def build_foundry_anthropic_client(
+    *,
+    endpoint: str,
+    api_key: Optional[str],
+    api_key_passthrough: Optional[bool],
+    default_headers: Optional[dict[str, str]],
+    http_client: Optional[httpx.AsyncClient],
+) -> AsyncAnthropic:
+    """Build a Foundry Anthropic client using key, passthrough, or Workload Identity auth."""
+    if api_key_passthrough and not api_key:
+        raise ValueError(
+            "No Azure credential resolved: provide the passthrough token before creating the Foundry Anthropic client"
+        )
+
+    kwargs: dict[str, Any] = {"base_url": endpoint.rstrip("/") + "/anthropic"}
+    if api_key:
+        kwargs["api_key"] = api_key
+    else:
+        kwargs["credentials"] = azure_access_token_provider(AI_FOUNDRY_SCOPE)
+
+    safe_headers = sanitize_azure_auth_headers(default_headers)
+    if safe_headers:
+        kwargs["default_headers"] = safe_headers
+    if http_client is not None:
+        kwargs["http_client"] = http_client
+
+    return AsyncAnthropic(**kwargs)

python/packages/kagent-adk/src/kagent/adk/models/_embedding.py

@@ -19,6 +19,13 @@
 from kagent.adk._bearer_token import bearer_token
 from kagent.adk.types import EmbeddingConfig
 
+from ._azure import (
+    build_azure_openai_client,
+    resolve_azure_api_key,
+    resolve_azure_openai_config,
+    resolve_foundry_config,
+)
+
 logger = logging.getLogger(__name__)
 
 
@@ -85,6 +92,8 @@ async def _call_provider(self, texts: List[str]) -> List[List[float]]:
 
         if provider in ("openai", "azure_openai"):
             return await self._embed_openai(texts)
+        if provider == "foundry":
+            return await self._embed_foundry(texts)
         if provider == "ollama":
             return await self._embed_ollama(texts)
         if provider in ("vertex_ai", "gemini"):
@@ -146,8 +155,8 @@ def _normalize_l2(self, x: Union[List[float], np.ndarray]) -> np.ndarray:
     def _passthrough_api_key(self) -> Optional[str]:
         """Bearer token to use as the API key when api_key_passthrough is
         enabled, mirroring BaseOpenAI.set_passthrough_key for chat models.
-        None falls back to the SDK's own env var lookup (OPENAI_API_KEY /
-        AZURE_OPENAI_API_KEY).
+        Azure providers treat a missing token as an error rather than falling
+        back to a provider environment key.
         """
         if not self.config.api_key_passthrough:
             return None
@@ -159,13 +168,20 @@ async def _embed_openai(self, texts: List[str]) -> List[List[float]]:
         api_key = self._passthrough_api_key()
 
         if provider == "azure_openai":
-            from openai import AsyncAzureOpenAI
-
-            api_version = os.environ.get("OPENAI_API_VERSION", "2024-02-15-preview")
-            api_base = self.config.base_url or os.environ.get("AZURE_OPENAI_ENDPOINT")
-            if not api_base:
-                raise ValueError("Azure OpenAI endpoint must be set via base_url or AZURE_OPENAI_ENDPOINT env var")
-            client = AsyncAzureOpenAI(api_version=api_version, azure_endpoint=api_base, api_key=api_key)
+            api_base, api_version = resolve_azure_openai_config(
+                self.config.endpoint or self.config.base_url, self.config.api_version
+            )
+            api_key = resolve_azure_api_key(
+                api_key,
+                api_key_passthrough=self.config.api_key_passthrough,
+                environment_variable="AZURE_OPENAI_API_KEY",
+            )
+            client = self._build_azure_client(
+                api_version=api_version,
+                endpoint=api_base,
+                deployment=self.config.deployment,
+                api_key=api_key,
+            )
         else:
             from openai import AsyncOpenAI
 
@@ -178,6 +194,52 @@ async def _embed_openai(self, texts: List[str]) -> List[List[float]]:
         )
         return [item.embedding for item in response.data]
 
+    async def _embed_foundry(self, texts: List[str]) -> List[List[float]]:
+        """Embed using the Azure AI Foundry OpenAI-compatible surface."""
+        endpoint, deployment, api_version = resolve_foundry_config(
+            self.config.endpoint, self.config.deployment, self.config.api_version
+        )
+        api_key = resolve_azure_api_key(
+            self._passthrough_api_key(),
+            api_key_passthrough=self.config.api_key_passthrough,
+            environment_variable="FOUNDRY_API_KEY",
+        )
+
+        client = self._build_azure_client(
+            api_version=api_version,
+            endpoint=endpoint,
+            deployment=deployment,
+            api_key=api_key,
+        )
+        response = await client.embeddings.create(
+            model=self.config.model,
+            input=texts,
+        )
+        return [item.embedding for item in response.data]
+
+    def _build_azure_client(
+        self,
+        *,
+        api_version: str,
+        endpoint: str,
+        deployment: Optional[str],
+        api_key: Optional[str],
+    ):
+        """Build an Azure embeddings client with implicit Workload Identity auth."""
+        return build_azure_openai_client(
+            api_version=api_version,
+            azure_endpoint=endpoint,
+            azure_deployment=deployment,
+            api_key=api_key,
+            api_key_passthrough=self.config.api_key_passthrough,
+            default_headers=None,
+            http_client=None,
+            missing_credential_hint=(
+                "No Azure credential resolved for embeddings: set an API key, enable "
+                "api_key_passthrough, or configure Azure Workload Identity"
+            ),
+        )
+
     async def _embed_ollama(self, texts: List[str]) -> List[List[float]]:
         """Embed using the Ollama SDK."""
         import ollama