nat.plugins.mcp.auth.auth_provider#

Attributes#

logger

Classes#

`OAuth2Endpoints`	OAuth2 endpoints discovered from MCP server.
`OAuth2Credentials`	OAuth2 client credentials from registration.
`DiscoverOAuth2Endpoints`	MCP-SDK parity discovery flow:
`DynamicClientRegistration`	Dynamic client registration utility.
`MCPOAuth2Provider`	MCP OAuth2 authentication provider that delegates to NAT framework.

Module Contents#

logger#

class OAuth2Endpoints(/, **data: Any)#

Bases: pydantic.BaseModel

OAuth2 endpoints discovered from MCP server.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

authorization_url: pydantic.HttpUrl = None#

token_url: pydantic.HttpUrl = None#

registration_url: pydantic.HttpUrl | None = None#

scopes: list[str] | None = None#

class OAuth2Credentials(/, **data: Any)#

Bases: pydantic.BaseModel

OAuth2 client credentials from registration.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

client_id: str = None#

client_secret: str | None = None#

class DiscoverOAuth2Endpoints( config: nat.plugins.mcp.auth.auth_provider_config.MCPOAuth2ProviderConfig, )#

MCP-SDK parity discovery flow:

If 401 + WWW-Authenticate has resource_metadata (RFC 9728), fetch it.
Else fetch RS well-known /.well-known/oauth-protected-resource.
If PR metadata lists authorization_servers, pick first as issuer.
Do path-aware RFC 8414 / OIDC discovery against issuer (or server base).

config#

_cached_endpoints: OAuth2Endpoints | None = None#

_flow_handler: nat.plugins.mcp.auth.auth_flow_handler.MCPAuthenticationFlowHandler#

async discover( response: httpx.Response | None = None, ) → tuple[OAuth2Endpoints, bool]#

Discover OAuth2 endpoints from MCP server.

Args:: reason: The reason for the discovery. www_authenticate: The WWW-Authenticate header from a 401 response.
Returns:: A tuple of OAuth2Endpoints and a boolean indicating if the endpoints have changed.

_authorization_base_url() → str#: Get the authorization base URL from the MCP server URL.

_extract_from_www_authenticate_header(hdr: str) → str | None#: Extract the resource_metadata URL from the WWW-Authenticate header.

async _fetch_pr_issuer(url: str) → str | None#: Fetch RFC 9728 Protected Resource Metadata and return the first issuer (authorization_server).

async _discover_via_issuer_or_base( base_or_issuer: str, ) → OAuth2Endpoints | None#: Perform path-aware RFC 8414 / OIDC discovery given an issuer or base URL.

_build_path_aware_discovery_urls(base_or_issuer: str) → list[str]#: Build path-aware discovery URLs.

class DynamicClientRegistration( config: nat.plugins.mcp.auth.auth_provider_config.MCPOAuth2ProviderConfig, )#

Dynamic client registration utility.

config#

_authorization_base_url() → str#: Get the authorization base URL from the MCP server URL.

async register( endpoints: OAuth2Endpoints, scopes: list[str] | None, ) → OAuth2Credentials#: Register an OAuth2 client with the Authorization Server using OIDC client registration.

class MCPOAuth2Provider( config: nat.plugins.mcp.auth.auth_provider_config.MCPOAuth2ProviderConfig, builder=None, )#

Bases: nat.authentication.interfaces.AuthProviderBase[nat.plugins.mcp.auth.auth_provider_config.MCPOAuth2ProviderConfig]

MCP OAuth2 authentication provider that delegates to NAT framework.

Initialize the AuthProviderBase with the given configuration.

Args:: config (AuthProviderBaseConfig): Configuration items for authentication.

_builder = None#

_discoverer#

_cached_endpoints: OAuth2Endpoints | None = None#

_registrar#

_cached_credentials: OAuth2Credentials | None = None#

_auth_code_provider = None#

_flow_handler#

_auth_callback = None#

_token_storage = None#

_token_storage_object_store_name = None#

_set_custom_auth_callback( auth_callback: collections.abc.Callable[[nat.authentication.oauth2.oauth2_auth_code_flow_provider_config.OAuth2AuthCodeFlowProviderConfig, nat.authentication.interfaces.AuthFlowType], collections.abc.Awaitable[nat.authentication.interfaces.AuthenticatedContext]], )#: Set the custom authentication callback.

async authenticate(

user_id: str | None = None,

**kwargs,

) → nat.data_models.authentication.AuthResult#

Authenticate using MCP OAuth2 flow via NAT framework.

If response is provided in kwargs (typically from a 401), performs: 1. Dynamic endpoints discovery (RFC9728 + RFC 8414 + OIDC) 2. Client registration (RFC7591) 3. Authentication

Otherwise, performs standard authentication flow.

property _effective_scopes: list[str]#: Get the effective scopes to be used for the authentication.

async _discover_and_register(response: httpx.Response | None = None)#: Discover OAuth2 endpoints and register an OAuth2 client with the Authorization Server using OIDC client registration.

async _nat_oauth2_authenticate( user_id: str | None = None, ) → nat.data_models.authentication.AuthResult#: Perform the OAuth2 flow using MCP-specific authentication flow handler.