Middleware API Reference

This page is a lookup reference for the server-side governance middleware API. It documents every class, parameter, and behavior in the waxell_observe.mcp module.

How to Read This Reference

Each parameter table uses the following columns:

Column	Meaning
Parameter	The constructor argument or field name
Type	Python type annotation
Default	Default value if not provided
Description	What the parameter controls

---

GovernanceMiddleware

from waxell_observe.mcp import GovernanceMiddleware

Server-side governance middleware for FastMCP servers. Subclasses fastmcp.server.middleware.Middleware and intercepts on_call_tool to apply policy checks, PII scanning, rate limiting, caller identity extraction, and OTel span recording.

Constructor Parameters

Parameter	Type	Default	Description
api_key	str	""	Waxell API key for controlplane policy checks. When empty, the middleware runs in standalone mode (no policy checks).
api_url	str	""	Waxell API URL (e.g., "https://acme.waxell.dev"). Required when api_key is provided.
policy_config	dict | None	None	Server-level governance defaults. Parsed into a MiddlewareConfig dataclass. See MiddlewareConfig for available keys.
pii_scanner	object | None	None	Custom PII scanner implementing the PIIScanner protocol. If None, uses RegexPIIScanner with default regex patterns. See Custom PII Scanner.
tool_configs	dict[str, ToolGovernanceConfig] | None	None	Per-tool governance configuration keyed by tool name. Overrides server-level defaults for specific tools. See ToolGovernanceConfig.

Usage

from fastmcp import FastMCP
from waxell_observe.mcp import GovernanceMiddleware, ToolGovernanceConfig

mcp = FastMCP("MyServer")

mcp.add_middleware(GovernanceMiddleware(
    api_key="wax_sk_...",
    api_url="https://acme.waxell.dev",
    policy_config={
        "default_pii_scan": "strict",
        "default_policy_action": "allow",
        "fail_open": True,
    },
    tool_configs={
        "delete_user": ToolGovernanceConfig(
            approval_required=True,
            pii_scan="strict",
            policy_action="block",
            rate_limit={"max_per_minute": 5},
        ),
    },
))

Auto-Initialization

The middleware automatically calls waxell_observe.init() on the first tool call. This is idempotent -- calling it multiple times has no effect. You do not need to initialize the SDK separately in your server code.

---

ToolGovernanceConfig

from waxell_observe.mcp import ToolGovernanceConfig

Per-tool governance configuration. Applied via the @governance decorator, the constructor tool_configs dict, or derived from server-level defaults.

Fields

Field	Type	Default	Description
approval_required	bool	False	Whether the tool call requires human approval before execution.
pii_scan	str	"standard"	PII scanning mode. See PII Scan Modes.
policy_action	str | None	None	Override the controlplane's policy action for this tool. When None, the controlplane's decision is used. Valid values: "allow", "block", "warn".
rate_limit	dict | None	None	Rate limit configuration for this tool. See Rate Limit Config.

Usage

from waxell_observe.mcp import ToolGovernanceConfig

config = ToolGovernanceConfig(
    approval_required=True,
    pii_scan="strict",
    policy_action="block",
    rate_limit={"max_per_minute": 10, "max_per_hour": 100},
)

PII Scan Modes

Mode	Behavior
"none"	No PII scanning performed
"standard"	Scan with default RegexPIIScanner. All PII findings produce warn action (tool executes).
"strict"	Scan with RegexPIIScanner configured to block on all PII types. PII in inputs blocks the tool call.

PII types scanned: SSN, credit card, email, phone, AWS access key, AWS secret key, API key, private key.

Rate Limit Config

The rate_limit dict supports two keys:

Key	Type	Description
max_per_minute	int	Maximum tool calls allowed per 60-second sliding window
max_per_hour	int	Maximum tool calls allowed per 3600-second sliding window

Both keys are optional. If neither is set, no rate limiting is applied for that tool.

ToolGovernanceConfig(
    rate_limit={"max_per_minute": 10, "max_per_hour": 100},
)

