mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 05:24:18 +02:00
Ensure custom routes are respected
This commit is contained in:
parent
4dc6fe2941
commit
189389a3a4
5 changed files with 107 additions and 3 deletions
|
|
@ -241,6 +241,7 @@ def create_sse_app(
|
|||
# Add custom routes with lowest precedence
|
||||
if routes:
|
||||
server_routes.extend(routes)
|
||||
server_routes.extend(server._additional_http_routes)
|
||||
|
||||
# Add middleware
|
||||
if middleware:
|
||||
|
|
@ -359,6 +360,7 @@ def create_streamable_http_app(
|
|||
# Add custom routes with lowest precedence
|
||||
if routes:
|
||||
server_routes.extend(routes)
|
||||
server_routes.extend(server._additional_http_routes)
|
||||
|
||||
# Add middleware
|
||||
if middleware:
|
||||
|
|
|
|||
|
|
@ -854,7 +854,6 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
auth_server_provider=self._auth_server_provider,
|
||||
auth_settings=self.settings.auth,
|
||||
debug=self.settings.debug,
|
||||
routes=self._additional_http_routes,
|
||||
middleware=middleware,
|
||||
)
|
||||
|
||||
|
|
@ -905,7 +904,6 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
json_response=self.settings.json_response,
|
||||
stateless_http=self.settings.stateless_http,
|
||||
debug=self.settings.debug,
|
||||
routes=self._additional_http_routes,
|
||||
middleware=middleware,
|
||||
)
|
||||
elif transport == "sse":
|
||||
|
|
@ -916,7 +914,6 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
auth_server_provider=self._auth_server_provider,
|
||||
auth_settings=self.settings.auth,
|
||||
debug=self.settings.debug,
|
||||
routes=self._additional_http_routes,
|
||||
middleware=middleware,
|
||||
)
|
||||
|
||||
|
|
|
|||
105
tests/server/http/test_custom_routes.py
Normal file
105
tests/server/http/test_custom_routes.py
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
import pytest
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import JSONResponse
|
||||
from starlette.routing import Route
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.http import create_sse_app, create_streamable_http_app
|
||||
|
||||
|
||||
class TestCustomRoutes:
|
||||
@pytest.fixture
|
||||
def server_with_custom_route(self):
|
||||
"""Create a FastMCP server with a custom route."""
|
||||
server = FastMCP()
|
||||
|
||||
@server.custom_route("/custom-route", methods=["GET"])
|
||||
async def custom_route(request: Request):
|
||||
return JSONResponse({"message": "custom route"})
|
||||
|
||||
return server
|
||||
|
||||
def test_custom_routes_via_server_http_app(self, server_with_custom_route):
|
||||
"""Test that custom routes are included when using server.http_app()."""
|
||||
# Get the app via server.http_app()
|
||||
app = server_with_custom_route.http_app()
|
||||
|
||||
# Verify that the custom route is included
|
||||
custom_route_found = False
|
||||
for route in app.routes:
|
||||
if isinstance(route, Route) and route.path == "/custom-route":
|
||||
custom_route_found = True
|
||||
break
|
||||
|
||||
assert custom_route_found, "Custom route was not found in app routes"
|
||||
|
||||
def test_custom_routes_via_streamable_http_app_direct(
|
||||
self, server_with_custom_route
|
||||
):
|
||||
"""Test that custom routes are included when using create_streamable_http_app directly."""
|
||||
# Create the app by calling the constructor function directly
|
||||
app = create_streamable_http_app(
|
||||
server=server_with_custom_route, streamable_http_path="/api"
|
||||
)
|
||||
|
||||
# Verify that the custom route is included
|
||||
custom_route_found = False
|
||||
for route in app.routes:
|
||||
if isinstance(route, Route) and route.path == "/custom-route":
|
||||
custom_route_found = True
|
||||
break
|
||||
|
||||
assert custom_route_found, "Custom route was not found in app routes"
|
||||
|
||||
def test_custom_routes_via_sse_app_direct(self, server_with_custom_route):
|
||||
"""Test that custom routes are included when using create_sse_app directly."""
|
||||
# Create the app by calling the constructor function directly
|
||||
app = create_sse_app(
|
||||
server=server_with_custom_route, message_path="/message", sse_path="/sse"
|
||||
)
|
||||
|
||||
# Verify that the custom route is included
|
||||
custom_route_found = False
|
||||
for route in app.routes:
|
||||
if isinstance(route, Route) and route.path == "/custom-route":
|
||||
custom_route_found = True
|
||||
break
|
||||
|
||||
assert custom_route_found, "Custom route was not found in app routes"
|
||||
|
||||
def test_multiple_custom_routes(
|
||||
self,
|
||||
):
|
||||
"""Test that multiple custom routes are included in both methods."""
|
||||
server = FastMCP()
|
||||
|
||||
custom_paths = ["/route1", "/route2", "/route3"]
|
||||
|
||||
# Add multiple custom routes
|
||||
for path in custom_paths:
|
||||
|
||||
@server.custom_route(path, methods=["GET"])
|
||||
async def custom_route(request: Request):
|
||||
return JSONResponse({"message": f"route {path}"})
|
||||
|
||||
# Test with server.http_app()
|
||||
app1 = server.http_app()
|
||||
|
||||
# Test with direct constructor call
|
||||
app2 = create_streamable_http_app(server=server, streamable_http_path="/api")
|
||||
|
||||
# Check all routes are in both apps
|
||||
for path in custom_paths:
|
||||
# Check in app1
|
||||
route_in_app1 = any(
|
||||
isinstance(route, Route) and route.path == path for route in app1.routes
|
||||
)
|
||||
assert route_in_app1, f"Route {path} not found in server.http_app()"
|
||||
|
||||
# Check in app2
|
||||
route_in_app2 = any(
|
||||
isinstance(route, Route) and route.path == path for route in app2.routes
|
||||
)
|
||||
assert route_in_app2, (
|
||||
f"Route {path} not found in create_streamable_http_app()"
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue