Middleware

Middleware

Base interfaces and built-in middleware.

arcade_mcp_server.middleware.base.Middleware

Base class for MCP middleware with typed handlers for each method.

Middleware can intercept and modify requests and responses at various stages of processing. Each handler receives the context and a call_next function to invoke the next handler in the chain.

__call__(context, call_next) async

Main entry point that orchestrates the middleware chain.

on_call_tool(context, call_next) async

Handle tool calls. Override to add tool-specific processing.

on_get_prompt(context, call_next) async

Handle prompt retrieval. Override to add prompt processing.

on_list_prompts(context, call_next) async

Handle prompt listing. Override to filter or modify prompt list.

on_list_resource_templates(context, call_next) async

Handle resource template listing. Override to filter or modify template list.

on_list_resources(context, call_next) async

Handle resource listing. Override to filter or modify resource list.

on_list_tools(context, call_next) async

Handle tool listing. Override to filter or modify tool list.

on_message(context, call_next) async

Handle any message. Override to add generic processing.

on_notification(context, call_next) async

Handle notification messages. Override to add notification processing.

on_read_resource(context, call_next) async

Handle resource reading. Override to add resource processing.

on_request(context, call_next) async

Handle request messages. Override to add request processing.

arcade_mcp_server.middleware.base.MiddlewareContext dataclass

Bases: Generic[T]

Context passed through the middleware chain.

Contains the message being processed and metadata about the request.

copy(**kwargs)

Create a copy with updated fields.

arcade_mcp_server.middleware.base.compose_middleware(*middleware)

Compose multiple middleware into a single handler.

The middleware are applied in reverse order, so the first middleware in the list is the outermost (runs first on request, last on response).

Built-ins

arcade_mcp_server.middleware.logging.LoggingMiddleware

Bases: Middleware

Middleware that logs all MCP messages and timing information.

__init__(log_level='INFO')

Initialize logging middleware.

Parameters:

Name	Type	Description	Default
log_level	str	The log level to use for message logging	'INFO'

on_message(context, call_next) async

Log all messages with timing information.

arcade_mcp_server.middleware.error_handling.ErrorHandlingMiddleware

Bases: Middleware

Middleware that handles errors and converts them to appropriate responses.

__init__(mask_error_details=True)

Initialize error handling middleware.

Parameters:

Name	Type	Description	Default
mask_error_details	bool	Whether to mask error details in responses	True

on_call_tool(context, call_next) async

Handle tool call errors specially.

on_message(context, call_next) async

Wrap all messages with error handling.

Examples

# Implement a custom middleware
from arcade_mcp_server.middleware.base import Middleware, MiddlewareContext

class TimingMiddleware(Middleware):
    async def __call__(self, context: MiddlewareContext, call_next):
        import time
        start = time.perf_counter()
        try:
            return await call_next(context)
        finally:
            elapsed_ms = (time.perf_counter() - start) * 1000
            # Attach timing info to context metadata
            context.metadata["elapsed_ms"] = round(elapsed_ms, 2)

# Compose middleware and create a server
from arcade_mcp_server.middleware.base import compose_middleware
from arcade_mcp_server.middleware.logging import LoggingMiddleware
from arcade_mcp_server.middleware.error_handling import ErrorHandlingMiddleware
from arcade_mcp_server.server import MCPServer
from arcade_core.catalog import ToolCatalog

middleware = compose_middleware([
    ErrorHandlingMiddleware(mask_error_details=False),
    LoggingMiddleware(log_level="INFO"),
    TimingMiddleware(),
])

server = MCPServer(catalog=ToolCatalog(), middleware=[middleware])