When a rate limit is exceeded, the middleware raises ToolError with the message "Rate limit exceeded for tool '{tool_name}'". Rate limits are enforced locally (no controlplane call needed) using a thread-safe sliding window.

---

MiddlewareConfig

from waxell_observe.mcp import MiddlewareConfig

Server-level governance defaults. Not passed directly to GovernanceMiddleware -- instead, pass a plain dict as policy_config and the constructor parses it into a MiddlewareConfig internally.

Fields

Field	Type	Default	Description
default_pii_scan	str	"standard"	Default PII scanning mode for tools without per-tool config. Valid values: "none", "standard", "strict".
default_policy_action	str	"allow"	Default policy action for tools without per-tool config. Valid values: "allow", "block", "warn".
fail_open	bool	True	If True, governance pipeline errors let tool calls proceed. If False, governance errors propagate as exceptions.

Usage via policy_config Dict

GovernanceMiddleware(
    policy_config={
        "default_pii_scan": "strict",
        "default_policy_action": "allow",
        "fail_open": True,
    },
)

The keys in the policy_config dict map directly to MiddlewareConfig fields. Unrecognized keys are silently ignored.

---

@governance Decorator

from waxell_observe.mcp import governance

Per-tool governance decorator for FastMCP server tools. Attaches a ToolGovernanceConfig as metadata on the tool function. The GovernanceMiddleware reads this config at runtime.

Parameters

Parameter	Type	Default	Description
approval_required	bool	False	Whether the tool call requires human approval before execution.
pii_scan	str	"standard"	PII scanning mode: "none", "standard", or "strict".
policy_action	str | None	None	Override the controlplane's policy action. Valid values: "allow", "block", "warn".
rate_limit	dict | None	None	Rate limit config: {"max_per_minute": N, "max_per_hour": N}.

All parameters are keyword-only.

Usage

from waxell_observe.mcp import governance

