Question: How to authorise a client with Bearer header with SSE?

[Gelembjuk]

I am building the MCP server application to connect some services to LLM .
One of things i want to implement is authorisation of a user with the token.

I see it must be possible somehow. Because MCP inspector has the Authentification and Bearer Token field

Most of tutorials about MCP are related to STDIO kind of a server run. My will be SSE.

There is my code:

from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI, Request, Depends, HTTPException

app = FastAPI()
mcp = FastMCP("SMB Share Server")

@mcp.tool()
def create_folder(parent_path: str, name: str) -> str:
    """Create new subfolder in the specified path"""
    return f"Folder {name} created in {parent_path}"

app.mount("/", mcp.sse_app())

How can i read Authorization header in case if it is sent by the client?

I tried to use approaches of FastAPI - setting dependency, adding request:Request to arguments but this doesn't work.

Is there a way?

[scott-clare1]

I was also hoping to do something like this too, it seems like this is possible in the typescript SDK currently. It looks like there's a couple of open PR's for adding OAuth to this project but I've not seen anything on bearer tokens.

[scott-clare1]

Have you tried mounting the server directly to a Starlette app rather than FastAPI? As per the docs: https://github.com/modelcontextprotocol/python-sdk?tab=readme-ov-file#mounting-to-an-existing-asgi-server. You'd then have to write some auth middleware? https://www.starlette.io/authentication/

[lucassampsouza]

I have the same question.
I created an SSE server and put basic authentication middleware and that is OK, but I can't retrieve authentication headers on tool run

To work around this, I created a class to save lastUserSession and I got that on tool run, but I know it's not the best solution.
But solved for testing purposes at the moment

But I'm waiting for someone with a better idea
Here is my sample code

from starlette.applications import Starlette
from starlette.routing import Mount
from mcp.server.fastmcp import FastMCP, Context
from mcp.server import Server
from urllib.parse import urlparse, parse_qs

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from starlette.applications import Starlette
from starlette.authentication import (
    AuthCredentials, AuthenticationBackend, AuthenticationError, SimpleUser
)
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
import base64
import binascii

class LastUserSession:
    def __init__(self):
        self._current_session = None
        self._current_username = None
    
    def set_session(self, session_id, username):
        """Store the username of the last request"""
        self._current_session = session_id
        self._current_username = username
        #print(f"Latest session updated: {session_id} for user {username}")
    
    def get_username(self, session_id=None):
        """Get the username from the last request"""
        # session_id parameter is kept for compatibility but ignored
        return self._current_username
    
    def get_current_session(self):
        """Get the session_id from the last request"""
        return self._current_session

# Instância da classe para armazenar apenas a última sessão
user_sessions = LastUserSession()

class BasicAuthBackend(AuthenticationBackend):
    async def authenticate(self, conn):
        
        # Extrair o session_id da URL
        parsed_url = urlparse(str(conn.url))
        query_params = parse_qs(parsed_url.query)
        session_id = query_params.get('session_id', [None])[0]
        
        if "Authorization" not in conn.headers:
            return

        auth = conn.headers["Authorization"]
        try:
            scheme, credentials = auth.split()
            if scheme.lower() != 'basic':
                return
            decoded = base64.b64decode(credentials).decode("ascii")
        except (ValueError, UnicodeDecodeError, binascii.Error) as exc:
            raise AuthenticationError('Invalid basic auth credentials')

        username, _, password = decoded.partition(":")
        
        # Armazena a relação entre session_id e username usando a classe
        user_sessions.set_session(session_id, username)
        
        return AuthCredentials(["authenticated"]), SimpleUser(username)

@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
    """Manage server startup and shutdown lifecycle."""
    # Initialize resources on startup
    yield {"user": user_sessions.get_username()}

# mcp = FastMCP("Example APP", lifespan=server_lifespan, log_level='DEBUG')
mcp = FastMCP("Example APP", lifespan=server_lifespan)

