Custom TokenVerifier

[jaimeescano]

Hi there,

Not sure if someone has faced this challenge. Currently I'm trying to implement a logic to authenticate users from different "token" sources. So to identify the source of the token, the user must provide a request header "X-API-Auth-Provider" ... which then I'm trying to capture and then implement different logic based on the value.

I have tried using the following code .. but it looks like the

from fastmcp.server.dependencies import get_http_request
class CustomTokenVerifier(TokenVerifier):
    async def verify_token(self, token: str) -> AccessToken | None:    
           headers: dict = await get_http_headers()
           print (headers)

But headers are empty ... not sure if it's a bug or is there a different way to access the current "HTTP connection" from a token verifier?

[kskarthik]

as far i understand, you have to use the Middleware for any custom validations. Although, I am facing a different challenge there #1362

[kskarthik]

@jaimeescano get_http_headers() does not work independently. See https://gofastmcp.com/patterns/http-requests#important-notes

They work only within tool, prompt, resources functions, also in Middleware

[jaimeescano]

Hi @kskarthik thanks for your answer.

Yeah, I checked about the "only works with tool prompts and resources" ... the problem on my side is that the "authentification" happends before any middleware is executed .. so I can't add a logic to "extract" any header and put it in the state ... so I can use it in my auth logic.

I just noticed that two days ago .. server/auth has changed quite a bit ... so will take a look. Just noticed they have a folder with "providers", so I might be able to implement my own provider and might have access to the "conn" variable (which tracing other oauth providers they do get).

Appreciated you feedback BTW.

[AshokChava]

Were you able to do this? Can you please share your findings?

[Norcim133]

@jaimeescano and @kskarthik
If I understand your need, you can do this.

You create your own wrapper and decorate your tools with it (my wrapper below... it's mid-refactor).

Then, in the wrapper, you can use get_http_headers (or get_access_token) and handle what you need.

Basically, you have logic for each "auth provider" and you attach it to ctx.

Voila, all tools that are wrapped can access that via ctx.

The beauty is the wrapper operates per request... so it will have the unique header and the unique ctx for just that particular auth-provider's request.

Wrapper

# helpers/auth_wrapper.py
import functools, os, re
from graph.controller import GraphController
from graph.client import get_user_client
from fastmcp.server.dependencies import get_http_headers
from starlette.requests import Request
from starlette.responses import JSONResponse

PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", "http://127.0.0.1:8080")

def with_graph(func):
    """
    Tool decorator:
      - injects GraphController into ctx.graph
      - returns sign-in instructions if no auth
      - returns a short message for missing/invalid tokens
    """
    @functools.wraps(func)
    async def wrapper(ctx, *args, **kwargs):

        # Get http request from dependencies
        headers = get_http_headers()
        access_token = headers.get("X-Microsoft-Token", None)

        # #Extract header to get access token

        #If Now access token, return 401
        if not access_token:
            return JSONResponse(
                {"error": "authentication_required"},
               
                status_code=401,
                headers={
                    "WWW-Authenticate": 'Bearer realm="https://voice-email-outlook.fly.dev/.well-known/oauth-protected-resource"'
                }
            )

        # Construct client with access token
        client = "TBD"

        # Attach ready-to-use GraphController for this request
        ctx.graph = GraphController(client)

        return await func(ctx, *args, **kwargs)

    return wrapper

Usage

    @mcp.tool(output_schema=None)
    @with_graph
    async def list_inbox_messages(ctx: Context, count: int = 50) -> str:
        """Key header details for inbox messages."""
        page = await ctx.graph.mail.get_inbox(count=count)
        return format_inbox_headers(page)

[Norcim133]

The alternative is you use FastMCP's StaticTokenVerifier.

Now, I understand it is stored in plaintext so it depends on your setup.

But you can set different tokens in the server side (make them long API keys) for each auth-provider.

Then, access is similar to the above... but them in a wrapper or in the tools themselves

Token Setup in Server

from fastmcp import FastMCP
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier

# Define development tokens and their associated claims
verifier = StaticTokenVerifier(
    tokens={
        "auth-provider-a": {
            "auth-provider-specific-info": "Blue",
            "scopes": ["read:data", "write:data", "admin:users"]
        },
        "auth-provider-b": {
            "auth-provider-specific-info": "Green",
            "scopes": ["read:data"]
        }
    },
    required_scopes=["read:data"]
)

Usage in Tool or Wrapper

auth-proider-specific-info = get_access_token().claims.get("client_id")

[kskarthik]

@Norcim133 My use case is that the user needs to provide an API key & a custom variable for context, via the headers.

So, basically I need to validate 2 headers & throw err if not supplied. API_TOKEN & MY_CUSTOM_HEADER

[Norcim133]

@kskarthik
And Middleware didn't work?

I have this one working in my system and it would give you the same thing with headers.

Mine is adding to context but you want yours to block.

# Connects starlette request info (e.g. verified msgraph object) into fastmcp context
class InjectUserContextMiddleware(Middleware):
    async def on_request(self, context: MiddlewareContext, call_next):
        # Add custom logic here
        request = get_http_request()
        
        if hasattr(request, 'state'):
            graph = getattr(request.state, 'graph', None)
            user_id = getattr(request.state, 'user_id', None)
            #ms_token = getattr(request.state, 'ms_token', None)
        context.fastmcp_context.graph=graph
        context.fastmcp_context.set_state("user_id", user_id)
        #context.fastmcp_context.set_state("ms_token", ms_token)
        result = await call_next(context)
        
        return result

I also hav this elsewhere which could go in Middleware too... I run this as a way to debug my middlewares are working.

@mcp.tool
async def testing_only_echo_auth_info(ctx: Context) -> dict:
    """Test tool to verify auth info is available"""
    headers = get_http_headers()
    request = get_http_request()
    user_id = ctx.get_state('user_id')

    return {
        "all_headers": dict(headers),
        "special_token_present": "X-Special-Token".lower() in headers,
        "auth_header": headers.get("Authorization", "none"),
        "timestamp": time.time(),
        "user_id": user_id
    }