Multi-tenant isolation for AI agents

How to enforce tenant scoping for every read and write your AI agent makes — and why doing it any other way leaks data.

The single most common production incident in agentic systems is a cross-tenant data leak. This guide explains how to make that incident structurally impossible.

The shape of the problem

Most SaaS data lives in shared tables with a tenant_id column. The application enforces tenancy by adding WHERE tenant_id = ? to every query. That works because every query is written by a developer who has been trained never to forget the clause.

An AI agent has not been trained. It will write queries like:

SELECT * FROM orders WHERE created_at > NOW() - INTERVAL '7d' ORDER BY total DESC LIMIT 10

The query is correct in isolation. It will return rows from every tenant in the database. The user who asked it (“show me my biggest orders this week”) sees data they shouldn’t.

Three things have to be true to prevent this:

The tenant identity must come from a trusted source, not from the agent’s input.
Every query the agent issues — direct, joined, aggregated — must be filtered by the right tenant.
The enforcement must be impossible to bypass via creativity, prompt injection, or honest mistake.

OrmAI’s tenant_scope() policy gives you all three.

Setup

In SQLAlchemy, every multi-tenant model has a tenant_id column:

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    tenant_id: Mapped[int] = mapped_column(index=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))
    # ...

Tell OrmAI:

policy = (
    PolicyBuilder(DEFAULT_PROD)
    .register_models([Customer, Order])
    .tenant_scope("tenant_id")  # column name on every multi-tenant model
    .build()
)

Now every call the agent makes through db.query, db.get, db.aggregate, etc. has the tenant filter injected.

Where the tenant ID comes from

Crucially: not from the agent. From the request context.

@app.post("/agent/tool")
async def call_tool(
    body: dict,
    x_tenant_id: int = Header(..., alias="X-Tenant-Id"),
):
    with SessionLocal() as session:
        ctx = RunContext.create(tenant_id=x_tenant_id, db=session)
        return await toolset.execute(body["name"], body["arguments"], ctx)

The header is set by your authenticated session middleware. The agent never sees it. If the agent’s tool call includes tenant_id in the where clause, the policy compiler ignores it and uses ctx.tenant_id instead.

What this looks like in practice

The agent calls:

{"name": "db.query", "arguments": {"model": "Order", "where": {"status": "pending"}}}

OrmAI compiles this into a SQLAlchemy query:

session.query(Order).filter(Order.status == "pending", Order.tenant_id == 42)

Where 42 came from ctx.tenant_id, not the agent.

Even if the agent tries to forge a tenant ID:

{"name": "db.query", "arguments": {"model": "Order", "where": {"status": "pending", "tenant_id": 7}}}

The compiler discards tenant_id: 7 because it is a scoped column. The audit log records the attempt for review.

Joins and includes

When the agent issues a join through include, OrmAI scopes the joined model too:

{"name": "db.query", "arguments": {"model": "Order", "include": ["customer"]}}

becomes:

session.query(Order).join(Customer, Order.customer_id == Customer.id) \
    .filter(Order.tenant_id == 42, Customer.tenant_id == 42)

If a join target is not declared multi-tenant in your policy (e.g. a Country lookup table), it isn’t scoped — but it also can’t be filtered by tenant fields, so the agent can’t use it as a side channel.

Aggregates

db.aggregate is scoped at the WHERE level before grouping:

{"name": "db.aggregate", "arguments": {
  "model": "Order",
  "group_by": ["status"],
  "aggregations": {"total_cents": "sum"},
}}

becomes:

SELECT status, SUM(total_cents) FROM orders WHERE tenant_id = 42 GROUP BY status

Cross-tenant aggregates are impossible to express through the tool surface.

The “platform admin” exception

There’s always one role that needs to query across tenants — the platform admin or a support engineer. Don’t shoehorn this into the same policy. Instead:

admin_policy = (
    PolicyBuilder(DEFAULT_PROD)
    .register_models([Customer, Order])
    .deny_fields("*secret*", "*token*")
    .max_rows(1000)
    # Note: no tenant_scope.
    .require_principal_role("platform_admin")
    .build()
)
admin_toolset = mount_sqlalchemy(engine=engine, session_factory=SessionLocal, policy=admin_policy)

Mount the admin toolset on a separate endpoint, behind your admin auth. Two policies, two tool surfaces, two audit channels. Don’t try to make one policy do both jobs.

Database row-level security: belt and suspenders

For the highest-stakes data, combine OrmAI’s policy with Postgres row-level security. The policy enforces correctness in normal flow; RLS catches developer mistakes (e.g. someone bypassing OrmAI by hand).

ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON orders
  USING (tenant_id = current_setting('app.tenant_id')::int);

Then in your session opener:

session.execute(text(f"SET LOCAL app.tenant_id = {ctx.tenant_id}"))

OrmAI’s tenant filter and the database’s RLS policy now both enforce isolation. Either one alone is sufficient. Both together is what we recommend for regulated workloads.

Common mistakes

Putting tenant_id in the agent’s prompt

“You are an agent for tenant 42; always filter by tenant_id=42 in your queries.” Don’t do this. The model will get it right most of the time and wrong some of the time. Move tenant identity out of the model’s view entirely.

Allowing the agent to “switch tenants” via a tool

Don’t expose a set_tenant tool. The tenant comes from the authenticated session. If a user has access to two tenants, your authentication layer issues two sessions, and the agent runs separately in each.

Using a global tenant context (thread-local, etc.) without scoping it to the request

Some teams use a thread-local for current_tenant. This works as long as your async framework doesn’t share state across requests. With FastAPI on asyncio, prefer a request-scoped object like RunContext.

Forgetting tenant scoping on join targets

Customer is multi-tenant. So is Order. Easy. But Address belongs to a Customer who belongs to a tenant. You need either: (a) a denormalized tenant_id on Address, or (b) trust that joins always go through Customer.tenant_id. We recommend (a) for write-heavy multi-tenant systems and audit-heavy compliance contexts. The denormalization is cheap and makes RLS straightforward.

Testing tenant isolation

Write a test that fails if the policy is broken:

def test_cross_tenant_isolation():
    # Seed data in two tenants
    seed_orders(tenant_id=1, count=5)
    seed_orders(tenant_id=2, count=5)

    ctx = RunContext.create(tenant_id=1, db=session)
    result = asyncio.run(toolset.execute("db.query", {"model": "Order"}, ctx))

    assert result.success
    assert all(row["tenant_id"] == 1 for row in result.data)
    assert len(result.data) == 5

    # Try to forge tenant 2 in the where clause
    result = asyncio.run(toolset.execute(
        "db.query", {"model": "Order", "where": {"tenant_id": 2}}, ctx,
    ))
    assert result.success
    assert all(row["tenant_id"] == 1 for row in result.data)

Run this in CI. Run it on every policy change.