Merge pull request #558 from jlowin/custom-routes

src/fastmcp/server/http.py

@@ -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:

src/fastmcp/server/server.py

@@ -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,
             )
 

tests/server/http/test_custom_routes.py

@@ -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()"
+            )