MCP Python SDK: How to Authorize a Client with Bearer Header over SSE

To authorize a client with a Bearer header when using the MCP Python SDK over SSE, you must intercept the HTTP request before it reaches the MCP server and extract the token from the Authorization header. The simplest approach is to add a FastAPI middleware that reads the header and stores the token in a thread-safe context, then call a validation function inside each tool. A cleaner, thread-safe solution uses the FastMCP middleware system, which avoids global variables and is the recommended approach for production use.

The Full Answer

Understanding the Problem

The MCP Python SDK (modelcontextprotocol/python-sdk) provides the FastMCP class for building MCP servers. When you mount an MCP server onto a FastAPI application using app.mount("/", mcp.sse_app()), the MCP server handles all requests at that mount point. The issue is that the MCP server's tool functions do not have direct access to the incoming HTTP request object. They are called by the MCP protocol layer after the request has been parsed, so you cannot simply add a Request parameter to your tool function and read headers from it.

This is a common challenge for anyone building a remote MCP server (as opposed to a local stdio server) that needs authentication. The SSE transport is deprecated according to the official documentation, but it is still widely used, and the same principles apply to HTTP transport as well.

Solution 1: FastAPI Middleware with Global Variable (Simple, Not Thread-Safe)

This solution was originally posted by Roman Gelembjuk on Stack Overflow. It is the quickest way to get authentication working, but it has a significant caveat: it uses a global variable to store the token, which is not thread-safe. If multiple requests arrive concurrently, the token from one request can overwrite the token from another, leading to incorrect authentication.

Here is the code:

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

# Global variable to keep a token 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())

How it works:

1. The auth_middleware FastAPI middleware runs on every HTTP request before it reaches the MCP server. It reads the Authorization header from the request.
2. If the header is present, it extracts the token (the part after "Bearer ") and stores it in the global variable auth_token.
3. Each tool function calls require_auth() at the start. This function checks if the stored token matches the expected value. If not, it raises a ValueError, which the MCP server converts into an error response.
4. The tool then proceeds with its actual logic.

When to use this:

This approach is suitable for:

• Prototyping and proof-of-concept work
• Single-user or low-concurrency scenarios where you are certain only one request will be processed at a time
• Learning the authentication flow before moving to a more robust solution

Caveats (community-reported):

• The global variable is not thread-safe. If two requests arrive simultaneously, the token from the second request overwrites the first before the first request's tool function reads it. This means the first request might be authenticated with the second request's token, or the second request might fail because the first request's token is still in the variable.
• The token is stored in memory and persists across requests. If a request does not include an Authorization header, the previous request's token remains in the variable, potentially allowing unauthorized access.
• The author of this solution (Roman Gelembjuk) explicitly states: "The solution i originally posted here is not great. It is not thread safe." He recommends the second solution instead.

Solution 2: FastMCP Middleware with Thread-Local Storage (Thread-Safe, Recommended)

This is the improved solution described in Roman Gelembjuk's blog post. It uses the FastMCP middleware system, which is designed to handle request context in a thread-safe manner. This is the recommended approach for production deployments.

The core idea is to use a middleware that runs within the FastMCP server's request processing pipeline, not at the FastAPI level. This middleware can access the request headers and store the token in a thread-local or context variable that is isolated per request.

Here is the code (adapted from the blog post):

from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI, Request
import subprocess
import shlex
from contextvars import ContextVar
from starlette.middleware.base import BaseHTTPMiddleware

# Thread-safe context variable to hold the token for the current request
auth_token_var: ContextVar[str] = ContextVar("auth_token", default="")

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

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        auth_header = request.headers.get("Authorization")
        if auth_header:
            token = auth_header.split(" ")[1]
            auth_token_var.set(token)
        else:
            auth_token_var.set("")
        response = await call_next(request)
        return response

app.add_middleware(AuthMiddleware)

def require_auth():
    """
    Check access and raise an error if the token is not valid.
    """
    token = auth_token_var.get()
    if 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()
    return run_cli(command, work_dir)

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

How it works:

