chore: remove duplication in azure client

Author: greysonlalonde

lib/crewai/src/crewai/llm/base_llm.py

@@ -12,10 +12,11 @@
 import logging
 import os
 import re
-from typing import TYPE_CHECKING, Any, ClassVar, Final
+from typing import TYPE_CHECKING, Any, Final
 
+from dotenv import load_dotenv
 import httpx
-from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+from pydantic import BaseModel, Field, field_validator
 
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.llm_events import (
@@ -42,6 +43,8 @@
     from crewai.utilities.types import LLMMessage
 
 
+load_dotenv()
+
 DEFAULT_CONTEXT_WINDOW_SIZE: Final[int] = 4096
 DEFAULT_SUPPORTS_STOP_WORDS: Final[bool] = True
 _JSON_EXTRACTION_PATTERN: Final[re.Pattern[str]] = re.compile(r"\{.*}", re.DOTALL)
@@ -65,10 +68,6 @@ class BaseLLM(BaseModel, ABC, metaclass=LLMMeta):
         stop: A list of stop sequences that the LLM should use to stop generation.
     """
 
-    model_config: ClassVar[ConfigDict] = ConfigDict(
-        extra="allow", populate_by_name=True
-    )
-
     # Core fields
     model: str = Field(..., description="The model identifier/name")
     temperature: float | None = Field(
@@ -100,7 +99,7 @@ class BaseLLM(BaseModel, ABC, metaclass=LLMMeta):
         "cached_prompt_tokens": 0,
     }
 
-    @field_validator("api_key", mode="before")
+    @field_validator("api_key", mode="after")
     @classmethod
     def _validate_api_key(cls, value: str | None) -> str | None:
         """Validate API key for authentication.
@@ -137,37 +136,6 @@ def _normalize_stop(cls, value: Any) -> list[str]:
             return value
         return []
 
-    @model_validator(mode="before")
-    @classmethod
-    def _extract_stop_and_validate(cls, values: dict[str, Any]) -> dict[str, Any]:
-        """Extract and normalize stop sequences before model initialization.
-
-        Args:
-            values: Input values dictionary
-
-        Returns:
-            Processed values dictionary
-        """
-        if not values.get("model"):
-            raise ValueError("Model name is required and cannot be empty")
-
-        stop = values.get("stop") or values.get("stop_sequences")
-        if stop is None:
-            values["stop"] = []
-        elif isinstance(stop, str):
-            values["stop"] = [stop]
-        elif isinstance(stop, list):
-            values["stop"] = stop
-        else:
-            values["stop"] = []
-
-        values.pop("stop_sequences", None)
-
-        if "provider" not in values or values["provider"] is None:
-            values["provider"] = "openai"
-
-        return values
-
     @property
     def additional_params(self) -> dict[str, Any]:
         """Get additional parameters stored as extra fields.
@@ -190,20 +158,6 @@ def additional_params(self, value: dict[str, Any]) -> None:
             self.__pydantic_extra__ = {}
         self.__pydantic_extra__.update(value)
 
-    def model_post_init(self, __context: Any) -> None:
-        """Initialize token usage tracking after model initialization.
-
-        Args:
-            __context: Pydantic context (unused)
-        """
-        self._token_usage = {
-            "total_tokens": 0,
-            "prompt_tokens": 0,
-            "completion_tokens": 0,
-            "successful_requests": 0,
-            "cached_prompt_tokens": 0,
-        }
-
     @abstractmethod
     def call(
         self,

lib/crewai/src/crewai/llm/internal/constants.py

@@ -0,0 +1,14 @@
+from crewai.llm.constants import SupportedNativeProviders
+
+
+PROVIDER_MAPPING: dict[str, SupportedNativeProviders] = {
+    "openai": "openai",
+    "anthropic": "anthropic",
+    "claude": "anthropic",
+    "azure": "azure",
+    "azure_openai": "azure",
+    "google": "gemini",
+    "gemini": "gemini",
+    "bedrock": "bedrock",
+    "aws": "bedrock",
+}

lib/crewai/src/crewai/llm/internal/meta.py

@@ -9,6 +9,7 @@
 import logging
 from typing import Any, cast
 
+from pydantic import ConfigDict
 from pydantic._internal._model_construction import ModelMetaclass
 
 from crewai.llm.constants import (
@@ -21,6 +22,7 @@
     SupportedModels,
     SupportedNativeProviders,
 )
+from crewai.llm.internal.constants import PROVIDER_MAPPING
 
 
 class LLMMeta(ModelMetaclass):
@@ -30,6 +32,41 @@ class LLMMeta(ModelMetaclass):
     native provider implementation based on the model parameter.
     """
 
+    def __new__(
+        mcs,
+        name: str,
+        bases: tuple[type, ...],
+        namespace: dict[str, Any],
+        **kwargs: Any,
+    ) -> type:
+        """Create new LLM class with proper model_config for custom LLMs.
+
+        Args:
+            name: Class name
+            bases: Base classes
+            namespace: Class namespace
+            **kwargs: Additional arguments
+
+        Returns:
+            New class
+        """
+        if name != "BaseLLM" and any(
+            base.__name__ in ("BaseLLM", "LLM") for base in bases
+        ):
+            if "model_config" not in namespace:
+                namespace["model_config"] = ConfigDict(
+                    extra="allow", populate_by_name=True
+                )
+            elif isinstance(namespace["model_config"], dict):
+                config_dict = cast(
+                    ConfigDict, cast(object, dict(namespace["model_config"]))
+                )
+                config_dict.setdefault("extra", "allow")
+                config_dict.setdefault("populate_by_name", True)
+                namespace["model_config"] = ConfigDict(**config_dict)
+
+        return super().__new__(mcs, name, bases, namespace)
+
     def __call__(cls, *args: Any, **kwargs: Any) -> Any:  # noqa: N805
         """Route to appropriate provider implementation at instantiation time.
 
@@ -57,7 +94,7 @@ def __call__(cls, *args: Any, **kwargs: Any) -> Any:  # noqa: N805
 
         if args and not kwargs.get("model"):
             kwargs["model"] = cast(SupportedModels, args[0])
-            args = args[1:]
+            _ = args[1:]
         explicit_provider = cast(SupportedNativeProviders, kwargs.get("provider"))
 
         if explicit_provider:
@@ -70,19 +107,7 @@ def __call__(cls, *args: Any, **kwargs: Any) -> Any:  # noqa: N805
                 model.partition("/"),
             )
 
-            provider_mapping: dict[str, SupportedNativeProviders] = {
-                "openai": "openai",
-                "anthropic": "anthropic",
-                "claude": "anthropic",
-                "azure": "azure",
-                "azure_openai": "azure",
-                "google": "gemini",
-                "gemini": "gemini",
-                "bedrock": "bedrock",
-                "aws": "bedrock",
-            }
-
-            canonical_provider = provider_mapping.get(prefix.lower())
+            canonical_provider = PROVIDER_MAPPING.get(prefix.lower())
 
             if canonical_provider and cls._validate_model_in_constants(
                 model_part, canonical_provider

lib/crewai/src/crewai/llm/providers/anthropic/completion.py

@@ -4,6 +4,7 @@
 import logging
 from typing import TYPE_CHECKING, Any, cast
 
+from dotenv import load_dotenv
 import httpx
 from pydantic import BaseModel, Field, PrivateAttr, model_validator
 from typing_extensions import Self
@@ -21,20 +22,24 @@
 
 
 if TYPE_CHECKING:
+    from anthropic.types import Message
+
     from crewai.agent.core import Agent
     from crewai.task import Task
 
 
 try:
     from anthropic import Anthropic
-    from anthropic.types import Message
     from anthropic.types.tool_use_block import ToolUseBlock
 except ImportError:
     raise ImportError(
         'Anthropic native provider not available, to install: uv add "crewai[anthropic]"'
     ) from None
 
 
+load_dotenv()
+
+
 class AnthropicCompletion(BaseLLM):
     """Anthropic native completion implementation.
 
@@ -66,19 +71,17 @@ class AnthropicCompletion(BaseLLM):
     top_p: float | None = Field(default=None, description="Nucleus sampling parameter")
     stream: bool = Field(default=False, description="Enable streaming responses")
     client_params: dict[str, Any] | None = Field(
-        default=None, description="Additional Anthropic client parameters"
-    )
-    client: Anthropic = Field(
-        default_factory=Anthropic, exclude=True, description="Anthropic client instance"
+        default_factory=dict, description="Additional Anthropic client parameters"
     )
+    _client: Anthropic = PrivateAttr(default=None)  # type: ignore[assignment]
 
     _is_claude_3: bool = PrivateAttr(default=False)
     _supports_tools: bool = PrivateAttr(default=False)
 
     @model_validator(mode="after")
     def setup_client(self) -> Self:
         """Initialize the Anthropic client and model-specific settings."""
-        self.client = Anthropic(**self._get_client_params())
+        self._client = Anthropic(**self._get_client_params())
 
         self._is_claude_3 = "claude-3" in self.model.lower()
         self._supports_tools = self._is_claude_3
@@ -98,9 +101,6 @@ def supports_tools(self) -> bool:
     def _get_client_params(self) -> dict[str, Any]:
         """Get client parameters."""
 
-        if self.api_key is None:
-            raise ValueError("ANTHROPIC_API_KEY is required")
-
         client_params = {
             "api_key": self.api_key,
             "base_url": self.base_url,
@@ -330,7 +330,7 @@ def _handle_completion(
             params["tool_choice"] = {"type": "tool", "name": "structured_output"}
 
         try:
-            response: Message = self.client.messages.create(**params)
+            response: Message = self._client.messages.create(**params)
 
         except Exception as e:
             if is_context_length_exceeded(e):
@@ -424,7 +424,7 @@ def _handle_streaming_completion(
         stream_params = {k: v for k, v in params.items() if k != "stream"}
 
         # Make streaming API call
-        with self.client.messages.stream(**stream_params) as stream:
+        with self._client.messages.stream(**stream_params) as stream:
             for event in stream:
                 if hasattr(event, "delta") and hasattr(event.delta, "text"):
                     text_delta = event.delta.text
@@ -552,7 +552,7 @@ def _handle_tool_use_conversation(
 
         try:
             # Send tool results back to Claude for final response
-            final_response: Message = self.client.messages.create(**follow_up_params)
+            final_response: Message = self._client.messages.create(**follow_up_params)
 
             # Track token usage for follow-up call
             follow_up_usage = self._extract_anthropic_token_usage(final_response)