Motivation and Context

As described in #652, the streamable HTTP client currently can be configured to include a static dictionary of headers with each request, including an Authorization header for authentication. However, to support authenticating to an MCP server using short-lived access tokens, the client needs the ability to obtain a fresh token across requests to the MCP Server. This PR introduces an AuthClientProvider base class to support custom authentication and adds the ability to pass in an instance of AuthClientProvider when instantiating a StreamableHTTPClient. If specified, the headers returned from auth_client_provider.get_headers() are passed along to each request to the Server. This resembles the setup in the TypeScript SDK.

In a follow-up PR, we'll complete parity with the TypeScript SDK & enhance the interface to support Oauth 2.0 flows, so that developers need only implement key pieces of the Oauth 2.0 authorization code flow rather than a full-fledged get_headers implementation, while still preserving the ability to override get_headers to support authenticating to an MCP server from environments where an oauth token or other auth is already available

How Has This Been Tested?

Unit Tests
For e2e test, built an example implementation of an interface against a OAuth compliant server

import asyncio
from datetime import datetime, timedelta
from typing import Optional

from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamablehttp_client, AuthClientProvider

class CustomAuthClientProvider(AuthClientProvider):
    def __init__(self):
        # Initialize Auth Provider

    async def get_headers(self) -> str:
        # Implement getting dynamic headers

# Auth Provider specified here
auth_client_provider = CustomAuthClientProvider()

async def query_mcp():
    async with streamablehttp_client(
        url="<URL>",
        auth_client_provider=auth_client_provider,
    ) as (read_stream, write_stream, _):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            tools = await session.list_tools()
            print(tools)


async def main():
    await query_mcp()


if __name__ == "__main__":
    asyncio.run(main())

Breaking Changes

No

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the MCP Documentation
• My code follows the repository's style guidelines
• New and existing tests pass locally
• I have added appropriate error handling
• I have added or updated documentation as needed

Additional context