View the source code for this module on GitHub: https://github.com/mcp-use/mcp-use/blob/main/libraries/python/mcp_use/client/middleware/middleware.py
- A typed MiddlewareContext to carry request data.
- A Middleware base class with a dispatcher that routes to strongly-typed hooks.
- A MiddlewareManager to build and execute the processing chain.
- A CallbackClientSession that acts as an adapter, creating the initial context without requiring changes to upstream callers like HttpConnector.
CallbackClientSession
from mcp_use.client.middleware.middleware import CallbackClientSession
method init
ParametersSignaturemcp.client.session.ClientSessionrequiredMCP session instancestrrequiredConnector instancemcp_use.client.middleware.middleware.MiddlewareManagerrequiredMiddleware instance
def __init__(client_session: mcp.client.session.ClientSession, connector_id: str, middleware_manager: mcp_use.client.middleware.middleware.MiddlewareManager):
MCPResponseContext
from mcp_use.client.middleware.middleware import MCPResponseContext
method init
ParametersSignaturestrrequiredString valueAnyrequiredParameter valueException | NonerequiredParameter valuefloatrequiredParameter valuedict[str, Any]requiredDictionary of key-value pairsmcp.types.JSONRPCResponse | Nonedefault:"None"Parameter value
def __init__(request_id: str, result: Any, error: Exception | None, duration: float, metadata: dict[str, Any], jsonrpc_response: mcp.types.JSONRPCResponse | None = None):
Middleware
from mcp_use.client.middleware.middleware import Middleware
method call
Main entry point that orchestrates the chainParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
SignatureAny
def __call__(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
method on_call_tool
ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
Signaturemcp.types.CallToolResult
def on_call_tool(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
method on_get_prompt
ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
Signaturemcp.types.GetPromptResult
def on_get_prompt(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
method on_initialize
ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
Signaturemcp.types.InitializeResult
def on_initialize(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
method on_list_prompts
ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
Signaturemcp.types.ListPromptsResult
def on_list_prompts(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
method on_list_resources
ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
Signaturemcp.types.ListResourcesResult
def on_list_resources(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
method on_list_tools
ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
Signaturemcp.types.ListToolsResult
def on_list_tools(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
method on_read_resource
ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
Signaturemcp.types.ReadResourceResult
def on_read_resource(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
method on_request
ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valuemcp_use.client.middleware.middleware.NextFunctionTrequiredParameter value
SignatureAny
def on_request(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
call_next: mcp_use.client.middleware.middleware.NextFunctionT
):
MiddlewareContext
from mcp_use.client.middleware.middleware import MiddlewareContext
method init
ParametersSignaturestrrequiredString valuestrrequiredString valuemcp_use.client.middleware.middleware.TrequiredParameter valuestrrequiredString valuefloatrequiredParameter valuedict[str, Any]default:"<factory>"Dictionary of key-value pairs
def __init__(id: str, method: str, params: mcp_use.client.middleware.middleware.T, connection_id: str, timestamp: float, metadata: dict[str, Any] = <factory>):
MiddlewareManager
from mcp_use.client.middleware.middleware import MiddlewareManager
method process_request
Runs the full middleware chain, captures timing and errors,
and returns a structured MCPResponseContext.ParametersReturnsmcp_use.client.middleware.middleware.MiddlewareContextrequiredParameter valueCallablerequiredParameter value
Signaturemcp_use.client.middleware.middleware.MCPResponseContext
def process_request(
context: mcp_use.client.middleware.middleware.MiddlewareContext,
original_call: Callable
):
NextFunctionT
from mcp_use.client.middleware.middleware import NextFunctionT