nat.plugins.a2a.auth.credential_service#

Bridge NAT AuthProviderBase to A2A SDK CredentialService.

Attributes#

logger

Classes#

A2ACredentialService

Adapts NAT AuthProviderBase to A2A SDK CredentialService interface.

Module Contents#

logger#
class A2ACredentialService(
auth_provider: nat.authentication.interfaces.AuthProviderBase,
agent_card: a2a.types.AgentCard | None = None,
)#

Bases: a2a.client.CredentialService

Adapts NAT AuthProviderBase to A2A SDK CredentialService interface.

This class bridges NAT’s authentication system with the A2A SDK’s authentication mechanism, allowing A2A clients to use NAT’s auth providers (API Key, OAuth2, etc.) to authenticate with A2A agents.

The adapter: - Calls NAT auth provider to obtain credentials - Maps NAT credential types to A2A security scheme requirements - Handles token expiration and automatic refresh - Supports session-based multi-user authentication

Args:

auth_provider: NAT authentication provider instance agent_card: Agent card containing security scheme definitions

_auth_provider#
_agent_card = None#
_cached_auth_result: nat.data_models.authentication.AuthResult | None = None#
_auth_lock#
async get_credentials(
security_scheme_name: str,
context: a2a.client.ClientCallContext | None,
) → str | None#

Retrieve credentials for a security scheme.

This method: 1. Gets user_id from NAT context 2. Authenticates via NAT auth provider 3. Handles token expiration and refresh 4. Maps credentials to the requested security scheme

Args:

security_scheme_name: Name of the security scheme from AgentCard context: Client call context with optional session information

Returns:

Credential string or None if not available

async _authenticate(
user_id: str | None,
) → nat.data_models.authentication.AuthResult | None#

Authenticate and get credentials from NAT auth provider.

Handles token expiration by triggering re-authentication if needed. Uses a lock to prevent concurrent authentication requests and race conditions.

Args:

user_id: User identifier for authentication

Returns:

AuthResult with credentials or None on failure

_extract_credential_for_scheme(
auth_result: nat.data_models.authentication.AuthResult,
security_scheme_name: str,
) → str | None#

Extract appropriate credential based on security scheme type.

Maps NAT credential types to A2A security scheme requirements: - BearerTokenCred -> OAuth2, OIDC, HTTP Bearer - HeaderCred -> API Key in header - QueryCred -> API Key in query - CookieCred -> API Key in cookie - BasicAuthCred -> HTTP Basic

Args:

auth_result: Authentication result containing credentials security_scheme_name: Name of the security scheme

Returns:

Credential string or None

_get_scheme_definition(
scheme_name: str,
) → a2a.types.SecurityScheme | None#

Get security scheme definition from agent card.

Args:

scheme_name: Name of the security scheme

Returns:

SecurityScheme definition or None

_validate_provider_compatibility() → None#

Validate that the auth provider type is compatible with agent’s security schemes.

This performs early validation at connection time to fail fast if there’s a configuration mismatch between the NAT auth provider and the A2A agent’s security requirements.

Raises:

ValueError: If the provider is incompatible with all required security schemes

_is_provider_compatible_with_scheme(
scheme: a2a.types.SecurityScheme,
) → bool#

Check if the current auth provider can satisfy a security scheme.

Args:

scheme: Security scheme from agent card

Returns:

True if provider is compatible with the scheme

static _is_bearer_compatible(
scheme_def: a2a.types.SecurityScheme | None,
) → bool#

Check if security scheme accepts Bearer tokens.

Bearer tokens are compatible with: - OAuth2SecurityScheme - OpenIdConnectSecurityScheme - HTTPAuthSecurityScheme with scheme=’bearer’

Args:

scheme_def: Security scheme definition

Returns:

True if Bearer token is compatible

static _is_header_compatible(
scheme_def: a2a.types.SecurityScheme | None,
header_name: str,
) → bool#

Check if security scheme accepts header-based API keys.

Args:

scheme_def: Security scheme definition header_name: Name of the header containing the credential

Returns:

True if header credential is compatible

static _is_query_compatible(
scheme_def: a2a.types.SecurityScheme | None,
param_name: str,
) → bool#

Check if security scheme accepts query parameter API keys.

Args:

scheme_def: Security scheme definition param_name: Name of the query parameter

Returns:

True if query credential is compatible

Check if security scheme accepts cookie-based API keys.

Args:

scheme_def: Security scheme definition cookie_name: Name of the cookie

Returns:

True if cookie credential is compatible

static _is_basic_compatible(
scheme_def: a2a.types.SecurityScheme | None,
) → bool#

Check if security scheme accepts HTTP Basic authentication.

Args:

scheme_def: Security scheme definition

Returns:

True if Basic auth is compatible