caronte-sdk

Python client for the Argos authorizer — handles app authentication, JWT validation, permission checking and automatic operation sync.

Installation

pip install caronte-sdk
# With FastAPI support
pip install "caronte-sdk[fastapi]"

Quick start with FastAPI

from fastapi import FastAPI, Request
from caronte.adapters.fastapi import Caronte, operation

app = FastAPI()

caronte = Caronte(
    app,
    authorizer_url="http://localhost:4000/api",
    app_id="your-app-uuid",
    secret="your-app-secret",
)


@app.get("/public")
@operation(id="public.info", level="public")
async def public_info():
    """Anyone can call this — no token needed."""
    return {"message": "Hello, world!"}


@app.get("/items")
@operation(id="items.list", level="private")
async def list_items(request: Request):
    """Requires any authenticated user (token with non-empty groups)."""
    user = request.state.user  # TokenClaims
    return {"user": user.sub, "items": []}


@app.delete("/items/{item_id}")
@operation(id="items.delete", level="protected")
async def delete_item(item_id: str, request: Request):
    """Requires a user whose group has explicit permission for items.delete."""
    return {"deleted": item_id}

Response envelope ({ status, message, data })

Every route answers in the same shape, on success and on error — status is the HTTP code, message is user-facing text (the UI shows it as a toast), data is the payload (None on error).

from caronte import Conflict, NotFound, ok
from caronte.adapters.fastapi import Envelope, install_envelope, operation

app = FastAPI()
install_envelope(app)   # EnvelopeError / HTTPException / 422 / unexpected 500 → { status, message, data: null }

@app.post("/tags", status_code=201, response_model=Envelope[TagOut])
@operation(id="tags:create", level="protected")
async def create_tag(body: TagCreate, db: Session = Depends(get_db)):
    return ok(TagService(db).create(body.name), "Tag criada.", 201)

class TagService:
    def get(self, tag_id):
        tag = self.repo.get_by_id(tag_id)
        if not tag:
            raise NotFound("Tag não encontrada.")   # → 404 { "status": 404, "message": "...", "data": null }
        return tag

• caronte.envelope is framework-agnostic: ok, fail, detail_text and the business errors BadRequest (400), Unauthorized (401), Forbidden (403), NotFound (404), Conflict (409), Unprocessable (422), Unavailable (503) — all EnvelopeError subclasses.
• Envelope[T] (FastAPI adapter) is the response model: it documents the contract in OpenAPI and filters data to the declared fields.
• Unexpected exceptions become 500 "Erro inesperado no serviço." — the detail goes to the log only.
• The ASGI middleware's own 401/403 also carry status/message/data (plus the previous error/detail keys, for compatibility).

Standalone client

import asyncio
from caronte import CaronteClient, OperationDescriptor

client = CaronteClient(
    authorizer_url="http://localhost:4000/api",
    app_id="your-app-uuid",
    secret="your-app-secret",
)

client.register_operation(
    OperationDescriptor(identifier="items.list", method="read", level="private")
)

async def main():
    await client.startup()

    # Validate a token from an incoming request
    claims = await client.validate_token("<bearer-token>")

    # Check whether the token has permission
    allowed = client.check_permission(claims, "items.list", "read")
    print(f"Allowed: {allowed}")

asyncio.run(main())

Credentials

A caller authenticates with exactly one of three headers (never Authorization / Bearer):

Opaque tokens (x-app-token, x-api-token) are resolved by POST /realms/{realm}/tokens/introspect, authenticated with this backend's own client-credentials token. The response is cached per token (INTROSPECTION_TTL, 60 s) and every operation is then decided locally, so a client calling many operations costs one round-trip per token per minute. An x-app-token becomes the same TokenClaims a JWT yields (raw.token_type === "app_token", raw.token_id), so handlers read the user exactly as before; logs record actor_type = app_token plus the token id.

Against an authorizer without the introspection endpoint (HTTP 404), x-api-token falls back to the previous per-operation POST /apps/tokens/validate and x-app-token is rejected.

Operation levels

Method auto-detection

The @operation decorator infers the method from the function name:

Configuration

import os
from caronte.adapters.fastapi import Caronte

caronte = Caronte(
    app,
    authorizer_url=os.environ["AUTHORIZER_URL"],
    app_id=os.environ["APP_ID"],
    secret=os.environ["APP_SECRET"],
)

Observability (OpenTelemetry)

Set otel_collector_url to enable tracing, metrics and logs — auth/JWKS calls (via httpx) and, when installed, your framework's requests (Django/FastAPI/ Flask) get traced and measured automatically and exported to the Collector; Python's standard logging output is forwarded too (via a LoggingHandler on the root logger). Never breaks startup(): a missing [otel] extra or an unreachable Collector just logs a warning.

pip install "caronte-sdk[otel]"

client = CaronteClient(
    authorizer_url=os.environ["AUTHORIZER_URL"],
    realm_id=os.environ["REALM_ID"],
    app_id=os.environ["APP_ID"],
    secret=os.environ["APP_SECRET"],
    otel_collector_url=os.environ.get("OTEL_COLLECTOR_URL"),  # e.g. http://localhost:4318
)

Every framework adapter (caronte.adapters.django/fastapi/flask/strawberry) accepts the same otel_collector_url (Django: CARONTE_OTEL_COLLECTOR_URL setting instead, since the middleware is constructed by Django itself) and forwards it to its internal CaronteClient.

Span attributes

Adapters generally don't create spans themselves — they enrich whichever span the auto-instrumentation registered by otel_collector_url already created for the current request (ASGI/httpx/etc). Django is the one exception: DjangoInstrumentor patches in by inserting itself into settings.MIDDLEWARE from inside CaronteMiddleware.__init__, but Django builds its middleware chain by iterating reversed(settings.MIDDLEWARE) — since CaronteMiddleware must be first in that list, its __init__ runs last in that loop, leaving no earlier slot for the auto-injection to claim. So CaronteMiddleware.__call__ creates the request's SPAN_KIND_SERVER span manually instead of relying on that auto-injection.

Each adapter's auth/permission hook adds, best-effort:

Routes/resolvers with no @operation decorator are left untouched — no caronte.* attributes are added, since there's no caronte-relevant decision being made.

Strawberry: composing with the official OTel extension

caronte.extension only enforces auth/permissions — it never creates spans itself. For per-resolver GraphQL tracing, add the official OTel extension alongside it; that's an app-level choice, not something caronte-sdk forces:

from opentelemetry.instrumentation.strawberry import OpenTelemetryExtensionSync

schema = strawberry.Schema(
    query=Query,
    extensions=[caronte.extension, OpenTelemetryExtensionSync()],
)

Note: without the second extension, resolve() still enriches whatever span is currently active (typically the single request-level span from your ASGI framework's instrumentation) — with multiple resolvers running per request, the last one to run wins on shared attribute keys. Adding OpenTelemetryExtensionSync() gives each resolver its own span instead.