Function connect_with_oauth

tests/interaction/auth/_harness.py:397–476

Connect a `Client` to a server's bearer-gated streamable-HTTP app, completing OAuth in process. Yields the connected `Client` and the `HeadlessOAuth` whose `authorize_url` records what the SDK put on the authorize request. `on_request` records every HTTP request the underlying `httpx.As

(
    server: Server,
    *,
    provider: InMemoryAuthorizationServerProvider,
    settings: AuthSettings | None = None,
    storage: InMemoryTokenStorage | None = None,
    client_metadata: OAuthClientMetadata | None = None,
    client_metadata_url: str | None = None,
    headless: HeadlessOAuth | None = None,
    auth: httpx.Auth | None = None,
    verify_tokens: bool = True,
    app_shim: Callable[[ASGIApp], ASGIApp] | None = None,
    on_request: Callable[[httpx.Request], None] | None = None,
)

@asynccontextmanager
async def connect_with_oauth(
    server: Server,
    *,
    provider: InMemoryAuthorizationServerProvider,
    settings: AuthSettings | None = None,
    storage: InMemoryTokenStorage | None = None,
    client_metadata: OAuthClientMetadata | None = None,
    client_metadata_url: str | None = None,
    headless: HeadlessOAuth | None = None,
    auth: httpx.Auth | None = None,
    verify_tokens: bool = True,
    app_shim: Callable[[ASGIApp], ASGIApp] | None = None,
    on_request: Callable[[httpx.Request], None] | None = None,
) -> AsyncIterator[tuple[Client, HeadlessOAuth]]:
    """Connect a `Client` to a server's bearer-gated streamable-HTTP app, completing OAuth in process.

    Yields the connected `Client` and the `HeadlessOAuth` whose `authorize_url` records what the
    SDK put on the authorize request. `on_request` records every HTTP request the underlying
    `httpx.AsyncClient` issues, including those yielded from inside the auth flow.

    `headless`: supply a pre-configured `HeadlessOAuth` to override the callback behaviour
    (state mismatch, error redirects). `verify_tokens=False` mounts the MCP endpoint without
    the bearer middleware so a flow driven by a shimmed 401 completes regardless of the granted
    scopes. `app_shim` wraps the built Starlette app before it reaches the bridge transport,
    for tests that need to intercept or rewrite specific server responses.

    `auth`: supply a pre-built `httpx.Auth` (such as `ClientCredentialsOAuthProvider`) to use
    instead of constructing the default `OAuthClientProvider`; in that case `storage`,
    `client_metadata`, `client_metadata_url`, and `headless` are unused (the yielded
    `HeadlessOAuth` is never invoked and its `authorize_url` stays None).
    """
    settings = settings if settings is not None else auth_settings()
    storage = storage if storage is not None else InMemoryTokenStorage()
    client_metadata = client_metadata if client_metadata is not None else oauth_client_metadata()
    headless = headless if headless is not None else HeadlessOAuth()

    oauth = (
        auth
        if auth is not None
        else OAuthClientProvider(
            server_url=f"{BASE_URL}/mcp",
            client_metadata=client_metadata,
            storage=storage,
            redirect_handler=headless.redirect_handler,
            callback_handler=headless.callback_handler,
            client_metadata_url=client_metadata_url,
        )
    )

    app: ASGIApp = server.streamable_http_app(
        auth=settings,
        token_verifier=ProviderTokenVerifier(provider) if verify_tokens else None,
        auth_server_provider=provider,
        transport_security=NO_DNS_REBINDING_PROTECTION,
    )
    if app_shim is not None:
        app = app_shim(app)