nat.plugins.mcp.auth.token_storage

Attributes

logger

Classes

TokenStorageBase	Abstract base class for token storage implementations.
ObjectStoreTokenStorage	Token storage implementation backed by a NeMo Agent toolkit object store.
InMemoryTokenStorage	In-memory token storage using NeMo Agent toolkit's built-in object store.

Module Contents

logger

class TokenStorageBase

Bases: abc.ABC

Abstract base class for token storage implementations.

Token storage implementations handle the secure persistence of authentication tokens for MCP OAuth2 flows. Implementations can use various backends such as object stores, databases, or in-memory storage.

abstractmethod store(

user_id: str,

auth_result: nat.data_models.authentication.AuthResult,

) → None

Async:

Store an authentication result for a user.

Args:

user_id: The unique identifier for the user auth_result: The authentication result to store

abstractmethod retrieve(

user_id: str,

) → nat.data_models.authentication.AuthResult | None

Async:

Retrieve an authentication result for a user.

Args:

user_id: The unique identifier for the user

Returns:

The authentication result if found, None otherwise

abstractmethod delete(user_id: str) → None

Async:

Delete an authentication result for a user.

Args:

user_id: The unique identifier for the user

abstractmethod clear_all() → None

Async:

Clear all stored authentication results.

class ObjectStoreTokenStorage(

object_store: nat.object_store.interfaces.ObjectStore,

)

Bases: TokenStorageBase

Token storage implementation backed by a NeMo Agent toolkit object store.

This implementation uses the object store infrastructure to persist tokens, which provides encryption at rest, access controls, and persistence across restarts when using backends like S3, MySQL, or Redis.

Initialize the object store token storage.

Args:

object_store: The object store instance to use for token persistence

_object_store

_get_key(user_id: str) → str

Generate the object store key for a user’s token.

Uses SHA256 hash to ensure the key is S3-compatible and doesn’t contain special characters like “://” that are invalid in object keys.

Args:

user_id: The user identifier

Returns:

The object store key

async store(

user_id: str,

auth_result: nat.data_models.authentication.AuthResult,

) → None

Store an authentication result in the object store.

Args:

user_id: The unique identifier for the user auth_result: The authentication result to store

async retrieve(

user_id: str,

) → nat.data_models.authentication.AuthResult | None

Retrieve an authentication result from the object store.

Args:

user_id: The unique identifier for the user

Returns:

The authentication result if found, None otherwise

async delete(user_id: str) → None

Delete an authentication result from the object store.

Args:

user_id: The unique identifier for the user

async clear_all() → None

Clear all stored authentication results.

Note: This implementation does not support clearing all tokens as the object store interface doesn’t provide a list operation. Individual tokens must be deleted explicitly.

class InMemoryTokenStorage

Bases: TokenStorageBase

In-memory token storage using NeMo Agent toolkit’s built-in object store.

This implementation uses the in-memory object store for token persistence, which provides a secure default option that doesn’t require external storage configuration. Tokens are stored in memory and cleared when the process exits.

Initialize the in-memory token storage.

_object_store

_storage

async store(

user_id: str,

auth_result: nat.data_models.authentication.AuthResult,

) → None

Store an authentication result in memory.

Args:

user_id: The unique identifier for the user auth_result: The authentication result to store

async retrieve(

user_id: str,

) → nat.data_models.authentication.AuthResult | None

Retrieve an authentication result from memory.

Args:

user_id: The unique identifier for the user

Returns:

The authentication result if found, None otherwise

async delete(user_id: str) → None

Delete an authentication result from memory.

Args:

user_id: The unique identifier for the user

async clear_all() → None

Clear all stored authentication results from memory.