Step 1: Put the tenant in the user's identity
The login service turns a signed-in user into a CallerContext. The token inside it holds the tenant claim and the plan's scopes.
support_desk/identity.pypython
def caller_for(self, tenant_id: str, user_id: str) -> CallerContext: """The identity one agent invocation runs as.""" user = USERS[(tenant_id, user_id)] return CallerContext( user_id=user.user_id, tenant_id=user.tenant_id, bearer_token=self.token_for(tenant_id, user_id), metadata={"name": user.name}, )
Both seed tenants have a user called alice. A user id alone must never be an isolation key.
Step 2: Build a backend that trusts only the token
The ticketing backend is an MCP server built with the Promptise MCP server SDK. One server serves every tenant.
support_desk/backend.pypython
server = MCPServer( "support-desk", instructions="Tickets and orders for the signed-in end user's company.", require_tenant=True, hide_unauthorized_tools=True,)# Order matters: identify the caller, record the call, apply the quota,# then hold gated tools for a decision.server.add_middleware( AuthMiddleware( JWTAuth(secret=secret or signing_secret(), audience=AUDIENCE, issuer=ISSUER), tenant_claim=TENANT_CLAIM, on_authenticate=attach_plan, ))server.add_middleware(audit)server.add_middleware(RateLimitMiddleware(PlanLimiter(), key_func=tenant_bucket))server.add_middleware(ApprovalGateMiddleware(refunds, timeout=30))
AuthMiddleware copies the org claim to ctx.client.tenant_id. With require_tenant=True, a token without a tenant is refused on every tool. HasScope guards on each tool enforce the plan. hide_unauthorized_tools=True lists each tenant only the tools its plan allows.
tenant_bucket returns plan|tenant as the rate-limit key. So every user of a company draws from one quota. When Globex runs out, it gets RATE_LIMIT_EXCEEDED. Acme keeps working.
Each tool reads the tenant from the verified request.
support_desk/backend.pypython
ticket = caller_records(ctx).ticket(ticket_id.strip().upper(), requester=ctx.client.client_id)if ticket is None: # Same answer whether the ticket belongs to another customer, another # tenant, or nobody: the reply must not confirm that an ID exists. raise ToolError(f"Ticket {ticket_id} was not found for this customer.")return ticket
A prompt injection has nothing to change here. The model can ask for ACME-1041 while serving Globex. The lookup runs inside Globex's records and finds nothing.
Step 4: Hold refunds for review
A refund moves money, so the backend gates it. The tool is declared with requires_approval=True:
support_desk/backend.pypython
@server.tool(destructive_hint=True, guards=[HasScope(REFUNDS_WRITE)], requires_approval=True)
ApprovalGateMiddleware calls RefundReview before the tool runs. The request carries the caller's tenant and user. Refunds up to the plan's limit are approved by policy. Larger ones go to the support leads' queue. The agent tells the customer what happens next. In production, swap in a PendingApprover or a WebhookApprovalHandler.
Step 5: Give each tenant its own help center
Each tenant gets its own RAGPipeline and vector store. There is no shared index, so there is no filter to forget.
support_desk/help_center.pypython
@toolasync def search_help_center(query: str, caller: CallerContext | None) -> str:
A parameter annotated CallerContext is filled by Promptise. It is not part of the tool's schema, so the model can't see or set it. The tool searches the index of caller.tenant_id.
Step 6: Write the instructions as code
The rules are the same for every tenant. The plan is not. A context provider adds the signed-in company to the prompt on every request.
support_desk/instructions.pypython
@prompt()@context(callable_context(signed_in_tenant, header="## Signed-in customer"))async def support_instructions() -> str: """Help the signed-in customer of this company with their question."""
signed_in_tenant reads the current caller with get_current_caller(). It lists what the company's plan includes and what it does not. The rules live in SUPPORT_BLOCKS (an Identity and a Rules block).
Step 7: Build the agent once
One agent serves every tenant.
agent.pypython
extra_tools=[help_center_tool(help_centers), preference_tool(memory)],# PER_USER + a caller with tenant_id: searched as "acme::alice" before each# answer. Only stated preferences are stored (remember_preference), not# every exchange: the chat history already lives in the conversation store.memory=memory,memory_max_results=3,# Sessions are owned by tenant::user: another tenant's user gets SessionAccessDenied.conversation_store=conversations or InMemoryConversationStore(),conversation_max_messages=20,guardrails=payment_data_guardrail(),observe=tracing(trace_dir),
The guardrail redacts card numbers and secrets in the user's message. The model, the session store and memory only see [CREDIT_CARD]. Observability records every model call and tool call with the chat session. The default ReAct pattern fits: support questions are short tool loops.
Step 8: Answer in the user's session
agent.pypython
answer = await agent.chat(question, session_id=session_id, caller=caller)
That one call checks session ownership. It loads the history. It scopes memory, renders the tenant's prompt and sends the user's token to the backend.