@governance(approval_required=True, pii_scan="strict")
@mcp.tool
async def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email to a recipient."""
    return {"status": "sent", "to": to}

Decorator Order

The @governance decorator is order-independent -- it can be applied above or below @mcp.tool:

# Both orderings work identically:

# Option A: @governance above @mcp.tool
@governance(pii_scan="strict")
@mcp.tool
async def tool_a(input: str) -> str: ...

# Option B: @governance below @mcp.tool
@mcp.tool
@governance(pii_scan="strict")
async def tool_b(input: str) -> str: ...

The decorator sets _waxell_governance on both the original function and its wrapper, so the middleware finds it regardless of decorator order.

---

Config Resolution Order

When a tool call arrives, the middleware resolves governance config using the following priority (highest wins):

Priority	Source	How to set
1 (highest)	@governance decorator	@governance(pii_scan="strict") on the tool function
2	Constructor tool_configs	GovernanceMiddleware(tool_configs={"tool_name": ToolGovernanceConfig(...)})
3 (lowest)	Server-level defaults	GovernanceMiddleware(policy_config={"default_pii_scan": "standard"})

If no config is found at any level, the built-in defaults apply: pii_scan="standard", policy_action=None (use controlplane decision), approval_required=False, rate_limit=None.

Resolution Example

mcp.add_middleware(GovernanceMiddleware(
    policy_config={"default_pii_scan": "standard"},  # Priority 3
    tool_configs={
        "send_email": ToolGovernanceConfig(pii_scan="strict"),  # Priority 2
    },
))

@governance(pii_scan="none")  # Priority 1 -- wins for this tool
@mcp.tool
async def send_email(to: str, body: str) -> dict: ...

@mcp.tool
async def read_file(path: str) -> str: ...
# send_email uses pii_scan="none" (decorator, priority 1)
# read_file uses pii_scan="standard" (server default, priority 3)

---

Custom PII Scanner

You can replace the default RegexPIIScanner with a custom scanner by implementing the PIIScanner protocol.

PIIScanner Protocol

from waxell_observe.scanning import PIIScanner

Method	Signature	Returns
scan	scan(self, text: str) -> dict	{"detected": bool, "count": int, "findings": list[dict]}

The protocol is @runtime_checkable, so you can use isinstance() to verify compliance.

Each finding dict in the findings list should contain:

Key	Type	Description
type	str	PII type identifier (e.g., "ssn", "email", "credit_card")
category	str	Category: "pii" or "credential"
action	str	Action to take: "warn", "redact", or "block"

Custom Scanner Example

from waxell_observe.scanning import PIIScanner


class MyPIIScanner:
    """Custom PII scanner using an external service."""

    def scan(self, text: str) -> dict:
        # Call your PII detection service
        findings = my_pii_service.detect(text)
        return {
            "detected": len(findings) > 0,
            "count": len(findings),
            "findings": [
                {"type": f.type, "category": "pii", "action": "warn"}
                for f in findings
            ],
        }


# Pass to GovernanceMiddleware
mcp.add_middleware(GovernanceMiddleware(
    pii_scanner=MyPIIScanner(),
))

Default Scanner: RegexPIIScanner

from waxell_observe.scanning import RegexPIIScanner

The default scanner uses regex patterns to detect PII and credentials. You can customize the action per PII type:

scanner = RegexPIIScanner(actions={
    "ssn": "block",
    "credit_card": "block",
    "email": "warn",
    "phone": "warn",
    "aws_access_key": "block",
    "aws_secret_key": "block",
    "api_key": "block",
    "private_key": "block",
})

mcp.add_middleware(GovernanceMiddleware(pii_scanner=scanner))

When no actions dict is provided, all PII types default to "warn".

---

Governance Pipeline (on_call_tool)

The GovernanceMiddleware.on_call_tool method processes every incoming tool call through a 12-section pipeline. Understanding this pipeline helps you predict how governance decisions are made and what span attributes are set.

Pipeline Sections

Section	Name	What happens	Can block?
1	Extract tool info	Read tool_name and arguments from context.message	No
2	OTel initialization	Auto-call waxell_observe.init() (idempotent)	No
3	Create span	Start a SpanKind.SERVER span with independent root trace (no client propagation)	No
4	Caller identity	Extract client_id, session_id, IP, user-agent from FastMCP context and HTTP headers	No
5	Get tool config	Resolve per-tool config via resolution order	No
6	Rate limit check	Check sliding window rate limits. Raises ToolError if exceeded.	Yes
7	Policy check	Send mcp:{tool_name} to controlplane. Apply action (block/warn/throttle). Skip if no controlplane.	Yes
8	PII scan (inputs)	Scan tool arguments for PII. Block if strict mode finds PII.	Yes
9	Execute tool	Call call_next(context) to run the actual tool	No
10	PII scan (outputs)	Scan tool result for PII. Warn only (tool already executed).	No
11	Audit trail	Record governance_checked, governance_timestamp, params_hash on span	No
12	Return result	Return the tool result to the caller	No

Fail-Open Wrapper

The entire pipeline (sections 4-11) is wrapped in a try/except. If fail_open=True (default) and any governance section raises an unexpected exception, the middleware:

1. Logs a warning
2. Sets policy_skipped=True and governance_checked=False on the span
3. Calls call_next(context) to execute the tool without governance

Intentional blocks (ToolError from rate limits, policy blocks, or PII blocks) always propagate, regardless of fail_open.

---

Server-Side Span Attributes

These attributes are set by GovernanceMiddleware on server-side spans. They use the same waxell.mcp.* namespace as the client-side auto-instrumentor for unified querying.

Core Middleware Attributes

Attribute	Type	Description	Example
waxell.mcp.middleware_active	bool	Whether governance middleware is processing this call	true
waxell.mcp.server_side	bool	Whether this span was emitted by server-side middleware	true
waxell.mcp.method_name	string	The MCP method being called	"tools/call"

Caller Identity Attributes

Attribute	Type	Description	Example
waxell.mcp.caller_client_id	string	MCP client ID from the initialize handshake	"claude-desktop-1.4"
waxell.mcp.caller_session_id	string	MCP session ID for the caller's connection	"sess_x7y8z9"
waxell.mcp.caller_ip	string	Caller IP from HTTP headers (HTTP/SSE transports only)	"192.168.1.42"
waxell.mcp.caller_user_agent	string	Caller user-agent header (HTTP/SSE transports only)	"mcp-client/1.0"

Transport availability

caller_ip and caller_user_agent are only available on HTTP and SSE transports. Stdio transport connections do not have HTTP headers.

Rate Limiting Attributes

Attribute	Type	Description	Example
waxell.mcp.rate_limited	bool	Whether the call was rejected due to rate limiting	true
waxell.mcp.rate_limit_remaining	int	Remaining calls allowed in the current window	7

Governance Decision Attributes

Attribute	Type	Description	Example
waxell.mcp.governance_checked	bool	Whether governance processing completed for this call	true
waxell.mcp.governance_action	string	The final policy action applied	"allow", "block", "warn"
waxell.mcp.governance_timestamp	string	ISO 8601 timestamp of governance processing	"2026-03-05T14:32:01.456Z"
waxell.mcp.policy_skipped	bool	Whether the policy check was skipped (controlplane unreachable, fail-open)	false
waxell.mcp.policy_reason	string	Human-readable policy decision explanation	"Tool blocked by policy"
waxell.mcp.throttle_delay_ms	int	Enforced delay in milliseconds when action is throttle (max 30000)	5000

PII Scan Attributes

Attribute	Type	Description	Example
waxell.mcp.pii_detected	bool	Whether PII was found in inputs or outputs	true
waxell.mcp.pii_count	int	Total PII findings across inputs and outputs	3
waxell.mcp.pii_scan_direction	string	What was scanned: "inputs", "outputs", or "both"	"inputs"
waxell.mcp.pii_action_taken	string	Highest-severity action taken for PII findings	"warn", "block"

Audit Trail Attributes

Attribute	Type	Description	Example
waxell.mcp.params_hash	string	SHA256 hash of tool arguments (truncated to 16 hex chars)	"a3f2b8c1e9d04f67"

Querying Server-Side Spans

Server-side spans can be distinguished from client-side spans using server_side:

# All server-side governance spans
{ span.waxell.mcp.server_side = true }

# Server-side spans that rate-limited a tool
{ span.waxell.mcp.server_side = true && span.waxell.mcp.rate_limited = true }

# Server-side spans where governance was skipped (fail-open)
{ span.waxell.mcp.server_side = true && span.waxell.mcp.policy_skipped = true }

---

Import Reference

All public middleware classes and functions are available from waxell_observe.mcp:

from waxell_observe.mcp import (
    GovernanceMiddleware,    # Main middleware class
    governance,              # Per-tool governance decorator
    ToolGovernanceConfig,    # Per-tool config dataclass
    MiddlewareConfig,        # Server-level config dataclass
)

PII scanning classes are in waxell_observe.scanning:

from waxell_observe.scanning import (
    PIIScanner,              # Protocol for custom scanners
    RegexPIIScanner,         # Default regex-based scanner
    scan_text_for_pii,       # Standalone PII scanning function
)

Install Extras

Extra	Command	What it installs
mcp (client)	pip install waxell-observe[mcp]	mcp>=1.25,<2
mcp-server	pip install waxell-observe[mcp-server]	fastmcp>=3.0,<4, pyyaml>=6.0
Both	pip install waxell-observe[mcp,mcp-server]	All MCP dependencies

mcp-server is not included in [all]

The mcp-server extra is opt-in for server builders. Running pip install waxell-observe[all] does not install FastMCP. You must explicitly install waxell-observe[mcp-server].

---

See Also

• Connecting to Controlplane -- Fail-open semantics and policy check flow
• Span Attributes Reference -- Complete attribute reference including client-side attributes
• API Reference -- Cross-product API reference covering middleware, proxy, and scanning