Model Context Protocol — Python

The MCP Python SDK has no app-level middleware chain. Three canonical interception points exist; Checkrd ships an adapter for each.

Install

pip install 'checkrd[mcp]'

Server-side: wrap a call_tool handler

The low-level Server class accepts a handler callback. Wrap it with wrap_call_tool_handler to policy-evaluate every tool invocation before the user's handler runs:

from mcp.server.lowlevel.server import Server
from mcp import types
from checkrd import Checkrd
from checkrd.integrations.mcp import wrap_call_tool_handler

with Checkrd(policy="policy.yaml") as client:
    async def my_tool_handler(ctx, params: types.CallToolRequestParams):
        if params.name == "search":
            return types.CallToolResult(content=[
                types.TextContent(type="text", text=f"results for {params.arguments['q']}")
            ])
        raise ValueError("unknown tool")

    server = Server(
        "my-server",
        on_call_tool=wrap_call_tool_handler(
            my_tool_handler,
            client=client,
            server_name="my-server",
        ),
    )

On deny in enforce mode, the wrapped handler raises CheckrdPolicyDenied. The MCP framework converts this into a JSON-RPC error response back to the client.

Server-side: wrap list_tools

wrap_list_tools_handler is shaped identically. List operations get a "*" target so operators can write rules like deny: { url: "my-server/tools-list/*" } to restrict which agents can enumerate tools.

Client-side: CheckrdClientSession

For agents that call MCP servers, subclass ClientSession:

from mcp import StdioServerParameters
from mcp.client.stdio import stdio_client
from checkrd.integrations.mcp import CheckrdClientSession

async with stdio_client(server_params) as (read, write):
    async with CheckrdClientSession(
        read, write,
        checkrd_client=client,
        server_name="github-mcp",
    ) as session:
        await session.initialize()

        # Each call is policy-evaluated before reaching the server.
        result = await session.call_tool("create_issue", {
            "title": "Bug report",
            "body": "...",
        })

CheckrdClientSession overrides call_tool, read_resource, and get_prompt. All other methods (initialize, list_tools, sampling_callback, etc.) pass through unchanged.

What gets enforced

Example policy:

agent: github-agent
default: deny # allowlist mode

rules:
  - name: allow-issue-create
    allow:
      url: "github-mcp/tools/create_issue"

  - name: allow-issue-list
    allow:
      url: "github-mcp/tools/list_issues"

  # Everything else — including delete_repo, force_push, etc. — is denied.

Streamable HTTP transport

When the server is exposed over Streamable HTTP, transport-level concerns (auth, rate limit, IP allowlist) are best handled with standard ASGI middleware. Use CheckrdASGIMiddleware:

from starlette.applications import Starlette
from starlette.middleware import Middleware
from checkrd.asgi import CheckrdASGIMiddleware

app = Starlette(
    middleware=[Middleware(CheckrdASGIMiddleware, agent_id="github-mcp")],
    routes=[Mount("/", app=mcp.streamable_http_app())],
)

This catches HTTP-layer concerns. Individual tool calls are still policy-evaluated via the handler wrap above — they're at different layers.

Caveats

• The MCP SDK is iterating quickly. The on_call_tool=... constructor-kwarg API on Server is the new low-level path; the older @server.call_tool() decorator pattern still works but is being phased out. Pin mcp >= 1.0.
• No native middleware chain. If you want middleware-style composition (server.use(...)), it's not in the SDK. Wrap each handler at registration instead.
• Trace correlation. The MCP SDK already emits OpenTelemetry spans via mcp.shared._otel. The Checkrd adapter uses the current OTel span's trace_id automatically, so cross-server traces stay correlated.