intent-api 0.4.3

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.

Structured Logging + Observability (v0.4.0)

Intent API ships native structlog integration with optional OpenTelemetry export. Every log emitted inside an intent handler automatically carries request_id, intent_id, actor_id, team_id, surface, and phase — zero manual context passing.

Quick start

from fastapi import FastAPI
from intent_api import IntentLogConfig, setup_intent_logging
from intent_api.runtime import IntentRuntime

app = FastAPI()

handle = setup_intent_logging(
    IntentLogConfig(service_name="my-app-api"),
    app=app,                                 # auto-applies IntentContextMiddleware
)

runtime = IntentRuntime.dev_mode()           # uses StructlogLoggerProvider by default
runtime._logger = handle.logger_provider     # or pass via IntentRuntime(logger_provider=...)

That's it. Now every log.info(...) call inside any handler — yours or the framework's — appears with full context:

2026-04-19T14:30:00Z [info] listing_brands  request_id=abc-123 intent_id=Brand.list actor_id=u-9 team_id=t-1 surface=machine
2026-04-19T14:30:00Z [info] intent.success  request_id=abc-123 intent_id=Brand.list elapsed_ms=12 status=success

Two events per intent execution

Every intent execution produces exactly two lifecycle events: intent.start + one of intent.success / intent.denied / intent.failure. Developer logs appear between them. Filter by intent_id in any observability platform → get the full timeline of one intent execution.

Event	Level	When
intent.start	info	Before policy checks
intent.success	info	After successful handler
intent.denied	warning	After IntentRuntimeError (permission/billing/quota) — includes intent_error_code
intent.failure	error	After unexpected exception — includes error class + message

success, denied, and failure also fire the AuditProvider with the same canonical event shape (including the serialized actor). Two streams, identical data.

OpenTelemetry export

pip install 'intent-api[otel]'

handle = setup_intent_logging(
    IntentLogConfig(
        service_name="my-app-api",
        service_version="1.2.3",
        otel_endpoint="https://api.honeycomb.io/v1/logs",
        otel_headers={"x-honeycomb-team": "..."},
    ),
    app=app,
)

# In your FastAPI lifespan:
@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    handle.shutdown()    # flushes batched OTel exports

The OTel structlog processor bypasses stdlib and emits directly to the OTLP LoggerProvider so all context fields arrive as proper typed OTLP attributes — searchable, indexable, not packed into the body.

Celery worker / scheduler

from celery import Celery
from intent_api import IntentLogConfig, configure_celery_logging

celery = Celery("my-app", ...)

handle = configure_celery_logging(
    celery,
    IntentLogConfig(
        service_name="my-app-worker",          # use -worker / -scheduler suffix
        otel_endpoint="...",
    ),
)

configure_celery_logging() registers Celery signals so:

• Deferred tasks rebind the original request_id + intent_id from the triggering HTTP request — full end-to-end correlation across processes.
• Beat / scheduled tasks generate fresh request_id prefixed beat- with surface="system", phase="scheduled" — distinguishable from user-triggered intents.

Celery is NOT a dependency of intent-api — configure_celery_logging() lazy-imports it and raises a clear ImportError if not installed.

stdlib log bridging (added in v0.4.1)

When OTEL_ENDPOINT is configured, every logging.getLogger().error(...) call — yours, third-party libraries, anything that uses Python stdlib logging — is bridged to the same OTel exporter automatically. structlog-originated events still flow through the structlog processor chain (no double-emission, deduped via the _logger-attribute filter the bridge applies).

Legacy log calls in your existing services, SQLAlchemy errors, uvicorn lifecycle messages, etc. all show up in your observability tool with full intent context — without touching them. The need for this was discovered while validating v0.4.0 against a real production app: a stdlib logger.error(...) from a SQLAlchemy session error rendered to console with full intent context (because the structlog formatter merges contextvars at format time) but never reached the OTel collector — the OTel processor lived only inside the structlog pipeline. v0.4.1 closes that gap.

v0.4.2 completes the bridge: contextvars are now merged onto bridged stdlib records as proper OTLP attributes (not just rendered into the message body), so you can filter your collector by request_id, intent_id, actor_id, etc. for stdlib logs the same way you can for structlog logs. Optional opt-in IntentLogConfig(strip_code_attributes=True, ...) strips OTel's auto-injected code.file.path / code.function.name / code.line.number if they bloat your collector.

To opt out:

setup_intent_logging(IntentLogConfig(
    service_name="my-app-api",
    otel_endpoint="...",
    bridge_stdlib_logging=False,   # structlog events only
))

Service naming convention

Process	service_name example	Distinguisher in logs
FastAPI (HTTP + MCP)	seo-brew-api	surface ∈ {standard, admin, guest, machine, mcp}
Celery Worker	seo-brew-worker	phase=deferred
Celery Beat	seo-brew-scheduler	phase=scheduled, surface=system

Migration from v0.3.0

# Before (v0.3.0):
from intent_api.providers import SimpleLoggerProvider
runtime = IntentRuntime(logger_provider=SimpleLoggerProvider(), ...)

# After (v0.4.0):
from intent_api import setup_intent_logging, IntentLogConfig
handle = setup_intent_logging(IntentLogConfig(service_name="my-app-api"), app=app)
runtime = IntentRuntime(logger_provider=handle.logger_provider, ...)

SimpleLoggerProvider still works — no breaking change.

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.