MCP Governance Quickstart

In this quickstart, you'll add MCP governance to a Python agent and see tool calls appear in your Waxell traces -- in under 5 minutes. You'll connect to an MCP server, make a tool call, and see the resulting span with governance attributes in your dashboard.

By the end, you'll have:

• Automatic tracing of every MCP tool call with span names like tools/call filesystem:read_file
• A working governance configuration with policy checks and PII scanning
• Understanding of how MCP spans nest in your trace tree

Prerequisites

• Python 3.10+
• A Waxell API key -- get one from your Waxell dashboard
• Node.js 18+ -- for running the MCP filesystem server (used in examples)

Step 1: Install

Install waxell-observe with the MCP extra, which pins the mcp package at a compatible version:

pip install waxell-observe[mcp]

This installs waxell-observe along with mcp>=1.25,<2 and wrapt (used for monkey-patching the MCP client).

Already have waxell-observe?

If you already have waxell-observe installed, just add the MCP extra:

pip install "waxell-observe[mcp]"

Step 2: Initialize

Call waxell.init() before importing any MCP client classes. This auto-patches mcp.client.session.ClientSession.call_tool to emit OpenTelemetry spans for every tool call.

import waxell_observe as waxell

waxell.init()

# Import MCP classes AFTER init()
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

Environment variables

You can configure your API key and URL via environment variables instead of passing them to init():

export WAXELL_API_KEY="wax_sk_..."
export WAXELL_API_URL="https://acme.waxell.dev"

Then call waxell.init() without arguments.

Step 3: Connect to an MCP Server

We'll use the official MCP filesystem server as our example -- it's free, requires no API key, and runs locally via npx. It exposes tools for reading, writing, and listing files.

The MCP client library uses an async context manager pattern to establish connections. Here's how to connect using the stdio transport:

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

server_params = StdioServerParameters(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp/mcp-test"],
)

async def main():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Make a tool call -- automatically traced by waxell-observe
            result = await session.call_tool(
                name="read_file",
                arguments={"path": "/tmp/mcp-test/hello.txt"},
            )
            print(result)

asyncio.run(main())

Before running, create a test file:

mkdir -p /tmp/mcp-test
echo "Hello from MCP governance!" > /tmp/mcp-test/hello.txt

Full runnable example

import asyncio
import waxell_observe as waxell

# Initialize BEFORE importing MCP classes
waxell.init()

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main():
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp/mcp-test"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # This tool call is automatically traced
            result = await session.call_tool(
                name="read_file",
                arguments={"path": "/tmp/mcp-test/hello.txt"},
            )

            # Print the result
            for item in result.content:
                print(item.text)


if __name__ == "__main__":
    asyncio.run(main())

When you run this script, Waxell automatically creates an OpenTelemetry span for the call_tool invocation. No manual instrumentation needed.

Step 4: See Your Traces

After running the example, open your Waxell dashboard and navigate to the Traces view. You'll see a span for the MCP tool call with these attributes:

Attribute	Example Value	Description
Span name	tools/call filesystem:read_file	Format: tools/call {server}:{tool}
waxell.mcp.server_name	filesystem	Server name from initialize() response
waxell.mcp.tool_name	read_file	Tool that was called
mcp.method.name	tools/call	MCP protocol method
waxell.mcp.transport	stdio	Transport used to connect
waxell.mcp.arguments	{"path": "/tmp/mcp-test/hello.txt"}	Tool call arguments (truncated to 4KB)
waxell.mcp.result	Hello from MCP governance!	Tool call result (truncated to 4KB)
waxell.mcp.is_error	false	Whether the tool returned an error
waxell.mcp.timeout_ms	30000	Configured timeout in milliseconds

How MCP Spans Nest in Your Trace Tree

When MCP tool calls happen inside an @waxell.observe() decorated agent function, the MCP spans automatically nest as children of the agent span. If your agent also makes LLM calls, you get a complete trace showing reasoning and action:

MCP spans use OpenTelemetry's automatic context propagation. If there's an active span when call_tool is invoked, the MCP span becomes its child. No explicit parent span wiring needed.

Step 5: Add Governance

So far, Waxell has been tracing MCP tool calls passively -- recording what happens without intervening. To add active governance (policy checks, PII scanning, and audit recording), use configure_session():

from waxell_observe.instrumentors.mcp_instrumentor import configure_session

# After session.initialize(), before making tool calls:
configure_session(
    session,
    server_name="filesystem",
    governance_config={
        "agent_name": "file-reader",
        "scan_inputs": True,
        "scan_outputs": True,
        "pii_actions": {
            "ssn": "block",
            "email": "warn",
            "api_key": "block",
        },
    },
)

With governance configured, every tool call now goes through:

1. Policy check -- Queries the Waxell controlplane with the policy ID mcp:filesystem:read_file. If the controlplane returns block, the tool call is stopped before it reaches the MCP server.
2. PII scan on inputs -- Scans tool call arguments for PII and credentials. If a social security number is found and the action is block, the call is stopped.
3. Tool execution -- The actual MCP tool call runs (with configurable timeout).
4. PII scan on outputs -- Scans the tool's response for PII. Output PII findings are logged as warnings (the tool already executed).
5. Audit recording -- All governance decisions, PII findings, and policy results are recorded as span attributes.

Full runnable example with governance

import asyncio
import waxell_observe as waxell

waxell.init()

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from waxell_observe.instrumentors.mcp_instrumentor import configure_session


