OrmAI
Hire Neul Labs

Guide

Quickstart: OrmAI with FastAPI and SQLAlchemy

Stand up a policy-enforced agent database layer in 5 minutes. SQLAlchemy models, FastAPI endpoints, OrmAI policy, and a working tool call you can hand to Claude or GPT.

Dipankar Sarkar · ·Updated April 15, 2026 quickstartfastapisqlalchemypython

This guide takes you from “no project” to “a working FastAPI app with a policy-enforced toolset that an AI agent can call safely.” It’s the path we use to onboard new teams.

What you’ll build

A small order-management API with two models (Customer, Order) and a single endpoint that accepts agent tool calls (/agent/tool). The agent will be able to:

  • Query orders for the current tenant.
  • Look up a customer (with email automatically masked).
  • Create a new order, requiring a reason argument.

In about 80 lines of code.

Prerequisites

  • Python 3.10 or later.
  • uv (recommended) or pip.
  • A few minutes.

1. Install

mkdir ormai-quickstart && cd ormai-quickstart
uv init
uv add 'ormai[sqlalchemy]' fastapi 'uvicorn[standard]' sqlalchemy

2. Define your models

Nothing OrmAI-specific here yet — these are plain SQLAlchemy 2.0 models. The convention OrmAI cares about: every multi-tenant model has a tenant_id column.

# models.py
from sqlalchemy import String, Integer, ForeignKey, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from datetime import datetime

class Base(DeclarativeBase): ...

class Customer(Base):
    __tablename__ = "customers"
    id: Mapped[int] = mapped_column(primary_key=True)
    tenant_id: Mapped[int] = mapped_column(index=True)
    name: Mapped[str] = mapped_column(String(120))
    email: Mapped[str] = mapped_column(String(120))
    phone: Mapped[str | None] = mapped_column(String(40), nullable=True)
    api_secret: Mapped[str | None] = mapped_column(String(64), nullable=True)

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"))
    total_cents: Mapped[int] = mapped_column(Integer)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    customer: Mapped["Customer"] = relationship()

3. Define your policy

This is where OrmAI starts. The policy is a single Python expression that declares everything the agent can and can’t do.

# policy.py
from ormai.utils import PolicyBuilder, DEFAULT_PROD
from models import Customer, Order

policy = (
    PolicyBuilder(DEFAULT_PROD)
    .register_models([Customer, Order])
    # Never expose secrets.
    .deny_fields("*secret*", "*token*", "*password*")
    # Customer email and phone are masked, not denied — the agent
    # can reason about identity without seeing raw values.
    .mask_fields(["customer.email", "customer.phone"])
    # Every read and write is automatically scoped to the request's tenant.
    .tenant_scope("tenant_id")
    # Agent can create orders, but each create needs a reason string,
    # and we cap how many it can create per minute.
    .enable_writes(models=["Order"], require_reason=True)
    .max_writes_per_minute(20)
    .max_rows(200)
    .statement_timeout_ms(3000)
    .build()
)

A handful of decisions buy you a lot of safety:

  • The agent will never see api_secret because *secret* matched the deny rule.
  • Email returned in any tool call is automatically masked (j***@acme.com).
  • Cross-tenant queries are structurally impossible — the policy compiler injects the tenant filter.
  • The agent can create orders, can’t create customers, can’t update or delete anything.
  • Every create is rate-limited and must include a justification string.

4. Mount the toolset

# app.py
from fastapi import FastAPI, Header, HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from ormai.quickstart import mount_sqlalchemy
from ormai.core.context import RunContext
from models import Base
from policy import policy

engine = create_engine("sqlite:///./quickstart.db", echo=False)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(engine, expire_on_commit=False)

toolset = mount_sqlalchemy(
    engine=engine,
    session_factory=SessionLocal,
    policy=policy,
)

app = FastAPI()

@app.post("/agent/tool")
async def call_tool(
    body: dict,
    x_tenant_id: int = Header(..., alias="X-Tenant-Id"),
    x_user_id: str = Header(..., alias="X-User-Id"),
):
    name, args = body.get("name"), body.get("arguments", {})
    if not name:
        raise HTTPException(400, "missing tool name")

    with SessionLocal() as session:
        ctx = RunContext.create(
            tenant_id=x_tenant_id,
            user_id=x_user_id,
            db=session,
        )
        result = await toolset.execute(name, args, ctx)

    if not result.success:
        raise HTTPException(400, {"error": result.error, "policy": result.policy_decision})
    return {"data": result.data, "audit_id": result.audit_id}