# Add an addition tool
@mcp.tool()
async def who_kill_odete_roitman(ctx: Context) -> str:
    """Answer the question Who Kill Odete Roitman"""
    
    return f"{ctx.request_context.lifespan_context['user']} killed Odete Roitman"

routes=[
        Mount('/', app=mcp.sse_app()),
    ]

middleware = [
    Middleware(AuthenticationMiddleware, backend=BasicAuthBackend()),
]

# Mount the SSE server to the existing ASGI server
app = Starlette(routes=routes, middleware=middleware)

[samirbajaj]

I have the same question -- it looks like this PR will fix it?

[josx]

mcp.sse_app() is one Route /sse and a Mount app on /messages/.

There is a way to use starlette permission with decoration and override handle_sse func.
Somethin like:

for route in app.routes:
  if route.path == "/sse":
    original_endpoint = route.endpoint
    route.endpoint = requires("authenticated")(original_endpoint)

But I dont know how to add this decorator to the Mounted app.

[josx]

Finally i make it work, here anexample.

import jwt
import datetime
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.authentication import (
    AuthCredentials, AuthenticationBackend, AuthenticationError, SimpleUser
)
from starlette.authentication import requires
from starlette.middleware import Middleware
from server import mcp
from pprint import pprint
from mcp.server.sse import SseServerTransport

JWT_SECRET = "your-secret-key"

async def welcome_message(request):
    return PlainTextResponse("Demo Mcp with jwt auth")


class JWTAuthentication(AuthenticationBackend):
    async def authenticate(self, request):
        auth_header = request.headers.get("Authorization")
        if not auth_header:
            return None
        token = auth_header.split(" ")[1]
        try:
            payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
            return AuthCredentials(["authenticated"]), SimpleUser(payload)
        except jwt.ExpiredSignatureError:
            raise AuthenticationError("Expired signature")
        except jwt.InvalidTokenError:
            raise AuthenticationError("Invalid token")


async def get_token(request):
    username = request.query_params.get("username", "demo_user")
    scope = request.query_params.get("scope", "mcp:access")
    token = jwt.encode({
        "username": username,
        "scope": scope,
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
    }, JWT_SECRET, algorithm="HS256")
    return JSONResponse({"token": token})


# Create SSE transport
sse = SseServerTransport("/messages/")


# MCP SSE handler function
async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as (
        read_stream,
        write_stream,
    ):
        await mcp._mcp_server.run(
            read_stream, write_stream, mcp._mcp_server.create_initialization_options()
        )

async def handle_messages(request):
    await sse.handle_post_message(request.scope, request.receive, request._send)


routes = [
    Route("/", endpoint=welcome_message),
    Route("/get_token", get_token),
    # MCP related routes
    Route("/sse", endpoint=requires("authenticated")(handle_sse)),
    Route("/messages/", endpoint=requires("authenticated")(handle_messages), methods=["POST"]),
]

app = Starlette(
            debug=True,
            routes=routes
        )
app.add_middleware(AuthenticationMiddleware, backend=JWTAuthentication())


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0")

it is working but in every call i got a exception:

ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File ".venv/lib/python3.11/site-packages/uvicorn/protocols/http/h11_impl.py", line 403, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".venv/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".venv/lib/python3.11/site-packages/starlette/applications.py", line 112, in __call__
    await self.middleware_stack(scope, receive, send)
  File ".venv/lib/python3.11/site-packages/starlette/middleware/errors.py", line 187, in __call__
    raise exc
  File ".venv/lib/python3.11/site-packages/starlette/middleware/errors.py", line 165, in __call__
    await self.app(scope, receive, _send)
  File ".venv/lib/python3.11/site-packages/starlette/middleware/authentication.py", line 48, in __call__
    await self.app(scope, receive, send)
  File ".venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 62, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File ".venv/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File ".venv/lib/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File ".venv/lib/python3.11/site-packages/starlette/routing.py", line 714, in __call__
    await self.middleware_stack(scope, receive, send)
  File ".venv/lib/python3.11/site-packages/starlette/routing.py", line 734, in app
    await route.handle(scope, receive, send)
  File ".venv/lib/python3.11/site-packages/starlette/routing.py", line 288, in handle
    await self.app(scope, receive, send)
  File ".venv/lib/python3.11/site-packages/starlette/routing.py", line 76, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File ".venv/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File ".venv/lib/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File ".venv/lib/python3.11/site-packages/starlette/routing.py", line 74, in app
    await response(scope, receive, send)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: 'NoneType' object is not callable

