Complete guide to adding CoreSDK to a FastAPI application — auth, RBAC, ABAC, multi-tenancy, tracing, and RFC 9457 errors.

FastAPI Integration

This guide walks through adding every CoreSDK feature to a FastAPI application. A complete, runnable version lives at python/fastapi-app/.

Install

pip install "coresdk[fastapi]" uvicorn

1. Initialize the SDK

from coresdk import CoreSDKClient, SDKConfig

_sdk = CoreSDKClient(SDKConfig(
    sidecar_addr="[::1]:50051",
    tenant_id="acme-corp",
    service_name="product-api",
    fail_mode="open",    # "closed" for production
))

SDKConfig reads all values from coresdk.toml. Every key can be overridden with CORESDK_* env vars.

2. Add middleware

from fastapi import FastAPI
from coresdk.middleware.fastapi import CoreSDKMiddleware

class SDKAdapter:
    """Thin adapter expected by CoreSDKMiddleware."""
    config = _sdk.config
    def authorize_sync(self, token, **kw):
        return _sdk.validate_token(token, **kw)

app = FastAPI()
app.add_middleware(
    CoreSDKMiddleware,
    sdk=SDKAdapter(),
    exclude_paths=["/healthz", "/docs", "/openapi.json", "/redoc"],
)

CoreSDKMiddleware intercepts every request, calls the sidecar to validate the JWT, and stores the decoded claims on request.state.coresdk_user.

Requests to exclude_paths bypass auth entirely.

3. Read claims

from fastapi import Request

def current_user(request: Request) -> dict:
    user = getattr(request.state, "coresdk_user", None)
    if isinstance(user, dict):
        return user
    # fail-open: sidecar allowed request but no real JWT
    return {"sub": "anonymous", "roles": [], "tenant_id": "acme-corp"}

Claims shape from a real JWT:

{
  "sub": "alice",
  "tenant_id": "acme-corp",
  "roles": ["viewer", "editor"],
  "email": "alice@acme.com"
}

4. Role-based access control

from fastapi import Depends, HTTPException

def require_role(role: str):
    def _check(request: Request):
        user = current_user(request)
        if role not in user.get("roles", []):
            raise HTTPException(status_code=403, detail={
                "type":   "https://coresdk.io/errors/forbidden",
                "title":  "Forbidden",
                "status": 403,
                "detail": f"Role '{role}' required. Your roles: {user.get('roles', [])}",
            })
        return user
    return Depends(_check)

@app.post("/products")
@trace(intent="create-product")
async def create_product(body: dict, _=require_role("editor")):
    ...

@app.delete("/products/{id}")
@trace(intent="delete-product")
async def delete_product(id: int, _=require_role("admin")):
    ...

5. Multi-tenant isolation

def get_tenant(request: Request) -> str:
    return current_user(request).get("tenant_id") or "default"

_db: dict[str, list] = {
    "acme-corp": [...],
    "globex": [...],
}

@app.get("/products")
async def list_products(request: Request):
    tenant = get_tenant(request)
    return {"tenant": tenant, "products": _db.get(tenant, [])}

The tenant comes from the JWT claim tenant_id — never from a query param or URL.

6. Rego policy evaluation (ABAC)

Attribute-based access control: pass resource attributes to the policy engine.

@app.get("/documents/{doc_id}")
@trace(intent="get-document-abac")
async def get_document(doc_id: str, request: Request):
    tenant = get_tenant(request)
    user   = current_user(request)
    doc    = get_doc(doc_id, tenant)   # raises 404 if missing

    allowed = _sdk.evaluate_policy("data.authz.allow", {
        "tenant_id":      tenant,
        "subject":        user.get("sub"),
        "action":         "read",
        "resource":       f"documents/{doc_id}",
        "resource_owner": doc["owner"],     # who owns this document
        "resource_tenant": tenant,
        "context":        {"roles": user.get("roles", [])},
    })
    if not allowed:
        raise HTTPException(status_code=403, detail={
            "type":  "https://coresdk.io/errors/forbidden",
            "title": "Forbidden", "status": 403,
            "detail": f"Access denied to document {doc_id}",
        })
    return doc

The matching Rego policy (load via CORESDK_POLICY_DIR=./policy):

# policy/authz.rego
package authz
import rego.v1

default allow := false

allow if { "admin" in input.context.roles }
allow if { "editor" in input.context.roles; input.action in {"read","write"} }
allow if { "viewer" in input.context.roles; input.action == "read" }
allow if { input.resource_owner == input.subject }   # owner always can read their own

7. PII-safe tracing

from coresdk.tracing.decorator import trace

@app.get("/products")
@trace(intent="list-products")    # creates OTel span named "list-products"
async def list_products(request: Request):
    ...

