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
billingrole 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:
- Tenant isolation is injective.
tenant_identers 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. - Approval lives where the tool lives. A tool declared
requires_approval=Truecannot 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:
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_id → isolation_key |
per-tenant cache / memory / conversation isolation |
What's Next¶
- Multi-Tenancy — the tenancy model in depth
- Approval Gates — elicitation, webhook, and pending approvers
- Authentication & Security — auth providers, guards,
on_authenticate - Multi-User Systems — per-user (not just per-tenant) identity flow
- Deployment — serve it over HTTP with
promptise serve