MCP Python SDK. How to authorise a client with Bearer header with SSE?

I am building the MCP server application to connect some services to LLM . I use the MCP Python SDK https://github.com/modelcontextprotocol/python-sdk One of things i want to implement is authorisation of a user with the token.

I see it must be possible somehow.

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?

If someone is still looking for this. There is the solution.

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

# Global variable to keep a token a for a request
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

def require_auth():
    """
    Check access and raise an error if the token is not valid.
    """

    if auth_token != "expected-token":
        raise ValueError("Invalid token")
    return None

def run_cli(command: str, cwd: str = None) -> str:
        """
        Execute a CLI command using subprocess."""

        if cwd == "":
            cwd = None

        command_list = shlex.split(command)
        
        run_result = subprocess.run(
            command_list,
            cwd=cwd,
            capture_output=True,
            text=True,
            check=False,
        )
        success = run_result.returncode == 0
        return f"STDOUT: {run_result.stdout}\n\nSTDERR: {run_result.stderr}\nRETURNCODE: {run_result.returncode}\nSUCCESS: {success}"

@mcp.tool()
def cli_command(command: str, work_dir: str | None = "") -> str:
    """
    Execute command line cli command on the Linux server. 
    
    Arguments:
        command - command to execute.
        work_dir - workdir will be changed to this path before executing the command.
    """
    require_auth() # we have to add this inside each tool method
    return run_cli(command, work_dir)

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

But as i understand there will be better ways to do this soon because developers of that python sdk have some ideas how to support this