Fix stream resource leaks and upgrade Starlette

- Add docstring for custom_route method in FastMCP server
- Fix stream resource leaks in SSE transport and streaming ASGI response
- Upgrade Starlette to 0.46.0+ to remove multipart deprecation warning
- Remove python-multipart dependency which is now included in Starlette

Author: bhosmer-ant

pyproject.toml

@@ -26,7 +26,7 @@ dependencies = [
     "httpx>=0.27",
     "httpx-sse>=0.4",
     "pydantic>=2.7.2,<3.0.0",
-    "starlette>=0.27",
+    "starlette>=0.46",
     "sse-starlette>=1.6.1",
     "pydantic-settings>=2.5.2",
     "uvicorn>=0.23.1",
@@ -110,12 +110,8 @@ mcp = { workspace = true }
 xfail_strict = true
 filterwarnings = [
     "error",
-    # this is a long-standing issue with fastmcp, which is just now being exercised by tests
-    "ignore:Unclosed:ResourceWarning",
     # This should be fixed on Uvicorn's side.
     "ignore::DeprecationWarning:websockets",
     "ignore:websockets.server.WebSocketServerProtocol is deprecated:DeprecationWarning",
     "ignore:Returning str or bytes.*:DeprecationWarning:mcp.server.lowlevel",
-    # this is a problem in starlette
-    "ignore:Please use `import python_multipart` instead.:PendingDeprecationWarning",
 ]

src/mcp/server/fastmcp/server.py

@@ -490,6 +490,25 @@ def custom_route(
         name: str | None = None,
         include_in_schema: bool = True,
     ):
+        """Decorator to register a custom HTTP route on the FastMCP server.
+
+        Allows adding arbitrary HTTP endpoints outside the standard MCP protocol,
+        which can be useful for OAuth callbacks, health checks, or admin APIs.
+        The handler function must be an async function that accepts a Starlette
+        Request and returns a Response.
+
+        Args:
+            path: URL path for the route (e.g., "/oauth/callback")
+            methods: List of HTTP methods to support (e.g., ["GET", "POST"])
+            name: Optional name for the route (used for URL reversing)
+            include_in_schema: Whether to include in OpenAPI schema, defaults to True
+
+        Example:
+            @server.custom_route("/health", methods=["GET"])
+            async def health_check(request: Request) -> Response:
+                return JSONResponse({"status": "ok"})
+        """
+
         def decorator(
             func: Callable[[Request], Awaitable[Response]],
         ) -> Callable[[Request], Awaitable[Response]]:

src/mcp/server/sse.py

@@ -120,15 +120,17 @@ async def sse_writer():
                         }
                     )
 
-        async with anyio.create_task_group() as tg:
-            response = EventSourceResponse(
-                content=sse_stream_reader, data_sender_callable=sse_writer
-            )
-            logger.debug("Starting SSE response task")
-            tg.start_soon(response, scope, receive, send)
-
-            logger.debug("Yielding read and write streams")
-            yield (read_stream, write_stream, response)
+        # Ensure all streams are properly closed
+        async with read_stream, write_stream, read_stream_writer, sse_stream_reader:
+            async with anyio.create_task_group() as tg:
+                response = EventSourceResponse(
+                    content=sse_stream_reader, data_sender_callable=sse_writer
+                )
+                logger.debug("Starting SSE response task")
+                tg.start_soon(response, scope, receive, send)
+
+                logger.debug("Yielding read and write streams")
+                yield (read_stream, write_stream, response)
 
     async def handle_post_message(
         self, scope: Scope, receive: Receive, send: Send

src/mcp/server/streaming_asgi_transport.py

@@ -173,6 +173,7 @@ async def process_messages() -> None:
                 # Ensure events are set even if there's an error
                 initial_response_ready.set()
                 response_complete.set()
+                await content_send_channel.aclose()
 
         # Create tasks for running the app and processing messages
         self.task_group.start_soon(run_app)
@@ -205,5 +206,8 @@ def __init__(
         self.receive_channel = receive_channel
 
     async def __aiter__(self) -> typing.AsyncIterator[bytes]:
-        async for chunk in self.receive_channel:
-            yield chunk
+        try:
+            async for chunk in self.receive_channel:
+                yield chunk
+        finally:
+            await self.receive_channel.aclose()