_DefaultHttpxClient Class — anthropic-sdk-python Architecture

Source Code

src/anthropic/_base_client.py lines 838–884

class _DefaultHttpxClient(httpx.Client):
    def __init__(self, **kwargs: Any) -> None:
        kwargs.setdefault("timeout", DEFAULT_TIMEOUT)
        kwargs.setdefault("limits", DEFAULT_CONNECTION_LIMITS)
        kwargs.setdefault("follow_redirects", True)

        if "transport" not in kwargs:
            socket_options: List[Tuple[int, int, Union[int, bool]]] = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, True)]

            TCP_KEEPINTVL = getattr(socket, "TCP_KEEPINTVL", None)

            if TCP_KEEPINTVL is not None:
                socket_options.append((socket.IPPROTO_TCP, TCP_KEEPINTVL, 60))
            elif sys.platform == "darwin":
                TCP_KEEPALIVE = getattr(socket, "TCP_KEEPALIVE", 0x10)
                socket_options.append((socket.IPPROTO_TCP, TCP_KEEPALIVE, 60))

            TCP_KEEPCNT = getattr(socket, "TCP_KEEPCNT", None)
            if TCP_KEEPCNT is not None:
                socket_options.append((socket.IPPROTO_TCP, TCP_KEEPCNT, 5))

            TCP_KEEPIDLE = getattr(socket, "TCP_KEEPIDLE", None)
            if TCP_KEEPIDLE is not None:
                socket_options.append((socket.IPPROTO_TCP, TCP_KEEPIDLE, 60))

            proxy_map = {key: None if url is None else Proxy(url=url) for key, url in get_environment_proxies().items()}

            transport_kwargs = {
                arg: kwargs[arg] for arg in ("verify", "cert", "trust_env", "http1", "http2", "limits") if arg in kwargs
            }

            transport_kwargs["socket_options"] = socket_options

            proxy_mounts = {
                key: None if proxy is None else HTTPTransport(proxy=proxy, **transport_kwargs)
                for key, proxy in proxy_map.items()
            }
            default_transport = HTTPTransport(**transport_kwargs)

            # Prioritize the mounts set by the user over the environment variables.
            proxy_mounts.update(kwargs.get("mounts", {}))
            kwargs["mounts"] = proxy_mounts

            # Sets the default transport so that HTTPX won't automatically configure proxies.
            kwargs["transport"] = default_transport

        super().__init__(**kwargs)