1. A ContextVar from Python's contextvars module is used to store the token. ContextVar is thread-safe and automatically scopes the value to the current asynchronous context (or thread). This means each request gets its own isolated copy of the token.
2. The AuthMiddleware class extends BaseHTTPMiddleware from Starlette (which FastAPI is built on). The dispatch method runs on every request. It reads the Authorization header and sets the token in the ContextVar. If no header is present, it sets an empty string.
3. The middleware is added to the FastAPI app with app.add_middleware(AuthMiddleware). This ensures it runs before the MCP server's handler.
4. The require_auth() function retrieves the token from the ContextVar and compares it to the expected value. If it does not match, it raises a ValueError.
5. Each tool function calls require_auth() at the start.

Why this is better:

• Thread-safe: ContextVar ensures that each request (or async task) has its own token value. Concurrent requests do not interfere with each other.
• No global state: The token is scoped to the request context and is automatically cleaned up when the request finishes.
• Works with async and sync code: ContextVar works in both synchronous and asynchronous contexts.
• Follows FastAPI/Starlette best practices for middleware.

When to use this:

This is the recommended solution for:

• Production deployments with multiple concurrent users
• Any scenario where request isolation is important
• When you want a clean, maintainable codebase

Solution 3: Using the headersHelper Configuration (Claude Code Specific)

If you are using Claude Code as the MCP client, you can configure authentication at the client side using the headersHelper field in the MCP server configuration. This is documented in the official Anthropic documentation for Claude Code.

When adding a remote HTTP or SSE server with claude mcp add, you can pass headers directly:

# For HTTP transport
claude mcp add --transport http secure-api https://api.example.com/mcp \
    --header "Authorization: Bearer your-token"

# For SSE transport (deprecated)
claude mcp add --transport sse private-api https://api.company.com/sse \
    --header "X-API-Key: your-key-here"

For more complex scenarios where the token needs to be generated dynamically (e.g., from an environment variable or a token refresh flow), you can use the headersHelper field in the JSON configuration. This is a function that runs at connect time and returns the headers.

Example JSON configuration for a WebSocket server (the same pattern applies to HTTP and SSE):

{
  "mcpServers": {
    "secure-api": {
      "type": "http",
      "url": "https://api.example.com/mcp",
      "headersHelper": "process.env.MY_TOKEN ? { 'Authorization': 'Bearer ' + process.env.MY_TOKEN } : {}"
    }
  }
}

This approach is useful when the token is not static and must be fetched or computed at connection time. The headersHelper is a string that is evaluated as JavaScript code in the Claude Code process context.

Important caveat from the documentation:

• The headersHelper field supports path placeholders like ${CLAUDE_PLUGIN_ROOT} for plugin-provided servers. Before v2.1.195, headersHelper passed the placeholder through as a literal string instead of substituting it.
• For WebSocket servers, authentication is header-only, so you must pass a static token in headers or generate one with headersHelper.

This solution is client-side configuration. It does not change how your server validates the token. You still need to implement the server-side validation using one of the middleware approaches above.

How to Test Your Authenticated MCP Server

Once you have implemented server-side authentication, you can test it with a simple HTTP client like curl:

# Test without token (should fail)
curl -X POST https://your-server.com/mcp \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}'

# Test with valid token (should succeed)
curl -X POST https://your-server.com/mcp \
  -H "Authorization: Bearer expected-token" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}'

# Test with invalid token (should fail)
curl -X POST https://your-server.com/mcp \
  -H "Authorization: Bearer wrong-token" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}'

When the token is invalid, the server should return an error response with a ValueError message. When the token is valid, it should return the list of available tools.

Integrating with Claude Code

To connect your authenticated MCP server to Claude Code, use the claude mcp add command with the appropriate headers:

claude mcp add --transport http my-server https://your-server.com/mcp \
  --header "Authorization: Bearer your-token"

If you are using SSE transport (deprecated):

claude mcp add --transport sse my-server https://your-server.com/sse \
  --header "Authorization: Bearer your-token"

Claude Code will include this header in every request to your MCP server. Your server's middleware will extract the token and validate it before allowing any tool calls.