Function create_mcp_http_client

src/mcp/shared/_httpx_utils.py:23–95

Create a standardized httpx AsyncClient with MCP defaults. Always enables follow_redirects and applies an SSE-friendly default timeout. Args: headers: Optional headers to include with all requests. timeout: Request timeout as httpx.Timeout object. Defaults to 30s for

(
    headers: dict[str, str] | None = None,
    timeout: httpx.Timeout | None = None,
    auth: httpx.Auth | None = None,
)

def create_mcp_http_client(
    headers: dict[str, str] | None = None,
    timeout: httpx.Timeout | None = None,
    auth: httpx.Auth | None = None,
) -> httpx.AsyncClient:
    """Create a standardized httpx AsyncClient with MCP defaults.

    Always enables follow_redirects and applies an SSE-friendly default timeout.

    Args:
        headers: Optional headers to include with all requests.
        timeout: Request timeout as httpx.Timeout object. Defaults to 30s for
            connect/write/pool and 300s for read (for long-lived SSE streams).
        auth: Optional authentication handler.

    Returns:
        Configured httpx.AsyncClient instance with MCP defaults.

    Note:
        The returned AsyncClient must be used as a context manager to ensure
        proper cleanup of connections.

    Example:
        Basic usage with MCP defaults:

        ```python
        async with create_mcp_http_client() as client:
            response = await client.get("https://api.example.com")
        ```

        With custom headers:

        ```python
        headers = {"Authorization": "Bearer token"}
        async with create_mcp_http_client(headers) as client:
            response = await client.get("/endpoint")
        ```

        With both custom headers and timeout:

        ```python
        timeout = httpx.Timeout(60.0, read=300.0)
        async with create_mcp_http_client(headers, timeout) as client:
            response = await client.get("/long-request")
        ```

        With authentication:

        ```python
        from httpx import BasicAuth
        auth = BasicAuth(username="user", password="pass")
        async with create_mcp_http_client(headers, timeout, auth) as client:
            response = await client.get("/protected-endpoint")
        ```
    """
    # Set MCP defaults
    kwargs: dict[str, Any] = {"follow_redirects": True}