Metadata-Version: 2.4
Name: intent-api
Version: 0.4.3
Summary: A constraint-driven API framework for Python. One endpoint. Typed intents. Zero boilerplate.
Author: Chris Bora
License-Expression: LicenseRef-Proprietary
Project-URL: Homepage, https://intentapi.dev
Project-URL: Documentation, https://intentapi.dev/docs
Project-URL: Repository, https://github.com/chrisboraai/intent-api
Keywords: api,fastapi,intent,cqrs,framework
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: fastapi>=0.100.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: fastmcp>=2.14.0
Requires-Dist: structlog>=24.0.0
Provides-Extra: otel
Requires-Dist: opentelemetry-sdk>=1.20.0; extra == "otel"
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.20.0; extra == "otel"
Requires-Dist: orjson>=3.9.0; extra == "otel"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: httpx>=0.24.0; extra == "dev"
Requires-Dist: uvicorn>=0.20.0; extra == "dev"
Requires-Dist: jsonschema>=4.0.0; extra == "dev"
Requires-Dist: celery>=5.3.0; extra == "dev"
Requires-Dist: opentelemetry-sdk>=1.20.0; extra == "dev"
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.20.0; extra == "dev"
Requires-Dist: orjson>=3.9.0; extra == "dev"
Dynamic: license-file

# 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

```bash
pip install intent-api
```

## Quickstart

```python
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:

```python
{
    "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:

```python
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:

```json
{
    "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.

```python
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:

```python
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:

```python
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:

```python
# 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](https://modelcontextprotocol.io) 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](https://docs.cursor.com/) at scale.
- **Smaller attack surface** — Tool Description Injection has a [94% success rate per Trail of Bits](https://www.kensai.app); 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

```python
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:

```python
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.

```python
# ✅ 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:

```python
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:

```python
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

```python
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](https://www.structlog.org) 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

```python
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

```bash
pip install 'intent-api[otel]'
```

```python
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

```python
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:

```python
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

```python
# 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)](./LICENSE) for details.