5. Run it

uvicorn app:app --reload

Seed a couple of rows manually:

# seed.py — run once
from sqlalchemy.orm import sessionmaker
from app import engine
from models import Customer, Order

S = sessionmaker(engine)
with S() as s:
    c = Customer(tenant_id=1, name="Acme", email="[email protected]", phone="555-1212", api_secret="DO_NOT_LEAK")
    s.add(c); s.flush()
    s.add(Order(tenant_id=1, customer_id=c.id, total_cents=10_000, status="pending"))
    s.add(Order(tenant_id=1, customer_id=c.id, total_cents=20_000, status="paid"))
    s.commit()

6. Call a tool

The agent’s tool call looks like a JSON POST:

curl -s localhost:8000/agent/tool \
  -H 'X-Tenant-Id: 1' \
  -H 'X-User-Id: agent-1' \
  -H 'Content-Type: application/json' \
  -d '{"name":"db.query","arguments":{"model":"Order","where":{"status":"pending"},"limit":10}}'

You get back something like:

{
  "data": [
    {"id": 1, "tenant_id": 1, "customer_id": 1, "total_cents": 10000, "status": "pending", "created_at": "2026-04-15T..."}
  ],
  "audit_id": "01J9..."
}

Now query a customer:

curl -s localhost:8000/agent/tool \
  -H 'X-Tenant-Id: 1' -H 'X-User-Id: agent-1' \
  -H 'Content-Type: application/json' \
  -d '{"name":"db.get","arguments":{"model":"Customer","id":1}}'
{
  "data": {"id": 1, "tenant_id": 1, "name": "Acme", "email": "a***@acme.com", "phone": "55*****12"},
  "audit_id": "01J9..."
}

Notice three things:

  • api_secret is missing — denied by policy.
  • email and phone are masked.
  • A new audit_id was returned. That row is in the audit table.

7. Try to do something the policy forbids

A cross-tenant attempt:

curl -s localhost:8000/agent/tool \
  -H 'X-Tenant-Id: 1' -H 'X-User-Id: agent-1' \
  -d '{"name":"db.query","arguments":{"model":"Order","where":{"tenant_id":2}}}'

The tenant_id=2 filter is silently ignored — the policy injects the authoritative tenant_id=1 from the context. There is no way for the agent to query another tenant’s data.

A delete attempt:

curl -s localhost:8000/agent/tool \
  -d '{"name":"db.delete","arguments":{"model":"Order","id":1,"reason":"oops"}}'
{"detail": {"error": "Tool db.delete not enabled for model Order", "policy": "model_writes_disabled"}}

A write without a reason:

curl -s localhost:8000/agent/tool \
  -d '{"name":"db.create","arguments":{"model":"Order","data":{"customer_id":1,"total_cents":5000}}}'
{"detail": {"error": "reason required for write", "policy": "reason_required"}}

8. Hook it up to a real agent

The /agent/tool endpoint accepts any JSON tool call. Here’s a minimal Anthropic SDK example:

from anthropic import Anthropic
from ormai.utils import openapi_tool_specs

client = Anthropic()
tool_specs = openapi_tool_specs(toolset)  # returns Anthropic-compatible tool defs

resp = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=1024,
    tools=tool_specs,
    messages=[{"role": "user", "content": "What's our pending revenue this week?"}],
)

The same tool_specs accessor works for OpenAI, Vercel AI SDK, and LangChain.

Custom domain tools

When you need something not expressible as db.query — a domain calculation, a join with an external service — register a domain tool:

from ormai.tools import tool

@tool(toolset, name="analytics.weekly_revenue", description="Sum of paid orders this calendar week.")
async def weekly_revenue(ctx: RunContext) -> dict:
    rows = ctx.db.execute(
        "SELECT SUM(total_cents) FROM orders WHERE status='paid' AND created_at > NOW() - INTERVAL '7 days' AND tenant_id = :t",
        {"t": ctx.tenant_id},
    ).fetchall()
    return {"cents": rows[0][0] or 0}

Domain tools share the policy substrate: the same audit, the same tenant injection, the same observability. They just contain real Python instead of being expressible as a generic query.

Where to go next

Got stuck? Email [email protected] — we read every message.

Found a typo or want to suggest a topic? Email [email protected].