feat: add support for llm message interceptor hooks

Author: greysonlalonde

docs/en/concepts/llms.mdx

@@ -1200,6 +1200,52 @@ Learn how to get the most out of your LLM configuration:
     )
     ```
   </Accordion>
+
+  <Accordion title="Transport Interceptors">
+    CrewAI provides message interceptors for several providers, allowing you to hook into request/response cycles at the transport layer.
+
+    **Supported Providers:**
+    - ✅ OpenAI
+    - ✅ Anthropic
+
+    **Basic Usage:**
+    ```python
+import httpx
+from crewai import LLM
+from crewai.llms.hooks import BaseInterceptor
+
+class CustomInterceptor(BaseInterceptor[httpx.Request, httpx.Response]):
+    """Custom interceptor to modify requests and responses."""
+
+    def on_outbound(self, request: httpx.Request) -> httpx.Request:
+        """Print request before sending to the LLM provider."""
+        print(request)
+        return request
+
+    def on_inbound(self, response: httpx.Response) -> httpx.Response:
+        """Process response after receiving from the LLM provider."""
+        print(f"Status: {response.status_code}")
+        print(f"Response time: {response.elapsed}")
+        return response
+
+# Use the interceptor with an LLM
+llm = LLM(
+    model="openai/gpt-4o",
+    interceptor=CustomInterceptor()
+)
+    ```
+
+    **Important Notes:**
+    - Both methods must return the received object or type of object.
+    - Modifying received objects may result in unexpected behavior or application crashes.
+    - Not all providers support interceptors - check the supported providers list above
+
+    <Info>
+      Interceptors operate at the transport layer. This is particularly useful for:
+      - Message transformation and filtering
+      - Debugging API interactions
+    </Info>
+  </Accordion>
 </AccordionGroup>
 
 ## Common Issues and Solutions

lib/crewai/src/crewai/llm.py

@@ -20,6 +20,7 @@
 )
 
 from dotenv import load_dotenv
+import httpx
 from pydantic import BaseModel, Field
 from typing_extensions import Self
 
@@ -53,6 +54,7 @@
     from litellm.utils import supports_response_schema
 
     from crewai.agent.core import Agent
+    from crewai.llms.hooks.base import BaseInterceptor
     from crewai.task import Task
     from crewai.tools.base_tool import BaseTool
     from crewai.utilities.types import LLMMessage
@@ -334,6 +336,8 @@ def __new__(cls, model: str, is_litellm: bool = False, **kwargs: Any) -> LLM:
                 return cast(
                     Self, native_class(model=model_string, provider=provider, **kwargs)
                 )
+            except NotImplementedError:
+                raise
             except Exception as e:
                 raise ImportError(f"Error importing native provider: {e}") from e
 
@@ -403,6 +407,7 @@ def __init__(
         callbacks: list[Any] | None = None,
         reasoning_effort: Literal["none", "low", "medium", "high"] | None = None,
         stream: bool = False,
+        interceptor: BaseInterceptor[httpx.Request, httpx.Response] | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize LLM instance.
@@ -442,6 +447,7 @@ def __init__(
         self.additional_params = kwargs
         self.is_anthropic = self._is_anthropic_model(model)
         self.stream = stream
+        self.interceptor = interceptor
 
         litellm.drop_params = True
 

lib/crewai/src/crewai/llms/hooks/__init__.py

@@ -0,0 +1,6 @@
+"""Interceptor contracts for crewai"""
+
+from crewai.llms.hooks.base import BaseInterceptor
+
+
+__all__ = ["BaseInterceptor"]

lib/crewai/src/crewai/llms/hooks/base.py

@@ -0,0 +1,82 @@
+"""Base classes for LLM transport interceptors.
+
+This module provides abstract base classes for intercepting and modifying
+outbound and inbound messages at the transport level.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Generic, TypeVar
+
+
+T = TypeVar("T")
+U = TypeVar("U")
+
+
+class BaseInterceptor(ABC, Generic[T, U]):
+    """Abstract base class for intercepting transport-level messages.
+
+    Provides hooks to intercept and modify outbound and inbound messages
+    at the transport layer.
+
+    Type parameters:
+        T: Outbound message type (e.g., httpx.Request)
+        U: Inbound message type (e.g., httpx.Response)
+
+    Example:
+        >>> class CustomInterceptor(BaseInterceptor[httpx.Request, httpx.Response]):
+        ...     def on_outbound(self, message: httpx.Request) -> httpx.Request:
+        ...         message.headers["X-Custom-Header"] = "value"
+        ...         return message
+        ...
+        ...     def on_inbound(self, message: httpx.Response) -> httpx.Response:
+        ...         print(f"Status: {message.status_code}")
+        ...         return message
+    """
+
+    @abstractmethod
+    def on_outbound(self, message: T) -> T:
+        """Intercept outbound message before sending.
+
+        Args:
+            message: Outbound message object.
+
+        Returns:
+            Modified message object.
+        """
+        ...
+
+    @abstractmethod
+    def on_inbound(self, message: U) -> U:
+        """Intercept inbound message after receiving.
+
+        Args:
+            message: Inbound message object.
+
+        Returns:
+            Modified message object.
+        """
+        ...
+
+    async def aon_outbound(self, message: T) -> T:
+        """Async version of on_outbound.
+
+        Args:
+            message: Outbound message object.
+
+        Returns:
+            Modified message object.
+        """
+        raise NotImplementedError
+
+    async def aon_inbound(self, message: U) -> U:
+        """Async version of on_inbound.
+
+        Args:
+            message: Inbound message object.
+
+        Returns:
+            Modified message object.
+        """
+        raise NotImplementedError

lib/crewai/src/crewai/llms/hooks/transport.py

@@ -0,0 +1,87 @@
+"""HTTP transport implementations for LLM request/response interception.
+
+This module provides internal transport classes that integrate with BaseInterceptor
+to enable request/response modification at the transport level.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import httpx
+
+
+if TYPE_CHECKING:
+    from crewai.llms.hooks.base import BaseInterceptor
+
+
+class HTTPTransport(httpx.HTTPTransport):
+    """HTTP transport that uses an interceptor for request/response modification.
+
+    This transport is used internally when a user provides a BaseInterceptor.
+    Users should not instantiate this class directly - instead, pass an interceptor
+    to the LLM client and this transport will be created automatically.
+    """
+
+    def __init__(
+        self,
+        interceptor: BaseInterceptor[httpx.Request, httpx.Response],
+        **kwargs: Any,
+    ) -> None:
+        """Initialize transport with interceptor.
+
+        Args:
+            interceptor: HTTP interceptor for modifying raw request/response objects.
+            **kwargs: Additional arguments passed to httpx.HTTPTransport.
+        """
+        super().__init__(**kwargs)
+        self.interceptor = interceptor
+
+    def handle_request(self, request: httpx.Request) -> httpx.Response:
+        """Handle request with interception.
+
+        Args:
+            request: The HTTP request to handle.
+
+        Returns:
+            The HTTP response.
+        """
+        request = self.interceptor.on_outbound(request)
+        response = super().handle_request(request)
+        return self.interceptor.on_inbound(response)
+
+
+class AsyncHTTPransport(httpx.AsyncHTTPTransport):
+    """Async HTTP transport that uses an interceptor for request/response modification.
+
+    This transport is used internally when a user provides a BaseInterceptor.
+    Users should not instantiate this class directly - instead, pass an interceptor
+    to the LLM client and this transport will be created automatically.
+    """
+
+    def __init__(
+        self,
+        interceptor: BaseInterceptor[httpx.Request, httpx.Response],
+        **kwargs: Any,
+    ) -> None:
+        """Initialize async transport with interceptor.
+
+        Args:
+            interceptor: HTTP interceptor for modifying raw request/response objects.
+            **kwargs: Additional arguments passed to httpx.AsyncHTTPTransport.
+        """
+        super().__init__(**kwargs)
+        self.interceptor = interceptor
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        """Handle async request with interception.
+
+        Args:
+            request: The HTTP request to handle.
+
+        Returns:
+            The HTTP response.
+        """
+        request = await self.interceptor.aon_outbound(request)
+        response = await super().handle_async_request(request)
+        return await self.interceptor.aon_inbound(response)

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

@@ -1,26 +1,30 @@
 from __future__ import annotations
 
-
 import json
 import logging
 import os
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
 from pydantic import BaseModel
 
 from crewai.events.types.llm_events import LLMCallType
 from crewai.llms.base_llm import BaseLLM
+from crewai.llms.hooks.transport import HTTPTransport
 from crewai.utilities.agent_utils import is_context_length_exceeded
 from crewai.utilities.exceptions.context_window_exceeding_exception import (
     LLMContextLengthExceededError,
 )
 from crewai.utilities.types import LLMMessage
 
 
+if TYPE_CHECKING:
+    from crewai.llms.hooks.base import BaseInterceptor
+
 try:
     from anthropic import Anthropic
     from anthropic.types import Message
     from anthropic.types.tool_use_block import ToolUseBlock
+    import httpx
 except ImportError:
     raise ImportError(
         'Anthropic native provider not available, to install: uv add "crewai[anthropic]"'
@@ -47,7 +51,8 @@ def __init__(
         stop_sequences: list[str] | None = None,
         stream: bool = False,
         client_params: dict[str, Any] | None = None,
-        **kwargs,
+        interceptor: BaseInterceptor[httpx.Request, httpx.Response] | None = None,
+        **kwargs: Any,
     ):
         """Initialize Anthropic chat completion client.
 
@@ -63,13 +68,15 @@ def __init__(
             stop_sequences: Stop sequences (Anthropic uses stop_sequences, not stop)
             stream: Enable streaming responses
             client_params: Additional parameters for the Anthropic client
+            interceptor: HTTP interceptor for modifying requests/responses at transport level.
             **kwargs: Additional parameters
         """
         super().__init__(
             model=model, temperature=temperature, stop=stop_sequences or [], **kwargs
         )
 
         # Client params
+        self.interceptor = interceptor
         self.client_params = client_params
         self.base_url = base_url
         self.timeout = timeout
@@ -102,6 +109,11 @@ def _get_client_params(self) -> dict[str, Any]:
             "max_retries": self.max_retries,
         }
 
+        if self.interceptor:
+            transport = HTTPTransport(interceptor=self.interceptor)
+            http_client = httpx.Client(transport=transport)
+            client_params["http_client"] = http_client  # type: ignore[assignment]
+
         if self.client_params:
             client_params.update(self.client_params)
 
@@ -110,7 +122,7 @@ def _get_client_params(self) -> dict[str, Any]:
     def call(
         self,
         messages: str | list[LLMMessage],
-        tools: list[dict] | None = None,
+        tools: list[dict[str, Any]] | None = None,
         callbacks: list[Any] | None = None,
         available_functions: dict[str, Any] | None = None,
         from_task: Any | None = None,
@@ -133,7 +145,7 @@ def call(
         try:
             # Emit call started event
             self._emit_call_started_event(
-                messages=messages,  # type: ignore[arg-type]
+                messages=messages,
                 tools=tools,
                 callbacks=callbacks,
                 available_functions=available_functions,
@@ -143,7 +155,7 @@ def call(
 
             # Format messages for Anthropic
             formatted_messages, system_message = self._format_messages_for_anthropic(
-                messages  # type: ignore[arg-type]
+                messages
             )
 
             # Prepare completion parameters
@@ -181,7 +193,7 @@ def _prepare_completion_params(
         self,
         messages: list[LLMMessage],
         system_message: str | None = None,
-        tools: list[dict] | None = None,
+        tools: list[dict[str, Any]] | None = None,
     ) -> dict[str, Any]:
         """Prepare parameters for Anthropic messages API.
 
@@ -218,7 +230,9 @@ def _prepare_completion_params(
 
         return params
 
-    def _convert_tools_for_interference(self, tools: list[dict]) -> list[dict]:
+    def _convert_tools_for_interference(
+        self, tools: list[dict[str, Any]]
+    ) -> list[dict[str, Any]]:
         """Convert CrewAI tool format to Anthropic tool use format."""
         anthropic_tools = []