ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-24 06:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Improve redirect handling to address 307's (#1387)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-08-08 11:55:30 -04:00 • committed by GitHub
commit 87e103222e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 175 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.gitignore vendored
View file

@ -63,3 +63,8 @@ dmypy.json
# Claude worktree management
.claude-wt/worktrees
# Common FastMCP test files
/test.py
/server.py
/client.py

71
src/fastmcp/server/http.py
View file

@ -32,6 +32,38 @@ if TYPE_CHECKING:
logger = get_logger(__name__)
class StreamableHTTPASGIApp:
"""ASGI application wrapper for Streamable HTTP server transport."""
def __init__(self, session_manager):
self.session_manager = session_manager
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
try:
await self.session_manager.handle_request(scope, receive, send)
except RuntimeError as e:
if str(e) == "Task group is not initialized. Make sure to use run().":
logger.error(
f"Original RuntimeError from mcp library: {e}", exc_info=True
)
new_error_message = (
"FastMCP's StreamableHTTPSessionManager task group was not initialized. "
"This commonly occurs when the FastMCP application's lifespan is not "
"passed to the parent ASGI application (e.g., FastAPI or Starlette). "
"Please ensure you are setting `lifespan=mcp_app.lifespan` in your "
"parent app's constructor, where `mcp_app` is the application instance "
"returned by `fastmcp_instance.http_app()`. \\n"
"For more details, see the FastMCP ASGI integration documentation: "
"https://gofastmcp.com/deployment/asgi"
)
# Raise a new RuntimeError that includes the original error's message
# for full context, but leads with the more helpful guidance.
raise RuntimeError(f"{new_error_message}\\nOriginal error: {e}") from e
else:
# Re-raise other RuntimeErrors if they don't match the specific message
raise
_current_http_request: ContextVar[Request | None] = ContextVar(
"http_request",
default=None,
@ -254,33 +286,8 @@ def create_streamable_http_app(
stateless=stateless_http,
)
# Create the ASGI handler
async def handle_streamable_http(
scope: Scope, receive: Receive, send: Send
) -> None:
try:
await session_manager.handle_request(scope, receive, send)
except RuntimeError as e:
if str(e) == "Task group is not initialized. Make sure to use run().":
logger.error(
f"Original RuntimeError from mcp library: {e}", exc_info=True
)
new_error_message = (
"FastMCP's StreamableHTTPSessionManager task group was not initialized. "
"This commonly occurs when the FastMCP application's lifespan is not "
"passed to the parent ASGI application (e.g., FastAPI or Starlette). "
"Please ensure you are setting `lifespan=mcp_app.lifespan` in your "
"parent app's constructor, where `mcp_app` is the application instance "
"returned by `fastmcp_instance.http_app()`. \\n"
"For more details, see the FastMCP ASGI integration documentation: "
"https://gofastmcp.com/deployment/asgi"
)
# Raise a new RuntimeError that includes the original error's message
# for full context, but leads with the more helpful guidance.
raise RuntimeError(f"{new_error_message}\\nOriginal error: {e}") from e
else:
# Re-raise other RuntimeErrors if they don't match the specific message
raise
# Create the ASGI app wrapper
streamable_http_app = StreamableHTTPASGIApp(session_manager)
# Add StreamableHTTP routes with or without auth
if auth:
@ -305,19 +312,19 @@ def create_streamable_http_app(
# Auth is enabled, wrap endpoint with RequireAuthMiddleware
server_routes.append(
Mount(
Route(
streamable_http_path,
app=RequireAuthMiddleware(
handle_streamable_http, required_scopes, resource_metadata_url
endpoint=RequireAuthMiddleware(
streamable_http_app, required_scopes, resource_metadata_url
),
)
)
else:
# No auth required
server_routes.append(
Mount(
Route(
streamable_http_path,
app=handle_streamable_http,
endpoint=streamable_http_app,
)
)

4
src/fastmcp/settings.py
View file

@ -222,9 +222,9 @@ class Settings(BaseSettings):
# HTTP settings
host: str = "127.0.0.1"
port: int = 8000
sse_path: str = "/sse/"
sse_path: str = "/sse"
message_path: str = "/messages/"
streamable_http_path: str = "/mcp/"
streamable_http_path: str = "/mcp"
debug: bool = False
# error handling

6
tests/client/test_openapi_experimental.py
View file

@ -58,13 +58,13 @@ def run_proxy_server(host: str, port: int, shttp_url: str, **kwargs) -> None:
@pytest.fixture(scope="module")
def shttp_server() -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="http") as url:
yield f"{url}/mcp/"
yield f"{url}/mcp"
@pytest.fixture(scope="module")
def sse_server() -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="sse") as url:
yield f"{url}/sse/"
yield f"{url}/sse"
@pytest.fixture(scope="module")
@ -74,7 +74,7 @@ def proxy_server(shttp_server: str) -> Generator[str, None, None]:
shttp_url=shttp_server,
transport="http",
) as url:
yield f"{url}/mcp/"
yield f"{url}/mcp"
async def test_fastapi_client_headers_streamable_http_resource(shttp_server: str):

6
tests/client/test_openapi_legacy.py
View file

@ -55,13 +55,13 @@ def run_proxy_server(host: str, port: int, shttp_url: str, **kwargs) -> None:
@pytest.fixture(scope="module")
def shttp_server() -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="http") as url:
yield f"{url}/mcp/"
yield f"{url}/mcp"
@pytest.fixture(scope="module")
def sse_server() -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="sse") as url:
yield f"{url}/sse/"
yield f"{url}/sse"
@pytest.fixture(scope="module")
@ -71,7 +71,7 @@ def proxy_server(shttp_server: str) -> Generator[str, None, None]:
shttp_url=shttp_server,
transport="http",
) as url:
yield f"{url}/mcp/"
yield f"{url}/mcp"
async def test_fastapi_client_headers_streamable_http_resource(shttp_server: str):

