Skip to content

Build a Secure Multi-Tenant Agent Platform

This is the capstone guide: you'll build the AI backend for a multi-tenant SaaS product — one server that many customer organizations share, where each tenant's data is provably isolated, sensitive actions require a human's sign-off, and every action lands in a tamper-evident audit log.

If you serve more than one customer from the same deployment, this is the shape you need. It combines four capabilities that most frameworks leave to you to bolt on (and get wrong): tenant isolation, role-based access, server-side approval gates, and tamper-evident audit — enforced at the tool boundary, so governance never depends on trusting the client.

What You'll Build

A billing-operations MCP server for a SaaS company whose customers are organizations (acme, globex, …). Support agents for each customer connect through it to look up invoices and issue refunds. Requirements a real platform has:

  • Tenant isolation — Acme's agent can never see Globex's data, even if both agents share a service client_id.
  • Role-based access — only the billing role may issue refunds.
  • Human-in-the-loop — a refund blocks until a different human approves it (four-eyes), enforced by the server for any client.
  • Fair usage — one tenant can't exhaust another's rate-limit quota.
  • Tamper-evident audit — every call recorded with the acting tenant.

Everything below runs in-process with TestClient (no cloud, no keys) so you can follow along; the Deployment guide covers serving it over HTTP.

Concepts: three identities and two invariants

A multi-tenant agent system carries three layers of identity:

Layer Who Carried by
Tenant the customer organization ClientContext.tenant_id (server) / CallerContext.tenant_id (agent)
Principal the user or service acting client_id (JWT sub) / roles + scopes
Session the conversation session_id

And it rests on two structural invariants — properties enforced by the framework, not conventions you have to remember:

  1. Tenant isolation is injective. tenant_id enters every isolation key (cache, memory, conversations, rate limits, audit). Two tenants with the same user id land in disjoint keyspaces — cross-tenant leakage is structurally impossible, not merely avoided.
  2. Approval lives where the tool lives. A tool declared requires_approval=True cannot execute — for any MCP client — until a human decides. Governance is a property of the tool, not a courtesy of the caller.

Step 1 — A tenant-aware MCP server

Map each credential to a (client_id, roles, tenant_id). Here we use API keys for brevity; in production JWTAuth extracts the tenant from a claim (AuthMiddleware(JWTAuth(...), tenant_claim="org")).

from promptise.mcp.server import MCPServer, AuthMiddleware
from promptise.mcp.server._auth import APIKeyAuth

# require_tenant=True makes tenancy a server-wide INVARIANT: every tool
# authenticates AND must carry a tenant, or the call is denied.
server = MCPServer(name="billing-ops", require_tenant=True)

auth = APIKeyAuth(keys={
    # Two customer orgs deliberately share a service client_id — to prove
    # isolation is by TENANT, not by client_id.
    "sk-acme":     {"client_id": "svc-agent", "roles": ["billing"],  "tenant_id": "acme"},
    "sk-globex":   {"client_id": "svc-agent", "roles": ["billing"],  "tenant_id": "globex"},
    # A human reviewer for Acme (different principal than the caller).
    "sk-approver": {"client_id": "dana",      "roles": ["approver"], "tenant_id": "acme"},
})
server.add_middleware(AuthMiddleware(auth))

Why require_tenant=True

It turns "we always pass a tenant" from a hope into an invariant. Every tool — from decorators, routers, mounts, or OpenAPI import — is forced to authenticate and carries a RequireTenant guard. A token without the tenant claim is denied on every call. See Multi-Tenancy.

Step 2 — Tenant-scoped tools with fair usage

Tools read ctx.client.tenant_id and scope their data access to it. A declared rate_limit is enforced automatically, per tenant — so a noisy tenant can't starve another's quota even when they share a client_id.

from promptise.mcp.server import HasRole, RequestContext

@server.tool(rate_limit="60/min", guards=[HasRole("billing")])
async def get_invoice(invoice_id: str, ctx: RequestContext) -> dict:
    """Look up an invoice — scoped to the caller's tenant."""
    tenant = ctx.client.tenant_id
    return await db.get_invoice(invoice_id, tenant=tenant)

The rate_limit="60/min" bucket key is tenant-qualified and injective, so Acme's traffic and Globex's traffic never share a bucket. (Declared limits are enforced with no extra wiring — see Caching & Performance.)

Step 3 — Human approval on destructive actions

Refunds move money — they need a human. Declare the tool requires_approval and install an ApprovalGateMiddleware. The gate is fail-closed: denied by default on timeout, denied on a handler crash, and it evaluates the tool's guards before it ever bothers a reviewer (so an unauthorized caller can't spam approvers).

from promptise.mcp.server import ApprovalGateMiddleware, PendingApprover

