Spaces:
Running
Running
File size: 3,374 Bytes
1534a61 60b71f5 1534a61 341332a 60b71f5 1534a61 95e8d10 1534a61 f64fa17 1534a61 c14bd76 c457a0d 56481cb c457a0d f64fa17 c457a0d fe933f9 9d030a0 c14bd76 9d030a0 c14bd76 60b71f5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | from __future__ import annotations
from typing import TYPE_CHECKING, ParamSpec, TypeVar
from mcp.server.auth.middleware.auth_context import (
get_access_token as _sdk_get_access_token,
)
from starlette.requests import Request
from fastmcp.server.auth import AccessToken
if TYPE_CHECKING:
from fastmcp.server.context import Context
P = ParamSpec("P")
R = TypeVar("R")
__all__ = [
"get_context",
"get_http_request",
"get_http_headers",
"get_access_token",
"AccessToken",
]
# --- Context ---
def get_context() -> Context:
from fastmcp.server.context import _current_context
context = _current_context.get()
if context is None:
raise RuntimeError("No active context found.")
return context
# --- HTTP Request ---
def get_http_request() -> Request:
from mcp.server.lowlevel.server import request_ctx
request = None
try:
request = request_ctx.get().request
except LookupError:
pass
if request is None:
raise RuntimeError("No active HTTP request found.")
return request
def get_http_headers(include_all: bool = False) -> dict[str, str]:
"""
Extract headers from the current HTTP request if available.
Never raises an exception, even if there is no active HTTP request (in which case
an empty dict is returned).
By default, strips problematic headers like `content-length` that cause issues if forwarded to downstream clients.
If `include_all` is True, all headers are returned.
"""
if include_all:
exclude_headers = set()
else:
exclude_headers = {
"host",
"content-length",
"connection",
"transfer-encoding",
"upgrade",
"te",
"keep-alive",
"expect",
"accept",
# Proxy-related headers
"proxy-authenticate",
"proxy-authorization",
"proxy-connection",
# MCP-related headers
"mcp-session-id",
}
# (just in case)
if not all(h.lower() == h for h in exclude_headers):
raise ValueError("Excluded headers must be lowercase")
headers = {}
try:
request = get_http_request()
for name, value in request.headers.items():
lower_name = name.lower()
if lower_name not in exclude_headers:
headers[lower_name] = str(value)
return headers
except RuntimeError:
return {}
# --- Access Token ---
def get_access_token() -> AccessToken | None:
"""
Get the FastMCP access token from the current context.
Returns:
The access token if an authenticated user is available, None otherwise.
"""
#
obj = _sdk_get_access_token()
if obj is None or isinstance(obj, AccessToken):
return obj
# If the object is not a FastMCP AccessToken, convert it to one if the fields are compatible
# This is a workaround for the case where the SDK returns a different type
# If it fails, it will raise a TypeError
try:
return AccessToken(**obj.model_dump())
except Exception as e:
raise TypeError(
f"Expected fastmcp.server.auth.auth.AccessToken, got {type(obj).__name__}. "
"Ensure the SDK is using the correct AccessToken type."
) from e
|