Feature Request: Support for Request/Response Interceptors

[Nisarg38]

Confirm this is a feature request for the Python library and not the underlying OpenAI API.

• This is a feature request for the Python library

Describe the feature or improvement you're requesting

Description

It would be great to have built-in support for interceptors in the openai-python library, allowing users to modify requests before they are sent and process responses after they are received. This would provide a clean and extensible way to add logging, metrics, retries, authentication adjustments, or any other middleware-like functionality without modifying the core library code.

Proposed Solution

Introduce a mechanism similar to request/response interceptors found in other API client libraries. This could be achieved by adding configurable hooks or middleware that developers can pass when initializing the OpenAI client.

Additional context

Implementation

This implementation introduces a flexible interceptor system for handling request modifications before they are sent and response processing after they are received.

Code Implementation

@dataclass
class InterceptorRequest:
    """Request data container for interceptor processing"""

    method: str
    url: str
    headers: Dict[str, str]
    params: Optional[Dict[str, Any]] = None
    body: Optional[Union[Body, bytes]] = None


@dataclass
class InterceptorResponse(Generic[T]):
    """Response data container for interceptor processing"""

    status_code: int
    headers: Dict[str, str]
    body: T
    request: InterceptorRequest
    raw_response: httpx.Response


class Interceptor(ABC):
    """Base class for request/response interceptors"""

    @abstractmethod
    def before_request(self, request: InterceptorRequest) -> InterceptorRequest:
        """Process request before sending"""
        pass

    @abstractmethod
    def after_response(self, response: InterceptorResponse[T]) -> InterceptorResponse[T]:
        """Process response after receiving"""
        pass


class InterceptorChain:
    """Chain of interceptors for sequential request/response processing"""

    def __init__(self, interceptors: Optional[list[Interceptor]] = None):
        self._interceptors = interceptors or []

    def add_interceptor(self, interceptor: Interceptor) -> None:
        """Adds an interceptor to the chain"""
        self._interceptors.append(interceptor)

    def execute_before_request(self, request: InterceptorRequest) -> InterceptorRequest:
        """Executes all interceptors before sending the request"""
        current_request = request
        for interceptor in self._interceptors:
            try:
                current_request = interceptor.before_request(current_request)
            except Exception:
                continue
        return current_request

    def execute_after_response(self, response: InterceptorResponse[T]) -> InterceptorResponse[T]:
        """Executes all interceptors after receiving the response"""
        current_response = response
        for interceptor in self._interceptors:
            try:
                current_response = interceptor.after_response(current_response)
            except Exception:
                continue
        return current_response

Related Pull Request

I’ve submitted a PR implementing this feature: #2032. Would love feedback and thoughts on how this could be integrated!

[RobertCraigie]

Thanks for the detailed suggestion and putting up a PR! However we're going to defer adding something like this for now as you can achieve the same thing with httpx's event hooks https://www.python-httpx.org/advanced/event-hooks/

[Nisarg38]

I'll take a look. I am trying to add a functionality and it might work with event hook. Thanks for the help!