nat.plugins.mcp.auth.service_account.token_client

Attributes

logger

Classes

ServiceAccountTokenClient

Generic OAuth2 client credentials token client for service accounts.

Module Contents

logger

class ServiceAccountTokenClient(

client_id: str,

client_secret: pydantic.SecretStr,

token_url: str,

scopes: str,

token_cache_buffer_seconds: int = 300,

)

Generic OAuth2 client credentials token client for service accounts.

Implements standard OAuth2 client credentials flow with token caching.

Initialize service account token client.

Args:

client_id: OAuth2 client identifier client_secret: OAuth2 client secret (SecretStr) token_url: OAuth2 token endpoint URL scopes: Space-separated list of scopes token_cache_buffer_seconds: Seconds before expiry to refresh (default: 5 min)

client_id

client_secret

token_url

scopes

token_cache_buffer_seconds = 300

_cached_token: pydantic.SecretStr | None = None

_token_expires_at: datetime.datetime | None = None

_lock = None

property token_expires_at: datetime.datetime | None

async _get_lock() → asyncio.Lock

Lazy initialization of asyncio.Lock.

_is_token_valid() → bool

Check if cached token is still valid (with buffer time).

async get_access_token() → pydantic.SecretStr

Get OAuth2 access token, using cache if valid.

Returns:

Access token as SecretStr

Raises:

RuntimeError: If token acquisition fails

async _fetch_new_token() → pydantic.SecretStr

Fetch a new token from the OAuth2 token endpoint.

Returns:

New access token as SecretStr

Raises:

RuntimeError: If token request fails