feat: handle error from the initialize middleware

In some situation, the initialize middleware can check the status of the
server and decide to raise an error.

Example use case: in a FastMCPProxy, an initialization middleware
overrides the on_initialize method and connect to the underlying proxied
client. When client respond with error, I want to pass this error to the
client.

src/fastmcp/server/low_level.py

@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any
 import anyio
 import mcp.types
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from mcp import McpError
 from mcp.server.lowlevel.server import (
     LifespanResultT,
     NotificationOptions,
@@ -104,9 +105,26 @@ class MiddlewareServerSession(ServerSession):
                     fastmcp_context=fastmcp_ctx,
                 )
 
-                return await self.fastmcp._apply_middleware(
-                    mw_context, call_original_handler
-                )
+                # return await self.fastmcp._apply_middleware(
+                #     mw_context, call_original_handler
+                # )
+                try:
+                    return await self.fastmcp._apply_middleware(
+                        mw_context, call_original_handler
+                    )
+                except McpError as e:
+                    # McpError can be thrown from middleware in `on_initialize`
+                    # send the error to responder.
+                    if not responder._completed:
+                        with responder:
+                            await responder.respond(e.error)
+                    else:
+                        # Don't re-raise: prevents responding to initialize request twice
+                        logger.warning(
+                            "Received McpError but responder is already completed. "
+                            "Cannot send error response as response was already sent.",
+                            exc_info=e,
+                        )
 
         # Fall through to default handling (task methods now handled via registered handlers)
         return await super()._received_request(responder)

tests/server/middleware/test_initialization_middleware.py

@@ -3,6 +3,9 @@
 from typing import Any
 
 import mcp.types as mt
+import pytest
+from mcp import McpError
+from mcp.types import ErrorData
 
 from fastmcp import Client, FastMCP
 from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
@@ -286,3 +289,83 @@ async def test_middleware_can_access_initialize_result():
         assert middleware.initialize_result.serverInfo.name == "TestServer"
         assert middleware.initialize_result.protocolVersion is not None
         assert middleware.initialize_result.capabilities is not None
+
+
+async def test_middleware_mcp_error_during_initialization():
+    """Test that McpError raised in middleware during initialization is sent to responder."""
+    server = FastMCP("TestServer")
+
+    class ErrorThrowingMiddleware(Middleware):
+        async def on_initialize(
+            self,
+            context: MiddlewareContext[mt.InitializeRequest],
+            call_next: CallNext[mt.InitializeRequest, None],
+        ) -> None:
+            raise McpError(
+                ErrorData(
+                    code=mt.INVALID_PARAMS, message="Invalid initialization parameters"
+                )
+            )
+
+    server.add_middleware(ErrorThrowingMiddleware())
+
+    with pytest.raises(Exception) as exc_info:
+        async with Client(server):
+            pass
+
+    assert "Invalid initialization parameters" in str(exc_info.value)
+
+
+async def test_middleware_mcp_error_before_call_next():
+    """Test McpError raised before calling next middleware."""
+    server = FastMCP("TestServer")
+
+    class EarlyErrorMiddleware(Middleware):
+        async def on_initialize(
+            self,
+            context: MiddlewareContext[mt.InitializeRequest],
+            call_next: CallNext[mt.InitializeRequest, None],
+        ) -> None:
+            raise McpError(
+                ErrorData(code=mt.INVALID_REQUEST, message="Request validation failed")
+            )
+
+    server.add_middleware(EarlyErrorMiddleware())
+
+    with pytest.raises(Exception) as exc_info:
+        async with Client(server):
+            pass
+
+    assert "Request validation failed" in str(exc_info.value)
+
+
+async def test_middleware_mcp_error_after_call_next():
+    """Test that McpError raised after call_next doesn't break the connection.
+
+    When an error is raised after call_next, the responder has already completed,
+    so the error is caught but not sent to the responder (checked via _completed flag).
+    """
+    server = FastMCP("TestServer")
+
+    class PostProcessingErrorMiddleware(Middleware):
+        def __init__(self):
+            super().__init__()
+            self.error_raised = False
+
+        async def on_initialize(
+            self,
+            context: MiddlewareContext[mt.InitializeRequest],
+            call_next: CallNext[mt.InitializeRequest, mt.InitializeResult | None],
+        ) -> mt.InitializeResult | None:
+            await call_next(context)
+            self.error_raised = True
+            raise McpError(
+                ErrorData(code=mt.INTERNAL_ERROR, message="Post-processing failed")
+            )
+
+    middleware = PostProcessingErrorMiddleware()
+    server.add_middleware(middleware)
+
+    # Connection succeeds because responder._completed check prevents re-responding
+    async with Client(server):
+        assert middleware.error_raised is True