[Gelembjuk]

This is how i did this for now.

from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI, Request

# global variable
auth_token = ""

app = FastAPI()
mcp = FastMCP("Server to manage a Linux instance")

@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    auth_header = request.headers.get("Authorization")
    if auth_header:
        # extract token from the header and keep it in the global variable
        global auth_token
        auth_token = auth_header.split(" ")[1]
    
    response = await call_next(request)
    return response

@mcp.tool()
def cli_command(command: str, work_dir: str = "") -> str:
    """
    ....
    """
    # We require each request to have the auth token, raises exception if wrong
    auth_manager.verify_token(auth_token)
    .... do the job

app.mount("/", mcp.sse_app())

[aadharsh-sndk]

Here is a simple trick that worked for us. We are building a "relay" MCP Server. Meaning, we need to do token forwarding from our MCP Client on to the MCP Server (and down to an API Server)

Sample server.py

from fastmcp import FastMCP, Context

mcp = FastMCP("Demo 🚀")

def get_bearer_token(ctx):
    request = ctx.get_http_request()
    headers = request.headers
    # Check if 'Authorization' header is present
    authorization_header = headers.get('Authorization')
    
    if authorization_header:
        # Split the header into 'Bearer <token>'
        parts = authorization_header.split()
        
        if len(parts) == 2 and parts[0] == 'Bearer':
            return parts[1]
        else:
            raise ValueError("Invalid Authorization header format")
    else:
        raise ValueError("Authorization header missing")
    
@mcp.tool()
def add(a: int, b: int, ctx:Context) -> int:
    """Add two numbers"""
    token = get_bearer_token(ctx)
    print(f"Client token: {token}")

    return a + b

if __name__ == "__main__":
    mcp.run(transport="sse", port=9000, host="localhost")

Sample client.py

import asyncio
from fastmcp.client import Client
from fastmcp.client.transports import SSETransport

async def main():

    # Define the Bearer token
    token = "expected-token-chistopher-nolan"

    # Create a custom SSE transport with the Authorization header
    transport = SSETransport(
        url="http://localhost:9000/sse",
        headers={"Authorization": f"Bearer {token}"}
    )

    # MCP client with JWT token in the header
    async with Client(transport) as client:
        # Make calls to tools, JWT is automatically sent with every request
        resp = await client.call_tool("add", {"a": 2, "b": 3})
        print("2 + 3 =", resp[0].text)

if __name__ == "__main__":
    asyncio.run(main())

[josx]

My last call on this

https://github.com/josx/mcp_demo/blob/main/server_asgi.py

[ammmir]

does anyone have a simple working example, like a literal if bearer == "foo" type of example? I'd prefer not to call a method in each MCP tool to fish out headers and do auth checks; that's what middleware is for.

[HoaX7]

im looking to solve this too. But if you send the headers manually from your client, how do you identify which tools require auth? this could lead to security issues right? if you are sending auth tokens to mcp servers you don't own.

[HoaX7]

how would you implement authentication for your mcp server to be used with something like claude desktop?

[mehmetaltuntas]

Here is a simple trick that worked for us. We are building a "relay" MCP Server. Meaning, we need to do token forwarding from our MCP Client on to the MCP Server (and down to an API Server)

Sample server.py

from fastmcp import FastMCP, Context

mcp = FastMCP("Demo 🚀")