12
tests/client/test_streamable_http.py
View file

@ -88,7 +88,7 @@ def run_server(host: str, port: int, stateless_http: bool = False, **kwargs) ->
def run_nested_server(host: str, port: int) -> None:
mcp_app = fastmcp_server().http_app(path="/final/mcp/")
mcp_app = fastmcp_server().http_app(path="/final/mcp")
mount = Starlette(routes=[Mount("/nest-inner", app=mcp_app)])
mount2 = Starlette(
@ -115,9 +115,7 @@ async def streamable_http_server(
with run_server_in_process(
run_server, stateless_http=stateless_http, transport="http"
) as url:
async with Client(transport=StreamableHttpTransport(f"{url}/mcp/")) as client:
assert await client.ping()
yield f"{url}/mcp/"
yield f"{url}/mcp"
@pytest.fixture()
@ -126,9 +124,7 @@ async def streamable_http_server_with_streamable_http_alias() -> AsyncGenerator[
]:
"""Test that the "streamable-http" transport alias works."""
with run_server_in_process(run_server, transport="streamable-http") as url:
async with Client(transport=StreamableHttpTransport(f"{url}/mcp/")) as client:
assert await client.ping()
yield f"{url}/mcp/"
yield f"{url}/mcp"
async def test_ping(streamable_http_server: str):
@ -211,7 +207,7 @@ async def test_nested_streamable_http_server_resolves_correctly():
with run_server_in_process(run_nested_server) as url:
async with Client(
transport=StreamableHttpTransport(f"{url}/nest-outer/nest-inner/final/mcp/")
transport=StreamableHttpTransport(f"{url}/nest-outer/nest-inner/final/mcp")
) as client:
result = await client.ping()
assert result is True

27
tests/server/auth/test_static_token_verifier.py
View file

@ -65,14 +65,37 @@ class TestStaticTokenVerifier:
# Create HTTP app
app = server.http_app(transport="http")
# Test unauthenticated request gets 401
# Test unauthenticated request gets 401 (use exact path match to avoid redirect)
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.post("/mcp/")
response = await client.post("/mcp")
assert response.status_code == 401
assert "WWW-Authenticate" in response.headers
async def test_server_with_token_verifier_redirect_behavior(self):
"""Test that FastMCP server redirects non-matching paths correctly."""
verifier = StaticTokenVerifier(
{"test-token": {"client_id": "test-client", "scopes": ["read", "write"]}}
)
server = FastMCP("TestServer", auth=verifier)
@server.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
# Create HTTP app (default path is /mcp)
app = server.http_app(transport="http")
# Test that non-matching path gets 307 redirect
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.post("/mcp/", follow_redirects=False)
assert response.status_code == 307
assert response.headers["location"] == "http://test/mcp"
def test_server_rejects_both_oauth_and_token_verifier(self):
"""Test that server raises error when both OAuth and TokenVerifier provided."""
from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider

18
tests/server/http/test_http_auth_middleware.py
View file

@ -1,6 +1,6 @@
import pytest
from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
from starlette.routing import Mount
from starlette.routing import Route
from fastmcp.server import FastMCP
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
@ -36,11 +36,11 @@ class TestStreamableHTTPAppResourceMetadataURL:
auth=bearer_auth_provider,
)
mount = next(r for r in app.routes if isinstance(r, Mount) and r.path == "/mcp")
route = next(r for r in app.routes if isinstance(r, Route) and r.path == "/mcp")
assert isinstance(mount.app, RequireAuthMiddleware)
assert isinstance(route.endpoint, RequireAuthMiddleware)
assert (
str(mount.app.resource_metadata_url)
str(route.endpoint.resource_metadata_url)
== "https://resource.example.com/.well-known/oauth-protected-resource"
)
@ -57,11 +57,11 @@ class TestStreamableHTTPAppResourceMetadataURL:
streamable_http_path="/mcp",
auth=provider,
)
mount = next(r for r in app.routes if isinstance(r, Mount) and r.path == "/mcp")
assert isinstance(mount.app, RequireAuthMiddleware)
route = next(r for r in app.routes if isinstance(r, Route) and r.path == "/mcp")
assert isinstance(route.endpoint, RequireAuthMiddleware)
# Should not have double slash
assert (
str(mount.app.resource_metadata_url)
str(route.endpoint.resource_metadata_url)
== "https://resource.example.com/.well-known/oauth-protected-resource"
)
@ -74,5 +74,5 @@ class TestStreamableHTTPAppResourceMetadataURL:
streamable_http_path="/mcp",
auth=None,
)
mount = next(r for r in app.routes if isinstance(r, Mount) and r.path == "/mcp")
assert not isinstance(mount.app, RequireAuthMiddleware)
route = next(r for r in app.routes if isinstance(r, Route) and r.path == "/mcp")
assert not isinstance(route.endpoint, RequireAuthMiddleware)

