Skip to content

auth_context

get_access_token

get_access_token() -> AccessToken | None

Get the access token from the current context.

Returns:

Type Description
AccessToken | None

The access token if an authenticated user is available, None otherwise.

Source code in src/mcp/server/auth/middleware/auth_context.py
13
14
15
16
17
18
19
20
def get_access_token() -> AccessToken | None:
    """Get the access token from the current context.

    Returns:
        The access token if an authenticated user is available, None otherwise.
    """
    auth_user = auth_context_var.get()
    return auth_user.access_token if auth_user else None

AuthContextMiddleware

Middleware that extracts the authenticated user from the request and sets it in a contextvar for easy access throughout the request lifecycle.

This middleware should be added after the AuthenticationMiddleware in the middleware stack to ensure that the user is properly authenticated before being stored in the context.

Source code in src/mcp/server/auth/middleware/auth_context.py
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class AuthContextMiddleware:
    """Middleware that extracts the authenticated user from the request
    and sets it in a contextvar for easy access throughout the request lifecycle.

    This middleware should be added after the AuthenticationMiddleware in the
    middleware stack to ensure that the user is properly authenticated before
    being stored in the context.
    """

    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        user = scope.get("user")
        if isinstance(user, AuthenticatedUser):
            # Set the authenticated user in the contextvar
            token = auth_context_var.set(user)
            try:
                await self.app(scope, receive, send)
            finally:
                auth_context_var.reset(token)
        else:
            # No authenticated user, just process the request
            await self.app(scope, receive, send)