Build Your Own Channel.

Authenticate

Validate the incoming JWT. Extract the tenant ID, user identity, and scopes. Every request is tied to a principal before any content is inspected.

# JWT validation — extract tenant + identity
token = request.headers.get("Authorization")
claims = verify_jwt(token, jwks_url=JWKS_ENDPOINT)
tenant_id = claims["tenant_id"]
user_id = claims["sub"]

Build Policy Context

Construct the context object the Security API needs: the channel name, the server or source identifier, and the direction (input or output).

# Policy context — tells the engine where this content came from
context = {
    "channel": "mcp",
    "server_id": request.json["server_id"],
    "tool_name": request.json["tool_name"],
    "direction": request.json.get("direction", "input"),
    "tenant_id": tenant_id,
    "user_id": user_id,
}

Call the Intake Pipeline

Send the content payload to the Security API. The engine runs all 3 DLP tiers, evaluates policy rules, and returns a decision — all in one call.

# One call — full DLP + policy + audit
result = security_api.evaluate(
    payload=request.json["payload"],
    context=context,
)
# result.decision: "allow" | "block" | "redact"
# result.findings: [{ "type": "PCI", "entity": "card_number", ... }]

Handle the Decision

Act on the engine's decision. Allow passes content through. Block returns an error to the caller. Redact strips sensitive entities and forwards the cleaned content.

# Enforce — your channel's only branching logic
if result.decision == "allow":
    return forward(request.json["payload"])
elif result.decision == "block":
    return error_response(403, result.findings)
elif result.decision == "redact":
    return forward(result.redacted_payload)

Audit Event Logged Automatically

You do not need to write audit code. The Security API records every evaluation — content hash, policy decision, findings, and tamper-proof audit trail link — automatically. Your channel gets full audit trail coverage for free.