# PendingApprover blocks the call and exposes role-guarded admin tools
# (approvals_list / approvals_decide) for a human reviewer.
approver = PendingApprover(server, approver_role="approver")
server.add_middleware(ApprovalGateMiddleware(approver, timeout=300))

@server.tool(guards=[HasRole("billing")], requires_approval=True)
async def issue_refund(order_id: str, amount: float, ctx: RequestContext) -> dict:
    """Issue a refund — blocks until a human approves."""
    return await billing.refund(order_id, amount, tenant=ctx.client.tenant_id)

An ungated declaration refuses to build

If you declare requires_approval=True but forget the gate, the server raises at build time rather than silently letting the call through. A declared approval that doesn't enforce would be worse than none.

Separation of duties is enforced. approvals_decide rejects an approval whose reviewer equals the original caller — you cannot approve your own refund, even if you also hold approver. Denying your own is always allowed. See Approval Gates for the elicitation and webhook approvers.

Step 4 — Tamper-evident audit, tenant-stamped

Add AuditMiddleware for an HMAC-chained, append-only record. Each entry carries the acting tenant, so per-customer forensics need no external join.

from promptise.mcp.server import AuditMiddleware

server.add_middleware(AuditMiddleware(secret="${AUDIT_SECRET}"))

Approval outcomes flow through the same pipeline: a denial surfaces as a structured APPROVAL_DENIED error (recorded like any error, with the approval request id), and grants proceed to a normal audited call.

Step 5 — The agent side: isolation follows the user

The server enforces isolation for tools. On the agent side, the same tenant_id isolates memory, cache, and conversations — so a support agent serving Acme's user alice can never surface Globex's data, even for a same-named user.

from promptise import CallerContext, build_agent

agent = await build_agent(
    model="openai:gpt-5-mini",
    servers={"billing": {"url": "https://billing.internal/mcp",
                         "transport": "http", "bearer_token": acme_jwt}},
    memory=ChromaProvider(persist_directory="./mem"),
    cache=SemanticCache(),
    conversation_store=SQLiteConversationStore("chat.db"),
)

acme_alice   = CallerContext(user_id="alice", tenant_id="acme")
globex_alice = CallerContext(user_id="alice", tenant_id="globex")  # sees NONE of acme

# Same user_id, different tenants — fully isolated across cache/memory/sessions:
await agent.chat("What did we discuss?", session_id="s1", caller=acme_alice)

One derivation — CallerContext.isolation_key — feeds every per-user surface, so isolation is guaranteed at the scoping layer, not re-implemented per feature. The tenant_id even rides through cross-agent delegation automatically.

Step 6 — Run it end to end

Here's the whole platform in one runnable file, driven through TestClient (no cloud, no key). It proves all four properties: per-tenant rate isolation, the require_tenant invariant, four-eyes approval, and guards-before-approval.

"""Server-side multi-tenancy + approval gates — self-contained, runnable demo.

Demonstrates two enterprise MCP-server features end to end, in-process via
``TestClient`` (no network, no API key needed):

  1. First-class multi-tenancy — ``ClientContext.tenant_id`` from the API-key
     config, tenant-qualified rate-limit buckets, and the ``require_tenant``
     server invariant. Two tenants with the SAME client id are isolated.
  2. Server-side approval gates — ``@server.tool(requires_approval=True)`` +
     ``ApprovalGateMiddleware``. A destructive tool blocks until a human
     approves, guards run before a reviewer is bothered, and a caller cannot
     approve their own request (four-eyes separation of duties).

Run:
    .venv/bin/python examples/mcp/tenancy_and_approval.py
"""

from __future__ import annotations

import asyncio

from promptise.mcp.server import (
    ApprovalGateMiddleware,
    AuthMiddleware,
    HasRole,
    MCPServer,
    PendingApprover,
    TestClient,
)
from promptise.mcp.server._auth import APIKeyAuth
from promptise.mcp.server._context import RequestContext


