How can I access the bearer token inside my MCP tool?

[abernardini-inc]

Question

Hi everyone,

I created a multi-client RAG system to organize client documents.

Currently, authentication works correctly using the keys stored in the .env file.

However, I'm trying to access the bearer token inside my search_documents tool, but I haven't been successful. I need the bearer token to retrieve the name of the associated client, as each client has their own personal bearer token.

I tried passing ctx: Context or AccessToken, but it always appears empty.

Here is my code:

import argparse
from typing import Dict, Any
from pydantic import AnyHttpUrl # type:ignore
from mcp.server.auth.settings import AuthSettings # type:ignore
from mcp.server.auth.provider import AccessToken, TokenVerifier # type:ignore
from mcp.server.fastmcp import Context, FastMCP # type:ignore
import sys
import time
import json

from config.logger import logger
from src.client_manager import MultiClientManager


# --- Inizializza il manager multi-cliente ---
try:
    client_manager = MultiClientManager()
    logger.info("🏢 Multi-Client Manager inizializzato")
except Exception as e:
    logger.error(f"❌ Errore inizializzazione manager: {e}")
    sys.exit(1)

class EnvironmentMultiClientTokenVerifier(TokenVerifier):
    "Token verifier che mappa token univoci ai client_id."

    def __init__(self):
        # Carica i token da variabili d'ambiente
        # Formato: MCP_TOKEN_<CLIENT_ID>=token_value
        self.token_to_client = {}
        
        # Scansiona tutte le variabili d'ambiente che iniziano con MCP_TOKEN_
        for key, value in os.environ.items():
            if key.startswith("MCP_TOKEN_"):
                client_id = key[10:].upper()  
                self.token_to_client[value] = client_id
        
        if not self.token_to_client:
            raise ValueError("Nessun token trovato! Definisci variabili MCP_TOKEN_<CLIENT_ID>")

    async def verify_token(self, token: str) -> AccessToken | None:
        "Verifica il token e restituisce il client_id associato."
        client_id = self.token_to_client.get(token)

        if client_id:
            access_token = AccessToken(
                token=token,
                client_id=client_id,
                scopes=["user"]
            )
            return access_token
        
        return None


# Initialize FastMCP server
mcp = FastMCP(
    "rag-server-multiclient-auth",
    # Token verifier for authentication
    token_verifier=EnvironmentMultiClientTokenVerifier(),
    # Auth settings for RFC 9728 Protected Resource Metadata
    auth=AuthSettings(
        issuer_url=AnyHttpUrl("https://auth.example.com"),  # Authorization Server URL
        resource_server_url=AnyHttpUrl("http://localhost:8000"),  # This server's URL
        required_scopes=["user"],
    ),
)

# --- MCP Tools ---
@mcp.tool()
def search_documents(query: str, limit: int) -> Dict[str, Any]:
    "Search documents for a specific client using RAG pipeline
    
    Args:
        query: The search query
        limit: Maximum number of results to return
    "

	# Get the client id from Beraer token or args params pass
	client_id = 'Client1'

    logger.info(f"🔍 Searching for client '{client_id}': '{query}' (limit: {limit})")
    start_time = time.time()
    
    try:
        results = client_manager.search_client_documents(
            client_id=client_id,
            query=query, 
            top_k=limit
        )
        
        search_time = time.time() - start_time
        
        return {
            "client_id": client_id,
            "query": query,
            "results": results,
            "total_found": len(results),
            "execution_time": f"{search_time:.2f}s",
            "status": "success"
        }
        
    except Exception as e:
        error_msg = f"Error during search for client {client_id}: {str(e)}"
        logger.error(error_msg)
        search_time = time.time() - start_time
        
        return {
            "client_id": client_id,
            "query": query,
            "results": [],
            "total_found": 0,
            "execution_time": f"{search_time:.2f}s",
            "status": "error",
            "error": error_msg
        }
    

if __name__ == "__main__":
    mcp.run(transport='sse')

Question:
Is there a way to access the bearer token within a MCP tool function like search_documents?

Thanks in advance for any help!

Additional Context

python=3.12.3
mcp==1.14.1

[shulkx]

Try this: #816 (comment)

[felixweinberger]

Try this: #816 (comment)

Thank you @shulkx! @abernardini-inc feel free to reopen if this doesn't resolve your question.