85
tests/server/test_streamable_http_no_redirect.py Normal file
View file

@ -0,0 +1,85 @@
"""Test that streamable HTTP routes avoid 307 redirects."""
import httpx
import pytest
from starlette.routing import Route
from fastmcp import FastMCP
@pytest.mark.parametrize(
"server_path",
["/mcp", "/mcp/"],
)
def test_streamable_http_route_structure(server_path: str):
"""Test that streamable HTTP routes use Route objects with correct paths."""
mcp = FastMCP("TestServer")
@mcp.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
# Create HTTP app with specific path
app = mcp.http_app(transport="http", path=server_path)
# Find the streamable HTTP route
streamable_routes = [
r
for r in app.routes
if isinstance(r, Route) and hasattr(r, "path") and r.path == server_path
]
# Verify route exists and uses Route (not Mount)
assert len(streamable_routes) == 1, (
f"Should have one streamable route for path {server_path}"
)
assert isinstance(streamable_routes[0], Route), "Should use Route, not Mount"
assert streamable_routes[0].path == server_path, (
f"Route path should match {server_path}"
)
async def test_streamable_http_redirect_behavior():
"""Test that non-matching paths get redirected correctly."""
mcp = FastMCP("TestServer")
@mcp.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
# Create HTTP app with /mcp path (no trailing slash)
app = mcp.http_app(transport="http", path="/mcp")
# Test that /mcp/ gets redirected to /mcp
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), base_url="http://test"
) as client:
response = await client.get("/mcp/", follow_redirects=False)
assert response.status_code == 307
assert response.headers["location"] == "http://test/mcp"
async def test_streamable_http_no_mount_routes():
"""Test that streamable HTTP app creates Route objects, not Mount objects."""
mcp = FastMCP("TestServer")
app = mcp.http_app(transport="http")
# Should not find any Mount routes for the streamable HTTP path
from starlette.routing import Mount
mount_routes = [
r
for r in app.routes
if isinstance(r, Mount) and hasattr(r, "path") and r.path == "/mcp"
]
assert len(mount_routes) == 0, "Should not have Mount routes for streamable HTTP"
# Should find Route objects instead
route_routes = [
r
for r in app.routes
if isinstance(r, Route) and hasattr(r, "path") and r.path == "/mcp"
]
assert len(route_routes) == 1, "Should have exactly one Route for streamable HTTP"