Model Context Protocol (Python)
Add Checkrd policy enforcement to MCP servers and clients in Python.
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 registers tool handlers via the @server.call_tool() decorator. Wrap the decorated function with wrap_call_tool_handler so every tool invocation is policy-evaluated 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(agent_id="mcp-server", api_key="ck_live_...") as client:
server = Server("my-server")
async def my_tool_handler(name: str, arguments: dict) -> list[types.ContentBlock]:
if name == "search":
return [
types.TextContent(type="text", text=f"results for {arguments['q']}")
]
raise ValueError(f"unknown tool: {name}")
# `wrap_call_tool_handler` returns an awaitable suitable for
# `@server.call_tool()`. Pass the user's handler + a Checkrd
# client; policy evaluation runs before delegation.
server.call_tool()(
wrap_call_tool_handler(
my_tool_handler,
client=client,
server_name="my-server",
)
)Where does the policy come from?
Checkrd(api_key=...) fetches your agent's currently-published DSSE-signed bundle from the control plane and installs it before returning. The dashboard is the source of truth — no policy= argument in app code. Updates stream over SSE as soon as you re-publish.
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:
import asyncio
from mcp import StdioServerParameters
from mcp.client.stdio import stdio_client
from checkrd import Checkrd
from checkrd.integrations.mcp import CheckrdClientSession
async def main() -> None:
server_params = StdioServerParameters(command="github-mcp-server")
with Checkrd(agent_id="github-agent", api_key="ck_live_...") as client:
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": "details..."},
)
asyncio.run(main())CheckrdClientSession overrides call_tool, read_resource, and get_prompt. All other methods (initialize, list_tools, sampling_callback, etc.) pass through unchanged.
What gets enforced
| Surface | Synthetic URL | Body |
|---|---|---|
call_tool | https://{server_name}/tools/{tool_name} | {"tool": ..., "arguments": ...} |
read_resource | https://{server_name}/resources/{uri} | {"uri": ...} |
get_prompt | https://{server_name}/prompts/{name} | {"name": ..., "arguments": ...} |
Example policy:
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 onServeris the new low-level path; the older@server.call_tool()decorator pattern still works but is being phased out. Pinmcp >= 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'strace_idautomatically, so cross-server traces stay correlated.