René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"How can I access the bearer token inside my MCP tool?","articleBody":"### Question\n\nHi everyone,\n\nI created a multi-client RAG system to organize client documents.\n\nCurrently, authentication works correctly using the keys stored in the .env file.\n\nHowever, 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.\n\nI tried passing ctx: Context or AccessToken, but it always appears empty.\n\nHere is my code:\n\n```python\nimport argparse\nfrom typing import Dict, Any\nfrom pydantic import AnyHttpUrl # type:ignore\nfrom mcp.server.auth.settings import AuthSettings # type:ignore\nfrom mcp.server.auth.provider import AccessToken, TokenVerifier # type:ignore\nfrom mcp.server.fastmcp import Context, FastMCP # type:ignore\nimport sys\nimport time\nimport json\n\nfrom config.logger import logger\nfrom src.client_manager import MultiClientManager\n\n\n# --- Inizializza il manager multi-cliente ---\ntry:\n    client_manager = MultiClientManager()\n    logger.info(\"🏢 Multi-Client Manager inizializzato\")\nexcept Exception as e:\n    logger.error(f\"❌ Errore inizializzazione manager: {e}\")\n    sys.exit(1)\n\nclass EnvironmentMultiClientTokenVerifier(TokenVerifier):\n    \"Token verifier che mappa token univoci ai client_id.\"\n\n    def __init__(self):\n        # Carica i token da variabili d'ambiente\n        # Formato: MCP_TOKEN_\u003cCLIENT_ID\u003e=token_value\n        self.token_to_client = {}\n        \n        # Scansiona tutte le variabili d'ambiente che iniziano con MCP_TOKEN_\n        for key, value in os.environ.items():\n            if key.startswith(\"MCP_TOKEN_\"):\n                client_id = key[10:].upper()  \n                self.token_to_client[value] = client_id\n        \n        if not self.token_to_client:\n            raise ValueError(\"Nessun token trovato! Definisci variabili MCP_TOKEN_\u003cCLIENT_ID\u003e\")\n\n    async def verify_token(self, token: str) -\u003e AccessToken | None:\n        \"Verifica il token e restituisce il client_id associato.\"\n        client_id = self.token_to_client.get(token)\n\n        if client_id:\n            access_token = AccessToken(\n                token=token,\n                client_id=client_id,\n                scopes=[\"user\"]\n            )\n            return access_token\n        \n        return None\n\n\n# Initialize FastMCP server\nmcp = FastMCP(\n    \"rag-server-multiclient-auth\",\n    # Token verifier for authentication\n    token_verifier=EnvironmentMultiClientTokenVerifier(),\n    # Auth settings for RFC 9728 Protected Resource Metadata\n    auth=AuthSettings(\n        issuer_url=AnyHttpUrl(\"https://auth.example.com\"),  # Authorization Server URL\n        resource_server_url=AnyHttpUrl(\"http://localhost:8000\"),  # This server's URL\n        required_scopes=[\"user\"],\n    ),\n)\n\n# --- MCP Tools ---\n@mcp.tool()\ndef search_documents(query: str, limit: int) -\u003e Dict[str, Any]:\n    \"Search documents for a specific client using RAG pipeline\n    \n    Args:\n        query: The search query\n        limit: Maximum number of results to return\n    \"\n\n\t# Get the client id from Beraer token or args params pass\n\tclient_id = 'Client1'\n\n    logger.info(f\"🔍 Searching for client '{client_id}': '{query}' (limit: {limit})\")\n    start_time = time.time()\n    \n    try:\n        results = client_manager.search_client_documents(\n            client_id=client_id,\n            query=query, \n            top_k=limit\n        )\n        \n        search_time = time.time() - start_time\n        \n        return {\n            \"client_id\": client_id,\n            \"query\": query,\n            \"results\": results,\n            \"total_found\": len(results),\n            \"execution_time\": f\"{search_time:.2f}s\",\n            \"status\": \"success\"\n        }\n        \n    except Exception as e:\n        error_msg = f\"Error during search for client {client_id}: {str(e)}\"\n        logger.error(error_msg)\n        search_time = time.time() - start_time\n        \n        return {\n            \"client_id\": client_id,\n            \"query\": query,\n            \"results\": [],\n            \"total_found\": 0,\n            \"execution_time\": f\"{search_time:.2f}s\",\n            \"status\": \"error\",\n            \"error\": error_msg\n        }\n    \n\nif __name__ == \"__main__\":\n    mcp.run(transport='sse')\n```\n\n**Question:**\nIs there a way to access the bearer token within a MCP tool function like `search_documents`?\n\nThanks in advance for any help!\n\n\n### Additional Context\n\npython=3.12.3\nmcp==1.14.1\n","author":{"url":"https://github.com/abernardini-inc","@type":"Person","name":"abernardini-inc"},"datePublished":"2025-09-30T11:00:38.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":2},"url":"https://github.com/1414/python-sdk/issues/1414"}