intent-api 0.3.0

A constraint-driven API framework for Python. One endpoint. Typed intents. Zero boilerplate.

Intent API

One endpoint. Typed intents. Zero boilerplate.

Intent API is a constraint-driven API framework for Python. Instead of writing dozens of REST endpoints, you declare intents — structured requests that describe what the caller wants to do. The framework dispatches them to the right handler automatically.

POST /api/intent
{
  "model": "Todo",
  "action": "create",
  "payload": { "title": "Ship Intent API", "done": false },
  "context": { "type": "user", "team_id": "abc-123" }
}

Install

pip install intent-api

Quickstart

from fastapi import FastAPI
from intent_api import IntentRouter, IntentService, MutationResponse

app = FastAPI()

# 1. Define a service
class TodoService(IntentService):
    async def create(self, *, db, user, context, payload):
        # Your create logic here
        return MutationResponse(success=True, id="1", message="Todo created")

    async def list(self, *, db, user, context, skip, limit):
        return {"items": [], "total": 0}

# 2. Create router and register services
router = IntentRouter()
router.register("Todo", TodoService())

# 3. Build and include the FastAPI router
app.include_router(router.build(
    get_user=my_auth_dependency,  # Your auth function
    get_db=my_db_dependency,      # Your DB session function
))

That's it. One endpoint handles all CRUD + custom commands for every model.

Core Concepts

IntentRequest

Every API call is an IntentRequest:

Field	Type	Description
model	str	Target resource (e.g., "Todo", "User")
action	str	"create", "read", "update", "delete", "list", "custom"
id	any?	Resource ID for read/update/delete
payload	dict?	Data for create/update/custom
command	str?	Named command when action is "custom"
context	IntentContext?	Caller context (role, team, org)
skip	int?	Pagination offset (default: 0)
limit	int?	Pagination limit (default: 10)

IntentContext

Context tells the backend who is calling and what scope they're in:

{
    "type": "member",          # Role/surface type
    "team_id": "uuid-123",     # Team scope
    "organization_id": null,   # Org scope (optional)
}

IntentService

Subclass IntentService for each model. Override the methods you need:

class UserService(IntentService):
    async def create(self, *, db, user, context, payload):
        ...

    async def read(self, *, db, user, id, context):
        ...

    async def update(self, *, db, user, id, context, payload):
        ...

    async def delete(self, *, db, user, id, context):
        ...

    async def list(self, *, db, user, context, skip, limit):
        ...

    async def custom_action(self, *, db, user, context, command, id, payload):
        if command == "archive":
            return await self._archive(db=db, id=id)
        raise ValueError(f"Unknown command: {command}")

Custom Commands

Custom commands let you go beyond CRUD:

{
    "model": "Report",
    "action": "custom",
    "command": "export_csv",
    "payload": { "date_from": "2024-01-01", "date_to": "2024-12-31" }
}

@custom_action decorator (recommended)

Replaces manual if/elif dispatch. Each decorated method is auto-registered, auto-discoverable via the MCP surface, and gets its own click-to-source line in the debug registry.

from intent_api import IntentService, custom_action
from pydantic import BaseModel

class GenerateBlogPostPayload(BaseModel):
    keywords: list[str]
    tone: str = "professional"

class BlogPostService(IntentService):
    async def create(self, *, db, user, context, payload):
        ...

    @custom_action(schema=GenerateBlogPostPayload)
    async def generate(self, *, db, user, context, id, payload):
        # Dispatched when {model: "BlogPost", action: "custom", command: "generate"}
        return {"generated": True}

    @custom_action(name="export_mdx")
    async def export(self, *, db, user, context, id, payload):
        # Dispatched as command "export_mdx" (not "export")
        return {"exported": True}

The decorator is fully backward compatible — services using the legacy custom_action() override continue to work. You may NOT mix both patterns on the same class (raises TypeError at class definition time).

Multiple Surfaces

Intent API supports multiple access levels from the same registry:

router = IntentRouter()
router.register("Todo", TodoService())
router.register("User", UserService())

# Standard: authenticated users
app.include_router(router.build(
    get_user=my_auth_dependency,
    get_db=get_db,
))

# Admin: requires additional authorization
app.include_router(router.build_admin(
    get_user=my_auth_dependency,
    get_db=get_db,
    authorize=lambda user, ctx: user.email.endswith("@mycompany.com"),
))

# Guest: unauthenticated, restricted actions
app.include_router(router.build_guest(
    get_db=get_db,
))

Guest Access Control

Mark services as guest-accessible:

class PublicFeedService(IntentService):
    is_guest_allowed = True
    allowed_guest_actions = ["list", "read"]

    async def list(self, *, db, user, context, skip, limit):
        # user will be None for guest requests
        return {"items": [...], "total": 10}

Auth Integration

Intent API is auth-agnostic. Provide your own get_user dependency:

# Example with Clerk
async def get_user(credentials, context, db):
    clerk_user_id = verify_clerk_token(credentials.credentials)
    return db.query(User).filter(User.clerk_id == clerk_user_id).first()