@trace creates an OTel span. The sidecar's PIIMaskingSpanProcessor strips emails, tokens, and API keys from span attributes before export — nothing sensitive reaches your collector.

8. RFC 9457 error responses

from fastapi.responses import JSONResponse
from coresdk.errors._rfc9457 import ProblemDetailError

@app.exception_handler(ProblemDetailError)
async def problem_detail_handler(request: Request, exc: ProblemDetailError):
    return JSONResponse(
        status_code=exc.status,
        content=exc.to_dict(),
        media_type="application/problem+json",
    )

Every 401/403/404 now returns:

{
  "type": "https://coresdk.io/errors/forbidden",
  "title": "Forbidden",
  "status": 403,
  "detail": "Role 'admin' required"
}

Complete minimal app

import os, tomllib
from pathlib import Path
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from coresdk import CoreSDKClient, SDKConfig
from coresdk.middleware.fastapi import CoreSDKMiddleware
from coresdk.tracing.decorator import trace
from coresdk.errors._rfc9457 import ProblemDetailError

_sdk = CoreSDKClient(SDKConfig(
    sidecar_addr="[::1]:50051",
    tenant_id="acme-corp",
    fail_mode="open",
))

class SDKAdapter:
    config = _sdk.config
    def authorize_sync(self, token, **kw):
        return _sdk.validate_token(token, **kw)

app = FastAPI()
app.add_middleware(CoreSDKMiddleware, sdk=SDKAdapter(),
                   exclude_paths=["/healthz", "/docs", "/openapi.json"])

def current_user(request: Request) -> dict:
    return getattr(request.state, "coresdk_user", None) or \
           {"sub": "anonymous", "roles": [], "tenant_id": "acme-corp"}

def require_role(role: str):
    def _check(request: Request):
        user = current_user(request)
        if role not in user.get("roles", []):
            raise HTTPException(status_code=403, detail={
                "type": "https://coresdk.io/errors/forbidden",
                "title": "Forbidden", "status": 403,
                "detail": f"Role '{role}' required",
            })
        return user
    return Depends(_check)

@app.get("/healthz")
async def healthz():
    return {"status": "ok"}

@app.get("/me")
@trace(intent="get-current-user")
async def me(request: Request):
    return current_user(request)

@app.get("/products")
@trace(intent="list-products")
async def list_products(request: Request):
    tenant = current_user(request).get("tenant_id", "acme-corp")
    return {"tenant": tenant, "products": []}

@app.post("/products")
@trace(intent="create-product")
async def create_product(body: dict, _=require_role("editor")):
    return {"created": body}

@app.exception_handler(ProblemDetailError)
async def problem_detail_handler(request: Request, exc: ProblemDetailError):
    return JSONResponse(status_code=exc.status, content=exc.to_dict(),
                        media_type="application/problem+json")

Run it:

uvicorn main:app --reload
curl http://localhost:8000/healthz
curl -H "Authorization: Bearer alice-token" http://localhost:8000/me

Testing

# test_main.py
import pytest
from fastapi.testclient import TestClient
from unittest.mock import MagicMock
from coresdk._types import AuthDecision
from main import app, SDKAdapter, _sdk

@pytest.fixture
def client():
    return TestClient(app, raise_server_exceptions=False)

def mock_auth(claims: dict):
    """Patch the SDK to return a valid decision with given claims."""
    decision = AuthDecision(allowed=True, claims=claims)
    _sdk._client = MagicMock()
    _sdk._client.validate_token.return_value = decision

def test_healthz(client):
    assert client.get("/healthz").status_code == 200

def test_me_no_token(client):
    resp = client.get("/me")
    assert resp.status_code == 401
    assert resp.headers["content-type"] == "application/problem+json"

def test_me_with_token(client):
    mock_auth({"sub": "alice", "tenant_id": "acme-corp", "roles": ["viewer"]})
    resp = client.get("/me", headers={"Authorization": "Bearer test"})
    assert resp.status_code == 200
    assert resp.json()["sub"] == "alice"

def test_create_product_requires_editor(client):
    mock_auth({"sub": "alice", "roles": ["viewer"]})
    resp = client.post("/products",
                       json={"name": "Widget"},
                       headers={"Authorization": "Bearer test"})
    assert resp.status_code == 403

Summary

Feature	How
JWT auth on every route	`CoreSDKMiddleware` validates `Authorization: Bearer`
Claims in handlers	`request.state.coresdk_user`
RBAC	`require_role("editor")` dependency
ABAC	`_sdk.evaluate_policy("data.authz.allow", {...})`
Multi-tenancy	`tenant_id` from JWT claims scopes all DB queries
PII-safe tracing	`@trace(intent=...)` on every route
RFC 9457 errors	`@app.exception_handler(ProblemDetailError)`
Config file	`coresdk.toml` + `CORESDK_*` env var overrides

FastAPI Integration

On this page