Skip to content

Storage

Token storage abstractions and implementations.

PostgresTokenStorage is importable from this module but loaded lazily so that asyncpg is only required when actually used. Install the postgres extra to enable it: pip install mcp-authflow[postgres]

TokenStorage

TokenStorage

Bases: ABC

Abstract interface for MCP token storage.

initialize abstractmethod async 

initialize() -> None

Initialize the storage backend.

Source code in mcp_authflow/storage/base.py 
57
58
59
60
@abstractmethod
async def initialize(self) -> None:
    """Initialize the storage backend."""
    ...

close abstractmethod async 

close() -> None

Close the storage backend and clean up resources.

Source code in mcp_authflow/storage/base.py 
62
63
64
65
@abstractmethod
async def close(self) -> None:
    """Close the storage backend and clean up resources."""
    ...

store_token abstractmethod async 

store_token(
    token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None

Store an access token.

Parameters:

Name Type Description Default
token str

The access token string

 required
client_id str

OAuth client ID

 required
scopes list[str]

List of granted scopes

 required
expires_at int

Unix timestamp when token expires

 required
resource str | None

Optional RFC 8707 resource binding

 None
user_id UserId | None

Optional ID of the user who authorized the token. May be an int or a str so it can match the consumer's user primary key (see :data:~mcp_authflow.storage.base.UserId)

 None
Source code in mcp_authflow/storage/base.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@abstractmethod
async def store_token(
    self,
    token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None:
    """Store an access token.

    Args:
        token: The access token string
        client_id: OAuth client ID
        scopes: List of granted scopes
        expires_at: Unix timestamp when token expires
        resource: Optional RFC 8707 resource binding
        user_id: Optional ID of the user who authorized the token. May be an
            int or a str so it can match the consumer's user primary key
            (see :data:`~mcp_authflow.storage.base.UserId`)
    """
    ...

load_token abstractmethod async 

load_token(token: str) -> dict[str, Any] | None

Load an access token.

Parameters:

Name Type Description Default
token str

The access token string to look up

 required

Returns:

Type Description
dict[str, Any] | None

Token data dict if found and not expired, None otherwise
Source code in mcp_authflow/storage/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@abstractmethod
async def load_token(self, token: str) -> dict[str, Any] | None:
    """Load an access token.

    Args:
        token: The access token string to look up

    Returns:
        Token data dict if found and not expired, None otherwise
    """
    ...

delete_token abstractmethod async 

delete_token(token: str) -> None

Delete a token.

Parameters:

Name Type Description Default
token str

The access token string to delete

 required
Source code in mcp_authflow/storage/base.py 
103
104
105
106
107
108
109
110
@abstractmethod
async def delete_token(self, token: str) -> None:
    """Delete a token.

    Args:
        token: The access token string to delete
    """
    ...

store_refresh_token abstractmethod async 

store_refresh_token(
    refresh_token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None

Store a refresh token.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string

 required
client_id str

OAuth client ID

 required
scopes list[str]

List of granted scopes

 required
expires_at int

Unix timestamp when token expires

 required
resource str | None

Optional RFC 8707 resource binding

 None
user_id UserId | None

Optional ID of the user who authorized the token. May be an int or a str so it can match the consumer's user primary key (see :data:~mcp_authflow.storage.base.UserId)

 None
Source code in mcp_authflow/storage/base.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@abstractmethod
async def store_refresh_token(
    self,
    refresh_token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None:
    """Store a refresh token.

    Args:
        refresh_token: The refresh token string
        client_id: OAuth client ID
        scopes: List of granted scopes
        expires_at: Unix timestamp when token expires
        resource: Optional RFC 8707 resource binding
        user_id: Optional ID of the user who authorized the token. May be an
            int or a str so it can match the consumer's user primary key
            (see :data:`~mcp_authflow.storage.base.UserId`)
    """
    ...

load_refresh_token abstractmethod async 

load_refresh_token(
    refresh_token: str,
) -> dict[str, Any] | None

Load a refresh token.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string to look up

 required

Returns:

Type Description
dict[str, Any] | None

Token data dict if found and not expired, None otherwise
Source code in mcp_authflow/storage/base.py 
136
137
138
139
140
141
142
143
144
145
146
@abstractmethod
async def load_refresh_token(self, refresh_token: str) -> dict[str, Any] | None:
    """Load a refresh token.

    Args:
        refresh_token: The refresh token string to look up

    Returns:
        Token data dict if found and not expired, None otherwise
    """
    ...

delete_refresh_token abstractmethod async 

delete_refresh_token(refresh_token: str) -> None

Delete a refresh token.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string to delete

 required
Source code in mcp_authflow/storage/base.py 
148
149
150
151
152
153
154
155
@abstractmethod
async def delete_refresh_token(self, refresh_token: str) -> None:
    """Delete a refresh token.

    Args:
        refresh_token: The refresh token string to delete
    """
    ...

revoke_client_tokens abstractmethod async 

revoke_client_tokens(client_id: str) -> int

Revoke every access and refresh token issued to a client.

Implementations should remove both token types atomically when their storage backend supports transactions.

Parameters:

Name Type Description Default
client_id str

OAuth client ID whose tokens should be revoked

 required

Returns:

Type Description
int

Total number of access and refresh tokens removed
Source code in mcp_authflow/storage/base.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@abstractmethod
async def revoke_client_tokens(self, client_id: str) -> int:
    """Revoke every access and refresh token issued to a client.

    Implementations should remove both token types atomically when their
    storage backend supports transactions.

    Args:
        client_id: OAuth client ID whose tokens should be revoked

    Returns:
        Total number of access and refresh tokens removed
    """
    ...

cleanup_expired_tokens abstractmethod async 

cleanup_expired_tokens() -> int

Remove all expired access tokens.

Returns:

Type Description
int

Number of tokens removed
Source code in mcp_authflow/storage/base.py 
172
173
174
175
176
177
178
179
@abstractmethod
async def cleanup_expired_tokens(self) -> int:
    """Remove all expired access tokens.

    Returns:
        Number of tokens removed
    """
    ...

cleanup_expired_refresh_tokens abstractmethod async 

cleanup_expired_refresh_tokens() -> int

Remove all expired refresh tokens.

Returns:

Type Description
int

Number of tokens removed
Source code in mcp_authflow/storage/base.py 
181
182
183
184
185
186
187
188
@abstractmethod
async def cleanup_expired_refresh_tokens(self) -> int:
    """Remove all expired refresh tokens.

    Returns:
        Number of tokens removed
    """
    ...

get_token_count abstractmethod async 

get_token_count() -> int

Get the total number of access tokens in storage.

Returns:

Type Description
int

Number of tokens stored
Source code in mcp_authflow/storage/base.py 
190
191
192
193
194
195
196
197
@abstractmethod
async def get_token_count(self) -> int:
    """Get the total number of access tokens in storage.

    Returns:
        Number of tokens stored
    """
    ...

get_refresh_token_count abstractmethod async 

get_refresh_token_count() -> int

Get the total number of refresh tokens in storage.

Returns:

Type Description
int

Number of refresh tokens stored
Source code in mcp_authflow/storage/base.py 
199
200
201
202
203
204
205
206
@abstractmethod
async def get_refresh_token_count(self) -> int:
    """Get the total number of refresh tokens in storage.

    Returns:
        Number of refresh tokens stored
    """
    ...

MemoryTokenStorage

MemoryTokenStorage 

MemoryTokenStorage()

Bases: TokenStorage

In-memory token storage for testing and development.

This implementation stores tokens in memory and does not persist them across restarts. Suitable for testing and development only.

Initialize in-memory token storage.

Source code in mcp_authflow/storage/memory.py 
20
21
22
23
24
def __init__(self) -> None:
    """Initialize in-memory token storage."""
    self._access_tokens: dict[str, dict[str, Any]] = {}
    self._refresh_tokens: dict[str, dict[str, Any]] = {}
    self._initialized = False

initialize async 

initialize() -> None

Initialize the storage (no-op for memory storage).

Source code in mcp_authflow/storage/memory.py 
26
27
28
29
async def initialize(self) -> None:
    """Initialize the storage (no-op for memory storage)."""
    logger.info("Initializing in-memory token storage")
    self._initialized = True

close async 

close() -> None

Close the storage and clear all tokens.

Source code in mcp_authflow/storage/memory.py 
31
32
33
34
35
36
async def close(self) -> None:
    """Close the storage and clear all tokens."""
    logger.info("Closing in-memory token storage")
    self._access_tokens.clear()
    self._refresh_tokens.clear()
    self._initialized = False

store_token async 

store_token(
    token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None

Store an access token in memory.

Parameters:

Name Type Description Default
token str

The access token string

 required
client_id str

OAuth client ID

 required
scopes list[str]

List of granted scopes

 required
expires_at int

Unix timestamp when token expires

 required
resource str | None

Optional RFC 8707 resource binding

 None
user_id UserId | None

Optional ID of the user who authorized the token. May be an int or a str so it can match the consumer's user primary key (see :data:~mcp_authflow.storage.base.UserId)

 None
Source code in mcp_authflow/storage/memory.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
async def store_token(
    self,
    token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None:
    """Store an access token in memory.

    Args:
        token: The access token string
        client_id: OAuth client ID
        scopes: List of granted scopes
        expires_at: Unix timestamp when token expires
        resource: Optional RFC 8707 resource binding
        user_id: Optional ID of the user who authorized the token. May be an
            int or a str so it can match the consumer's user primary key
            (see :data:`~mcp_authflow.storage.base.UserId`)
    """
    self._store_to(
        self._access_tokens, token, client_id, scopes, expires_at, resource, user_id, "token"
    )

load_token async 

load_token(token: str) -> dict[str, Any] | None

Load an access token from memory.

Parameters:

Name Type Description Default
token str

The access token string to look up

 required

Returns:

Type Description
dict[str, Any] | None

Token data dict if found and not expired, None otherwise
Source code in mcp_authflow/storage/memory.py 
144
145
146
147
148
149
150
151
152
153
async def load_token(self, token: str) -> dict[str, Any] | None:
    """Load an access token from memory.

    Args:
        token: The access token string to look up

    Returns:
        Token data dict if found and not expired, None otherwise
    """
    return await self._load_from(self._access_tokens, token, "token")

delete_token async 

delete_token(token: str) -> None

Delete a token from memory.

Parameters:

Name Type Description Default
token str

The access token string to delete

 required
Source code in mcp_authflow/storage/memory.py 
155
156
157
158
159
160
161
async def delete_token(self, token: str) -> None:
    """Delete a token from memory.

    Args:
        token: The access token string to delete
    """
    self._delete_from(self._access_tokens, token, "token")

cleanup_expired_tokens async 

cleanup_expired_tokens() -> int

Remove all expired access tokens from memory.

Returns:

Type Description
int

Number of tokens removed
Source code in mcp_authflow/storage/memory.py 
163
164
165
166
167
168
169
async def cleanup_expired_tokens(self) -> int:
    """Remove all expired access tokens from memory.

    Returns:
        Number of tokens removed
    """
    return self._cleanup_from(self._access_tokens, "token")

get_token_count async 

get_token_count() -> int

Get the total number of access tokens in storage.

Returns:

Type Description
int

Number of tokens stored
Source code in mcp_authflow/storage/memory.py 
171
172
173
174
175
176
177
178
async def get_token_count(self) -> int:
    """Get the total number of access tokens in storage.

    Returns:
        Number of tokens stored
    """
    self._require_initialized()
    return len(self._access_tokens)

store_refresh_token async 

store_refresh_token(
    refresh_token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None

Store a refresh token in memory.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string

 required
client_id str

OAuth client ID

 required
scopes list[str]

List of granted scopes

 required
expires_at int

Unix timestamp when token expires

 required
resource str | None

Optional RFC 8707 resource binding

 None
user_id UserId | None

Optional ID of the user who authorized the token. May be an int or a str so it can match the consumer's user primary key (see :data:~mcp_authflow.storage.base.UserId)

 None
Source code in mcp_authflow/storage/memory.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
async def store_refresh_token(
    self,
    refresh_token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None:
    """Store a refresh token in memory.

    Args:
        refresh_token: The refresh token string
        client_id: OAuth client ID
        scopes: List of granted scopes
        expires_at: Unix timestamp when token expires
        resource: Optional RFC 8707 resource binding
        user_id: Optional ID of the user who authorized the token. May be an
            int or a str so it can match the consumer's user primary key
            (see :data:`~mcp_authflow.storage.base.UserId`)
    """
    self._store_to(
        self._refresh_tokens,
        refresh_token,
        client_id,
        scopes,
        expires_at,
        resource,
        user_id,
        "refresh token",
    )

load_refresh_token async 

load_refresh_token(
    refresh_token: str,
) -> dict[str, Any] | None

Load a refresh token from memory.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string to look up

 required

Returns:

Type Description
dict[str, Any] | None

Token data dict if found and not expired, None otherwise
Source code in mcp_authflow/storage/memory.py 
212
213
214
215
216
217
218
219
220
221
async def load_refresh_token(self, refresh_token: str) -> dict[str, Any] | None:
    """Load a refresh token from memory.

    Args:
        refresh_token: The refresh token string to look up

    Returns:
        Token data dict if found and not expired, None otherwise
    """
    return await self._load_from(self._refresh_tokens, refresh_token, "refresh token")

delete_refresh_token async 

delete_refresh_token(refresh_token: str) -> None

Delete a refresh token from memory.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string to delete

 required
Source code in mcp_authflow/storage/memory.py 
223
224
225
226
227
228
229
async def delete_refresh_token(self, refresh_token: str) -> None:
    """Delete a refresh token from memory.

    Args:
        refresh_token: The refresh token string to delete
    """
    self._delete_from(self._refresh_tokens, refresh_token, "refresh token")

revoke_client_tokens async 

revoke_client_tokens(client_id: str) -> int

Revoke every access and refresh token issued to a client.

Source code in mcp_authflow/storage/memory.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
async def revoke_client_tokens(self, client_id: str) -> int:
    """Revoke every access and refresh token issued to a client."""
    self._require_initialized()
    access_keys = [
        key
        for key, token_data in self._access_tokens.items()
        if token_data["client_id"] == client_id
    ]
    refresh_keys = [
        key
        for key, token_data in self._refresh_tokens.items()
        if token_data["client_id"] == client_id
    ]

    for key in access_keys:
        del self._access_tokens[key]
    for key in refresh_keys:
        del self._refresh_tokens[key]

    count = len(access_keys) + len(refresh_keys)
    if count > 0:
        logger.info("Revoked %s token(s) for client %s", count, client_id)
    return count

cleanup_expired_refresh_tokens async 

cleanup_expired_refresh_tokens() -> int

Remove all expired refresh tokens from memory.

Returns:

Type Description
int

Number of tokens removed
Source code in mcp_authflow/storage/memory.py 
255
256
257
258
259
260
261
async def cleanup_expired_refresh_tokens(self) -> int:
    """Remove all expired refresh tokens from memory.

    Returns:
        Number of tokens removed
    """
    return self._cleanup_from(self._refresh_tokens, "refresh token")

get_refresh_token_count async 

get_refresh_token_count() -> int

Get the total number of refresh tokens in storage.

Returns:

Type Description
int

Number of refresh tokens stored
Source code in mcp_authflow/storage/memory.py 
263
264
265
266
267
268
269
270
async def get_refresh_token_count(self) -> int:
    """Get the total number of refresh tokens in storage.

    Returns:
        Number of refresh tokens stored
    """
    self._require_initialized()
    return len(self._refresh_tokens)

PostgresTokenStorage

PostgresTokenStorage 

PostgresTokenStorage(database_url: str | None = None)

Bases: TokenStorage

Database-backed storage for MCP access tokens using PostgreSQL.

Initialize token storage.

Parameters:

Name Type Description Default
database_url str | None

PostgreSQL connection URL. If not provided, will be read from DATABASE_URL environment variable.

 None
Source code in mcp_authflow/storage/postgres.py 
45
46
47
48
49
50
51
52
53
def __init__(self, database_url: str | None = None):
    """Initialize token storage.

    Args:
        database_url: PostgreSQL connection URL. If not provided,
                     will be read from DATABASE_URL environment variable.
    """
    self.database_url = database_url or os.environ.get("DATABASE_URL")
    self._pool: asyncpg.Pool | None = None

initialize async 

initialize() -> None

Initialize the database connection pool.

Source code in mcp_authflow/storage/postgres.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def initialize(self) -> None:
    """Initialize the database connection pool."""
    if not self.database_url:
        raise StorageConfigError(
            "DATABASE_URL environment variable is required for token storage"
        )

    logger.info("Initializing database connection pool for token storage")
    self._pool = await asyncpg.create_pool(
        self.database_url,
        min_size=2,
        max_size=10,
        command_timeout=30,
    )
    logger.info("Database connection pool initialized")
    await self._verify_schema()

close async 

close() -> None

Close the database connection pool.

Source code in mcp_authflow/storage/postgres.py 
122
123
124
125
126
127
async def close(self) -> None:
    """Close the database connection pool."""
    if self._pool:
        await self._pool.close()
        self._pool = None
        logger.info("Database connection pool closed")

store_token async 

store_token(
    token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None

Store an access token in the database.

Parameters:

Name Type Description Default
token str

The access token string

 required
client_id str

OAuth client ID

 required
scopes list[str]

List of granted scopes

 required
expires_at int

Unix timestamp when token expires

 required
resource str | None

Optional RFC 8707 resource binding

 None
user_id UserId | None

Optional ID of the user who authorized the token. May be an int or a str so it can match the consumer's user primary key (see :data:~mcp_authflow.storage.base.UserId)

 None
Source code in mcp_authflow/storage/postgres.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
async def store_token(
    self,
    token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None:
    """Store an access token in the database.

    Args:
        token: The access token string
        client_id: OAuth client ID
        scopes: List of granted scopes
        expires_at: Unix timestamp when token expires
        resource: Optional RFC 8707 resource binding
        user_id: Optional ID of the user who authorized the token. May be an
            int or a str so it can match the consumer's user primary key
            (see :data:`~mcp_authflow.storage.base.UserId`)
    """
    await self._store_to(
        "mcp_access_tokens",
        token,
        client_id,
        scopes,
        expires_at,
        resource,
        user_id,
        "token",
    )

load_token async 

load_token(token: str) -> dict[str, Any] | None

Load an access token from the database.

Parameters:

Name Type Description Default
token str

The access token string to look up

 required

Returns:

Type Description
dict[str, Any] | None

Token data dict if found and not expired, None otherwise
Source code in mcp_authflow/storage/postgres.py 
276
277
278
279
280
281
282
283
284
285
async def load_token(self, token: str) -> dict[str, Any] | None:
    """Load an access token from the database.

    Args:
        token: The access token string to look up

    Returns:
        Token data dict if found and not expired, None otherwise
    """
    return await self._load_from("mcp_access_tokens", token, "token")

delete_token async 

delete_token(token: str) -> None

Delete a token from the database.

Parameters:

Name Type Description Default
token str

The access token string to delete

 required
Source code in mcp_authflow/storage/postgres.py 
287
288
289
290
291
292
293
async def delete_token(self, token: str) -> None:
    """Delete a token from the database.

    Args:
        token: The access token string to delete
    """
    await self._delete_from("mcp_access_tokens", token, "token")

cleanup_expired_tokens async 

cleanup_expired_tokens() -> int

Remove all expired tokens from the database.

Returns:

Type Description
int

Number of tokens removed
Source code in mcp_authflow/storage/postgres.py 
295
296
297
298
299
300
301
async def cleanup_expired_tokens(self) -> int:
    """Remove all expired tokens from the database.

    Returns:
        Number of tokens removed
    """
    return await self._cleanup_from("mcp_access_tokens", "token")

get_token_count async 

get_token_count() -> int

Get the total number of tokens in storage.

Returns:

Type Description
int

Number of tokens stored
Source code in mcp_authflow/storage/postgres.py 
303
304
305
306
307
308
309
310
311
312
313
async def get_token_count(self) -> int:
    """Get the total number of tokens in storage.

    Returns:
        Number of tokens stored
    """
    pool = self._require_pool()

    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT COUNT(*) as count FROM mcp_access_tokens")
    return row["count"] if row else 0

store_refresh_token async 

store_refresh_token(
    refresh_token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None

Store a refresh token in the database.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string

 required
client_id str

OAuth client ID

 required
scopes list[str]

List of granted scopes

 required
expires_at int

Unix timestamp when token expires

 required
resource str | None

Optional RFC 8707 resource binding

 None
user_id UserId | None

Optional ID of the user who authorized the token. May be an int or a str so it can match the consumer's user primary key (see :data:~mcp_authflow.storage.base.UserId)

 None
Source code in mcp_authflow/storage/postgres.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
async def store_refresh_token(
    self,
    refresh_token: str,
    client_id: str,
    scopes: list[str],
    expires_at: int,
    resource: str | None = None,
    user_id: UserId | None = None,
) -> None:
    """Store a refresh token in the database.

    Args:
        refresh_token: The refresh token string
        client_id: OAuth client ID
        scopes: List of granted scopes
        expires_at: Unix timestamp when token expires
        resource: Optional RFC 8707 resource binding
        user_id: Optional ID of the user who authorized the token. May be an
            int or a str so it can match the consumer's user primary key
            (see :data:`~mcp_authflow.storage.base.UserId`)
    """
    await self._store_to(
        "mcp_refresh_tokens",
        refresh_token,
        client_id,
        scopes,
        expires_at,
        resource,
        user_id,
        "refresh token",
    )

load_refresh_token async 

load_refresh_token(
    refresh_token: str,
) -> dict[str, Any] | None

Load a refresh token from the database.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string to look up

 required

Returns:

Type Description
dict[str, Any] | None

Token data dict if found and not expired, None otherwise
Source code in mcp_authflow/storage/postgres.py 
347
348
349
350
351
352
353
354
355
356
async def load_refresh_token(self, refresh_token: str) -> dict[str, Any] | None:
    """Load a refresh token from the database.

    Args:
        refresh_token: The refresh token string to look up

    Returns:
        Token data dict if found and not expired, None otherwise
    """
    return await self._load_from("mcp_refresh_tokens", refresh_token, "refresh token")

delete_refresh_token async 

delete_refresh_token(refresh_token: str) -> None

Delete a refresh token from the database.

Parameters:

Name Type Description Default
refresh_token str

The refresh token string to delete

 required
Source code in mcp_authflow/storage/postgres.py 
358
359
360
361
362
363
364
async def delete_refresh_token(self, refresh_token: str) -> None:
    """Delete a refresh token from the database.

    Args:
        refresh_token: The refresh token string to delete
    """
    await self._delete_from("mcp_refresh_tokens", refresh_token, "refresh token")

revoke_client_tokens async 

revoke_client_tokens(client_id: str) -> int

Atomically revoke every access and refresh token issued to a client.

Source code in mcp_authflow/storage/postgres.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
async def revoke_client_tokens(self, client_id: str) -> int:
    """Atomically revoke every access and refresh token issued to a client."""
    pool = self._require_pool()
    async with pool.acquire() as conn:
        has_refresh_table = await conn.fetchval(
            "SELECT to_regclass('mcp_refresh_tokens') IS NOT NULL"
        )
        if has_refresh_table:
            row = await conn.fetchrow(
                """
                WITH deleted_access AS (
                    DELETE FROM mcp_access_tokens
                    WHERE client_id = $1
                    RETURNING 1
                ), deleted_refresh AS (
                    DELETE FROM mcp_refresh_tokens
                    WHERE client_id = $1
                    RETURNING 1
                )
                SELECT
                    (SELECT COUNT(*) FROM deleted_access)
                    + (SELECT COUNT(*) FROM deleted_refresh) AS count
                """,
                client_id,
            )
            count = row["count"] if row else 0
        else:
            result = await conn.execute(
                "DELETE FROM mcp_access_tokens WHERE client_id = $1", client_id
            )
            count = int(result.split()[-1]) if result else 0

    if count > 0:
        logger.info("Revoked %s token(s) for client %s", count, client_id)
    return count

cleanup_expired_refresh_tokens async 

cleanup_expired_refresh_tokens() -> int

Remove all expired refresh tokens from the database.

Returns:

Type Description
int

Number of tokens removed
Source code in mcp_authflow/storage/postgres.py 
402
403
404
405
406
407
408
async def cleanup_expired_refresh_tokens(self) -> int:
    """Remove all expired refresh tokens from the database.

    Returns:
        Number of tokens removed
    """
    return await self._cleanup_from("mcp_refresh_tokens", "refresh token")

get_refresh_token_count async 

get_refresh_token_count() -> int

Get the total number of refresh tokens in storage.

Returns:

Type Description
int

Number of refresh tokens stored
Source code in mcp_authflow/storage/postgres.py 
410
411
412
413
414
415
416
417
418
419
420
async def get_refresh_token_count(self) -> int:
    """Get the total number of refresh tokens in storage.

    Returns:
        Number of refresh tokens stored
    """
    pool = self._require_pool()

    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT COUNT(*) as count FROM mcp_refresh_tokens")
    return row["count"] if row else 0

UserId

UserId module-attribute 

UserId: TypeAlias = int | str

Identifier of the user who authorized a token.

Deliberately not narrowed to int: consuming applications key users on anything from a SERIAL/BIGINT counter to a UUID or an external subject string. Storage backends pass the value straight through, so the type you hand in has to match the user_id column type in your DDL (see the README "Choosing a user_id column type" section for the BIGINT/UUID/TEXT variants). Pass UUIDs in their string form.