Handle the McpError from the middleware when initialize the session #2552
Description
Enhancement
It is possible to create a middleware that customizes the on_initialize method.
I would like to make it possible to fail the initialize request by returning a JSON RPC Error in the middleware. For example, if the users credentials is invalid, fail the connection.
For example, in a proxy server:
class InitializeMiddleware(Middleware):
"""Intecept MCP initialize request and initialize the proxy client."""
def __init__(self, client_factory: AWSMCPProxyClientFactory) -> None:
"""Create a middleware with client factory."""
super().__init__()
@override
async def on_initialize(
self,
context: MiddlewareContext[mt.InitializeRequest],
call_next: CallNext[mt.InitializeRequest, None],
) -> None:
if not is_valid(credentials):
# the error data will be sent as a json rpc error as the error of initialize
raise McpError(ErrorData(...))
return await call_next(context)Implemented in #2531