def build_server() -> tuple[MCPServer, PendingApprover]:
    # require_tenant=True → every tool authenticates AND must carry a tenant.
    server = MCPServer(name="billing", require_tenant=True)

    # API keys map to (client_id, roles, tenant_id). Two tenants deliberately
    # share the client id "svc-agent" to prove isolation is by tenant, not id.
    auth = APIKeyAuth(
        keys={
            "sk-acme": {"client_id": "svc-agent", "roles": ["billing"], "tenant_id": "acme"},
            "sk-globex": {"client_id": "svc-agent", "roles": ["billing"], "tenant_id": "globex"},
            "sk-approver": {"client_id": "dana", "roles": ["approver"], "tenant_id": "acme"},
        }
    )
    server.add_middleware(AuthMiddleware(auth))

    approver = PendingApprover(server, approver_role="approver")
    server.add_middleware(ApprovalGateMiddleware(approver, timeout=10.0))

    @server.tool(rate_limit="1/min", guards=[HasRole("billing")])
    async def get_invoice(invoice_id: str, ctx: RequestContext) -> dict:
        """Look up an invoice (rate-limited per tenant)."""
        return {"invoice": invoice_id, "tenant": ctx.client.tenant_id}

    @server.tool(guards=[HasRole("billing")], requires_approval=True)
    async def issue_refund(order_id: str, amount: float, ctx: RequestContext) -> dict:
        """Issue a refund — requires human approval before it executes."""
        return {"refunded": order_id, "amount": amount, "tenant": ctx.client.tenant_id}

    return server, approver


async def main() -> None:
    server, approver = build_server()
    client = TestClient(server)

    print("=== 1. Tenant isolation in rate limiting ===")
    print("Both tenants use client id 'svc-agent'; declared limit is 1/min.")
    r = await client.call_tool(
        "get_invoice", {"invoice_id": "INV-1"}, headers={"x-api-key": "sk-acme"}
    )
    print(f"  acme   call 1 -> {r[0].text}")
    r = await client.call_tool(
        "get_invoice", {"invoice_id": "INV-1"}, headers={"x-api-key": "sk-acme"}
    )
    print(f"  acme   call 2 -> {'RATE LIMITED' if 'RATE_LIMIT' in r[0].text else r[0].text}")
    r = await client.call_tool(
        "get_invoice", {"invoice_id": "INV-9"}, headers={"x-api-key": "sk-globex"}
    )
    print(f"  globex call 1 -> {r[0].text}  (own bucket — unaffected by acme)")

    print("\n=== 2. require_tenant invariant ===")
    # No x-api-key at all → unauthenticated → denied before the handler.
    r = await client.call_tool("get_invoice", {"invoice_id": "INV-1"})
    print(
        f"  unauthenticated -> {'DENIED' if 'error' in r[0].text.lower() or 'denied' in r[0].text.lower() else r[0].text}"
    )

    print("\n=== 3. Server-side approval gate (four-eyes) ===")
    call = asyncio.create_task(
        client.call_tool(
            "issue_refund",
            {"order_id": "ORD-42", "amount": 250.0},
            headers={"x-api-key": "sk-acme"},
        )
    )
    # Wait for the request to land in the pending store
    for _ in range(200):
        if approver.pending():
            break
        await asyncio.sleep(0.01)
    pending = approver.pending()
    print(
        f"  refund is blocked, awaiting approval: {pending[0]['tool']} "
        f"{pending[0]['arguments']} from tenant={pending[0]['tenant_id']}"
    )

    # dana (a different human with the approver role) releases it
    decide = await client.call_tool(
        "approvals_decide",
        {"request_id": pending[0]["request_id"], "approve": True},
        headers={"x-api-key": "sk-approver"},
    )
    print(f"  approver 'dana' decides -> {decide[0].text}")

    result = await asyncio.wait_for(call, timeout=10)
    print(f"  refund now executes -> {result[0].text}")

    print("\n=== 4. Guards run before approval (no reviewer spam) ===")
    # An 'approver'-only caller lacks the 'billing' role the tool guards on,
    # so the gate rejects it WITHOUT ever creating a pending approval.
    r = await client.call_tool(
        "issue_refund",
        {"order_id": "ORD-99", "amount": 1.0},
        headers={"x-api-key": "sk-approver"},
    )
    print(
        f"  wrong-role caller -> {'DENIED before approval' if 'ACCESS_DENIED' in r[0].text else r[0].text}"
    )
    print(f"  pending queue still empty: {approver.pending() == []}")


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

Run it:

python examples/mcp/tenancy_and_approval.py

You'll see: Acme and Globex get independent rate buckets despite a shared client_id; an unauthenticated call is denied; a refund blocks until a different approver releases it; and a wrong-role caller is rejected before any approval is ever created.

Security architecture summary

Layer Mechanism What it guarantees
Transport JWTAuth / APIKeyAuth every caller is identified
Tenancy tenant_id in every isolation key provable cross-tenant isolation
Authorization RequireTenant, HasRole, HasScope, HasTenant who may call which tool
Human-in-the-loop requires_approval + ApprovalGateMiddleware destructive actions need a human (four-eyes)
Fair usage tenant-qualified rate limits no cross-tenant quota starvation
Audit AuditMiddleware (HMAC-chained, tenant-stamped) tamper-evident record of every action
Agent-side CallerContext.tenant_idisolation_key per-tenant cache / memory / conversation isolation

What's Next