def get_bearer_token(ctx):
request = ctx.get_http_request()
headers = request.headers
# Check if 'Authorization' header is present
authorization_header = headers.get('Authorization')

if authorization_header:
    # Split the header into 'Bearer <token>'
    parts = authorization_header.split()
    
    if len(parts) == 2 and parts[0] == 'Bearer':
        return parts[1]
    else:
        raise ValueError("Invalid Authorization header format")
else:
    raise ValueError("Authorization header missing")

@mcp.tool()
def add(a: int, b: int, ctx:Context) -> int:
"""Add two numbers"""
token = get_bearer_token(ctx)
print(f"Client token: {token}")

return a + b

if name == "main":
mcp.run(transport="sse", port=9000, host="localhost")
Sample client.py

import asyncio
from fastmcp.client import Client
from fastmcp.client.transports import SSETransport

async def main():

# Define the Bearer token
token = "expected-token-chistopher-nolan"

# Create a custom SSE transport with the Authorization header
transport = SSETransport(
    url="http://localhost:9000/sse",
    headers={"Authorization": f"Bearer {token}"}
)

# MCP client with JWT token in the header
async with Client(transport) as client:
    # Make calls to tools, JWT is automatically sent with every request
    resp = await client.call_tool("add", {"a": 2, "b": 3})
    print("2 + 3 =", resp[0].text)

if name == "main":
asyncio.run(main())

A good option. Driving this all that by from mcp.server.fastmcp import FastMCP bit challenging. How is possible to pass anything in the header from the client end if it is happened to be an API endpoint?

    agent_plugin = MCPSsePlugin(
        name="Agent",
        description="""
        Agent for data analysis and workspace interaction
        """,
        url="http://agent.com",
        request_timeout=10000000
    )
    await agent_plugin.connect()
    kernel.add_plugin(agent_plugin)

Alternatively, I had to capture the incoming request header and strip out the authZ value and send it down to mcp server as an argument by injecting it in the filter ( middleware )

[FirefoxMetzger]

If you run a recent version of the MCP Python SDK you can use the built-in BearerAuth logic. All you need is to define a token_verifier function and add some stub config.

Here is a small example that exports a Stockholm Time tool:

import datetime
import zoneinfo
from mcp.server.fastmcp import FastMCP
from mcp.server.auth.provider import TokenVerifier
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.settings import AuthSettings


class MyBearerValidator(TokenVerifier):
    async def verify_token(self, token: str) -> AccessToken | None:
        # TODO: replace with your custom bearer validation
        if token == "my-super-secret-bearer-token":
            # You must return token, client_id, scopes
            # You can optionally set an expiration timestamp and resource string
            return AccessToken(
                token=token,
                client_id="that_guy",
                scopes=["authenticated"],
            )
        return None


# Note: A stub config means that some /.well-known/ endpoints will be
# missconfigured
auth_settings = AuthSettings(
    issuer_url="https://awesome-mcp.com/auth",  # stub

    # Bug: resource_server_url should be None if the MCP has tools
    # somebody forgot to add a default value  in the field(...) function (:
    resource_server_url=None,
)

mcp = FastMCP(
    "MCP with Bearer auth",
    token_verifier=MyBearerValidator(),
    auth=auth_settings,
)


@mcp.tool()
def get_current_time() -> str:
    """Returns the current time in Stockholm."""
    stockholm_tz = zoneinfo.ZoneInfo("Europe/Stockholm")
    return datetime.datetime.now(stockholm_tz).strftime("%Y-%m-%d %H:%M:%S")


if __name__ == "__main__":
    # Note: works for both SSE and streamable-http
    mcp.run(transport="streamable-http")

The way this works is that you run this server as a standalone (e.g., dedicated terminal where you do python server.py). Then, in the MCP inspector you change the following settings (all others should be pre-populated correctly):

1. Transport Type: Streamable HTTP
2. URL: http://localhost:8000/mcp (or what ever you have configured)
3. Bearer Token: my-super-secret-bearer-token

Enjoy!