# Example with Auth0
async def get_user(credentials, context, db):
    payload = decode_auth0_token(credentials.credentials)
    return db.query(User).filter(User.auth0_id == payload["sub"]).first()

# Wire it up
app.include_router(router.build(get_user=get_user, get_db=get_db))

MCP Surface

Intent API ships with a built-in Model Context Protocol server. One line exposes your entire registered service catalog to MCP-compatible AI hosts (Claude Desktop, Cursor, Claude Code, ChatGPT, etc.) — with OAuth 2.1 auth, dynamic discovery, and zero per-service configuration.

Why one tool, not many?

Intent API exposes a single MCP tool named intent_api, plus three discovery Resources. This is a deliberate design choice:

• Avoids context bloat — token cost stays under ~1.5k regardless of model count. Compare to per-tool MCP servers that hit the 40-tool Cursor cap and 128-tool Copilot cap at scale.
• Smaller attack surface — Tool Description Injection has a 94% success rate per Trail of Bits; one tool means one description, one schema, one audit chokepoint.
• Dynamic discovery — The intent://schema, intent://models, and intent://models/{model} Resources are read-only and only cost tokens when the LLM explicitly fetches them.

Setup

from fastapi import FastAPI
from intent_api import IntentRouter, IntentService
from intent_api.mcp_auth import clerk_mcp_auth

router = IntentRouter()
router.register("Brand", BrandService())
router.register("BlogPost", BlogPostService())
router.register("Internal", InternalService(), expose_mcp=False)  # hidden from MCP

# Build the MCP app once
mcp_app = router.build_mcp(
    get_user=clerk_mcp_auth(
        secret_key=settings.CLERK_SECRET_KEY,
        authorization_server=settings.CLERK_FRONTEND_API,
    ),
    get_db=get_db,
    resource_metadata_url="https://api.example.com/.well-known/oauth-protected-resource",
)

# Wire it into FastAPI — lifespan is REQUIRED
app = FastAPI(lifespan=mcp_app.lifespan)
app.mount("/mcp", mcp_app)

# OAuth metadata MUST be at app root (RFC 9728)
app.include_router(router.build_mcp_well_known(
    resource_url="https://api.example.com/mcp",
    authorization_server="https://example.clerk.accounts.dev",
))

Auth options

Clerk OAuth 2.1 — install clerk-backend-api and use clerk_mcp_auth().

Custom JWT / API key — bring your own bearer-token verifier:

from intent_api.mcp_auth import bearer_token_auth

def verify(token: str):
    return decode_my_jwt(token)  # raise on invalid

async def resolve(decoded, db):
    return db.query(User).filter(User.sub == decoded["sub"]).first()

get_user = bearer_token_auth(verify_fn=verify, resolve_user=resolve)

API key (machine surface reuse) — your existing build_machine get_user works as-is, provided it has the (credentials, context, db) signature.

mount vs include_router

The four REST surfaces use app.include_router(...). The MCP surface uses app.mount("/mcp", ...). The difference is real and intentional: MCP speaks JSON-RPC 2.0 over Streamable HTTP — it is a mounted protocol, not a REST router. Don't try to wrap it in a Router.

get_user contract

get_user functions used with build_mcp() MUST accept exactly (credentials, context, db) as keyword args. Functions that use FastAPI Depends() parameters in their signature are not compatible — refactor or wrap them.

# ✅ Works with MCP
async def get_user(credentials, context, db):
    ...

# ❌ Does NOT work with MCP — Depends() in signature
async def get_user(creds=Depends(security), db=Depends(get_db)):
    ...

Security: context is server-derived

The MCP intent_api tool input schema deliberately omits the context field. IntentContext is built server-side from the authenticated session inside the tool handler. The LLM cannot pass, forge, or influence team_id, role, or any context field through the MCP surface — closing a multi-tenancy bypass that would otherwise be trivial to exploit.

Resources

URI	Returns
intent://schema	Full registry document — every MCP-visible model with actions, commands, and payload schemas
intent://models	Lightweight list of all MCP-visible models with one-line descriptions
intent://models/{model}	Full schema for a single model: actions, commands, per-command payload schemas

All Resources are JSON Schema 2020-12 compliant.

Intent Runtime — Governance, Billing, Quota, Audit (v0.3.0)

Intent API ships with an optional runtime layer that enforces permission, billing, quota, audit, and logging policies before any handler runs. Every execution path — HTTP, MCP, Celery, Beat, internal — flows through the same pipeline. Handlers contain pure business logic only.

Why this exists

As AI generates more of your code, you need a layer that:

• Catches hallucinated permissions at startup (PolicyRegistry validates every string)
• Centralizes audit — one chokepoint, every dispatch logged
• Pre-empts billing leaks — quota consumed at request time, rolled back on failure
• Stops policy drift — handlers cannot bypass the pipeline (it's framework-owned)

Quick start with dev_mode()

Get from zero to fully governed in one call:

from intent_api import IntentRouter, IntentService, intent
from intent_api.runtime import IntentRuntime

class CampaignService(IntentService):
    """Campaigns."""
    __default_intent__ = intent.defaults(
        permission_prefix="campaign",
        audit=True,
    )

    async def list(self, *, db, user, context, skip, limit):
        return {"items": [...], "total": 12}

    async def create(self, *, db, user, context, payload):
        return {"created": True}

runtime = IntentRuntime.dev_mode(
    role_permissions={
        "admin": ["*"],
        "member": ["campaign:list", "campaign:read"],
    },
)

router = IntentRouter(runtime=runtime)
router.register("Campaign", CampaignService())

That's it. member users can list and read campaigns. Tries to create → 403 PERMISSION_DENIED. admin can do anything. Every dispatch is audited and logged.

The pipeline

Request side (HTTP/MCP/internal):
  1. Resolve actor (from get_user)
  2. Resolve IntentPolicySpec (from @intent or class default)
  3. Permission check (if spec.permission)
  4. Billing check (if spec.feature)
  5. Quota check + consume (if spec.quota)
  6a. IMMEDIATE → execute handler → return result
  6b. DEFERRED → dispatch to TaskProvider → return task ref
                   (rollback quota on dispatch failure)
  7. Audit log (if spec.audit)
  8. Logger emit (always)

Worker side (deferred handlers, via runtime.execute_deferred):
  1. Deserialize actor
  2. Resolve handler from registry
  3. Execute handler
  4. Audit + log
  (no permission/billing/quota — already enforced at request time)

@intent — declarative policy on methods

Three usage patterns:

from intent_api import intent, custom_action

# 1. Raw decorator with explicit fields
@intent(permission="campaign:export", feature="exports", quota="exports_per_month", audit=True, execution="deferred")
@custom_action()
async def export(self, *, db, user, context, id, payload): ...

# 2. Preset-based
@intent.preset("standard_deferred_paid_write",
               permission="campaign:export",
               feature="exports",
               quota="exports_per_month")
async def export(self, ...): ...

# 3. Class-level default — derives policies for undecorated CRUD methods
class CampaignService(IntentService):
    __default_intent__ = intent.defaults(
        permission_prefix="campaign",
        write_preset="standard_write",
        read_preset="standard_read",
        audit=True,
    )
    # create → permission="campaign:create", preset=standard_write, audit=True
    # read   → permission="campaign:read",   preset=standard_read,  audit=True
    # ...

Built-in presets: standard_read, standard_write, standard_deferred_paid_write, admin_write, machine_task, system_job. Define custom presets with intent.define_preset().

PolicyRegistry — hallucination protection

from intent_api import PolicyRegistry

policy = PolicyRegistry(
    permissions=["campaign:create", "campaign:read", "campaign:export"],
    features=["exports", "ai_generation"],
    quotas=["exports_per_month", "ai_generations"],
)

runtime = IntentRuntime(policy=policy, ..., strict=True)

At startup, every resolved IntentPolicySpec is cross-referenced against the registry. AI typos like "campagin:create" are caught before any request runs. In strict=True mode, validation errors raise IntentValidationError and prevent app startup.

Provider interfaces

Seven Protocols define HOW policy is evaluated. Plug in your own implementations or use the SimpleProviders:

Provider	Purpose	Production replacement
PermissionProvider	Role/permission lookup	DB-backed roles, Casbin, OPA
BillingProvider	Plan/feature lookup	Stripe metadata, internal billing service
QuotaProvider	Consume + rollback quotas	Redis with INCR/DECR
TaskProvider	Dispatch deferred handlers	Celery, Dramatiq, RQ
AuditProvider	Durable audit records	Postgres audit_log table
LoggerProvider	Observability events	OpenTelemetry, Datadog, StatsD
ActorSerializer	(De)serialize actor across processes	Custom — re-fetch User from DB by id

Inspector — full policy visibility

When debug=True, four endpoints expose the entire governance layer:

GET /api/intent-debug/governance   — every intent + resolved policy + source
GET /api/intent-debug/providers    — configured/missing providers + warnings
GET /api/intent-debug/validation   — startup validation issues
GET /api/intent-debug/registry     — extended with policy per method

Migration from v0.2.0

The runtime is opt-in. v0.2.0 apps work unchanged. To adopt incrementally:

1. Add a runtime — IntentRouter(runtime=IntentRuntime.dev_mode(...)). No behavior change yet — services without @intent or __default_intent__ are still ungoverned (warning logged).
2. Add __default_intent__ to one service — that service is now governed.
3. Add @intent to specific methods for fine-grained control or paid features.
4. Add a PolicyRegistry with explicit allowlists when you're ready to lock down.
5. Switch to strict=True to make ungoverned writes a startup error.

Why Intent API?

Traditional REST	Intent API
GET /users, POST /users, GET /users/:id, PUT /users/:id, DELETE /users/:id, POST /users/:id/archive, GET /reports/export ...	POST /api/intent
40+ endpoints to maintain	1 endpoint, N services
Auth middleware on every route	Auth once at the intent surface
No standard request format	Every request is an IntentRequest
Hard to audit ("which endpoints access sensitive data?")	One place to log all access

License

Copyright 2026 Chris Bora. All rights reserved.

Free for any use, including commercial. The only restriction is you can't use it to build a competing framework or hosted service. See the INTENT API LICENSE (IACL) for details.