async def main():
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp/mcp-test"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Configure governance
            configure_session(
                session,
                server_name="filesystem",
                governance_config={
                    "agent_name": "file-reader",
                    "scan_inputs": True,
                    "scan_outputs": True,
                    "pii_actions": {
                        "ssn": "block",
                        "email": "warn",
                        "api_key": "block",
                    },
                },
            )

            # This tool call now goes through policy + PII checks
            result = await session.call_tool(
                name="read_file",
                arguments={"path": "/tmp/mcp-test/hello.txt"},
            )

            for item in result.content:
                print(item.text)


if __name__ == "__main__":
    asyncio.run(main())

Governance is fail-open

If the Waxell controlplane is unreachable or the PII scanner throws an error, the tool call proceeds normally. Governance errors never break your agent. This fail-open behavior is logged so you can monitor governance health.

What governance adds to your traces

With governance configured, your tool call spans gain additional attributes:

Attribute	Example Value	Description
waxell.mcp.governance_checked	true	Governance was active for this call
waxell.mcp.governance_action	allow	Policy decision: allow, block, warn, or throttle
waxell.mcp.policy_id	mcp:filesystem:read_file	Policy ID used for the check
waxell.mcp.pii_detected	false	Whether PII was found in inputs or outputs
waxell.mcp.governance_timestamp	2026-03-05T22:30:00Z	When the governance check ran
waxell.mcp.params_hash	a1b2c3d4e5f67890	SHA256 fingerprint of the arguments

For the full list of governance span attributes, see the Reference page.

Deeper governance features

Each governance feature has its own detailed guide:

• Policy Configuration -- Set up allowlists, blocklists, and conditional rules using the mcp:{server}:{tool} naming convention
• Approval Workflows -- Add human-in-the-loop approval with on_policy_block handlers for sensitive operations
• PII Scanning -- Configure per-type PII actions (block SSNs, warn on emails) and build custom scanners
• Rug Pull Detection -- Detect when tool descriptions or schemas change between sessions

Other Transports

The auto-instrumentor patches ClientSession.call_tool(), which is the same method regardless of transport. Whether you connect via stdio, streamable HTTP, or SSE, the governance wrapper runs identically. Only the session setup differs.

Streamable HTTP

Streamable HTTP is the recommended transport for remote MCP servers. It uses standard HTTP requests with optional streaming for long-running operations.

from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

async with streamable_http_client("https://mcp-server.example.com/mcp") as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()

        # configure_session() works the same as with stdio
        configure_session(
            session,
            server_name="remote-server",
            governance_config={
                "agent_name": "my-agent",
                "scan_inputs": True,
                "scan_outputs": True,
            },
        )

        result = await session.call_tool(
            name="search",
            arguments={"query": "test"},
        )

Transport detection

The auto-instrumentor detects the transport type automatically and records it on the span as waxell.mcp.transport. You don't need to specify the transport in configure_session().

SSE (Server-Sent Events)

Deprecated

SSE transport is deprecated in the MCP specification in favor of streamable HTTP. It still works and is fully supported by the auto-instrumentor, but new integrations should use streamable HTTP instead.

from mcp import ClientSession
from mcp.client.sse import sse_client

async with sse_client("https://mcp-server.example.com/sse") as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()

        # configure_session() works the same as with stdio and HTTP
        configure_session(
            session,
            server_name="sse-server",
            governance_config={
                "agent_name": "my-agent",
                "scan_inputs": True,
                "scan_outputs": True,
            },
        )

        result = await session.call_tool(
            name="search",
            arguments={"query": "test"},
        )

Full runnable example with all three transports

import asyncio
import waxell_observe as waxell

waxell.init()

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.streamable_http import streamable_http_client
from mcp.client.sse import sse_client
from waxell_observe.instrumentors.mcp_instrumentor import configure_session

GOVERNANCE = {
    "agent_name": "multi-transport-agent",
    "scan_inputs": True,
    "scan_outputs": True,
    "pii_actions": {"ssn": "block", "email": "warn"},
}


async def use_stdio():
    """Connect via stdio (local server process)."""
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp/mcp-test"],
    )
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            configure_session(
                session,
                server_name="filesystem",
                governance_config=GOVERNANCE,
            )
            return await session.call_tool(
                name="read_file",
                arguments={"path": "/tmp/mcp-test/hello.txt"},
            )


async def use_http(url: str):
    """Connect via streamable HTTP (remote server)."""
    async with streamable_http_client(url) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            configure_session(
                session,
                server_name="remote",
                governance_config=GOVERNANCE,
            )
            return await session.call_tool(
                name="search",
                arguments={"query": "test"},
            )


async def use_sse(url: str):
    """Connect via SSE (deprecated but supported)."""
    async with sse_client(url) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            configure_session(
                session,
                server_name="sse-server",
                governance_config=GOVERNANCE,
            )
            return await session.call_tool(
                name="search",
                arguments={"query": "test"},
            )


async def main():
    # stdio is always available (uses npx)
    result = await use_stdio()
    for item in result.content:
        print(f"[stdio] {item.text}")

    # Uncomment to test with a remote server:
    # result = await use_http("https://your-mcp-server.example.com/mcp")
    # result = await use_sse("https://your-mcp-server.example.com/sse")


if __name__ == "__main__":
    asyncio.run(main())

Next Steps

Now that you have MCP tool calls appearing in your traces with governance attributes, dive deeper into each governance feature:

• Policy Configuration -- Set up allowlists, blocklists, and conditional rules for tool calls
• Approval Workflows -- Add human-in-the-loop approval for sensitive operations
• PII Scanning -- Configure per-type PII detection and actions
• Rug Pull Detection -- Detect tool description and schema changes between sessions
• Reference -- Complete API reference for all MCP governance attributes and configuration options