Gentiq Backend Framework (Python)

The core Python engine for building high-performance, production-ready Agentic AI backends.

gentiq is a modular framework built on top of FastAPI and PydanticAI. It handles all the heavy lifting—persistence, security, and streaming—allowing you to focus entirely on defining your agents and tools.

🚀 Key Features

• GentiqApp Factory: Rapidly initialize a production-ready FastAPI application with just an agent.
• Deep PydanticAI Integration: Fully supports PydanticAI's type-safe agent system and dependency injection.
• Injected AgentDeps: Automatic access to UserStore, ChatStore, and the current User inside every tool.
• Plans, Credits & Quotas: Each user is on a plan with calendar-aligned limits, a pool of credits, or unlimited — enforced atomically per turn.
• Per-Model Cost Accounting: Every request priced by the provider and model that actually served it.
• Pluggable Persistence: Support for SQLite, MongoDB, S3, and MinIO out of the box.
• JWT Auth with argon2id: User and admin token domains, permission-checked admin routes.
• Observability: First-class support for Logfire for tracing agent reasoning and tool execution.

📦 Installation

pip install gentiq

The base install runs on SQLite and the local filesystem — no services required. For the MongoDB and MinIO engines:

pip install "gentiq[engines]"     # or: uv add "gentiq[engines]"

For monorepo development, install in editable mode:

# In your app's pyproject.toml
[tool.uv.sources]
gentiq = { path = "../../../packages/gentiq-python", editable = true }

💡 Quick Start

from gentiq import AgentDeps, CORSConfig, GentiqApp
from pydantic_ai import Agent

# 1. Define your agent (typed with Gentiq dependencies)
agent = Agent[AgentDeps[None]]("openai:gpt-5.1")

# 2. Boot the app
app = GentiqApp(
    agent,
    app_name="MyAI",
    app_version="1.2.3",
    # No CORS middleware is installed unless you ask for it. Omit this when the
    # frontend is same-origin or proxied; list every browser origin that calls
    # this API directly otherwise.
    cors=CORSConfig(allow_origins=["http://localhost:5173"]),
)

# GentiqApp.api is a regular FastAPI instance
# Run with: uv run uvicorn main:app.api --reload --port 8000

If your app already exposes a version constant, pass that value into app_version so Gentiq uses the same source of truth as the rest of your backend.

Environment

Gentiq reads a .env relative to the process's working directory (point ENV_FILE elsewhere to override). Two variables have no default and fail closed:

Usage Cost Tracking

Usage is priced and snapshotted automatically, per request, against genai-prices — the rate data pydantic-ai already ships. It covers every provider pydantic-ai supports, resolves aliases and dated snapshots (gpt-4o-2024-08-06 → gpt-4o), and carries cache-read, cache-write and audio rates, long-context tiers, and rates that change over time. Each response is priced by the provider and model that actually served it, so a FallbackModel run or a per-run model override is still billed correctly.

UsagePricing is an override layer, consulted first and empty by default. Use it for negotiated or resale rates, or to correct a model the bundled data has wrong:

from gentiq import GentiqApp, ModelPrice, UsagePricing

app = GentiqApp(
    agent,
    usage_pricing=UsagePricing(
        prices={
            # Keys are "{provider}:{model}", matched case-insensitively.
            "openai:gpt-5.1": ModelPrice(
                input_per_million="1.50",
                output_per_million="12.00",
                cache_read_per_million="0.15",
            )
        }
    ),
)

Pass a complete rate card in another currency — or one that should be the only source of truth — with use_price_data=False, which leaves anything not listed unpriced:

custom_pricing = UsagePricing(
    currency="EUR",
    prices={"openai:my-model": ModelPrice("2.00", "8.00")},
    use_price_data=False,
)

Only input_per_million and output_per_million are required; a bucket left as None is billed at the rate of the bucket it is carved out of (cached input at the input rate, and so on) rather than at zero. Each ledger row records the rates applied, the canonical model billed (billed_as) and the price_source (override or genai-prices).

Plans & Usage Quotas

Gentiq meters every chat turn and enforces what each user may use. It supplies the mechanism — meters, windows, plans, credits, enforcement — and your app owns the catalogue: which plans exist and what each allows.

Every user has exactly one kind of access at a time, the way a chat product works:

• A plan — limits that reset on the calendar: 200 messages a day, 2M tokens a month.
• Credits — a fixed number to spend, which never refills on its own.
• Unlimited — nothing is counted against them. This is the default.

from gentiq import GentiqApp, Limit, Plan, PlanDuration

app = GentiqApp(
    agent,
    plans=[
        Plan(key="free", name="Free", limits=[Limit(meter="messages", amount=20, per="day")]),
        Plan(
            key="plus",
            name="Plus",
            duration=PlanDuration(per="year"),
            limits=[
                Limit(meter="tokens", amount=250_000, per="day"),
                Limit(meter="tokens", amount=2_000_000, per="week"),
                Limit(meter="messages", amount=10, per="minute"),
            ],
            features={"web_search": True},
        ),
    ],
    default_plan="free",  # or default_credits=500; with neither, users are unlimited
    quota_timezone="Asia/Tehran",
    week_start="saturday",
)

• Meters. messages (one accepted chat turn, reserved atomically so the limit holds under concurrent requests), tokens, model_requests and cost (microunits of your pricing currency), counted when the turn ends. Any other name is an app-defined meter you record from a tool.
• Windows. per="minute" | "hour" | "day" | "week" | "month" align to the calendar in quota_timezone: a day starts at local midnight, a week on week_start, a month on the 1st of the quota calendar's month — 1 Mehr rather than 1 October with quota_calendar="persian", which is the default for Asia/Tehran. With anchor="subscription" a window counts from the day the subscription starts instead, in steps of every — a billing month that renews on the 14th, or every 2 days. These still reset on fixed points: days, weeks and months at local midnight, hours on the hour — never at the minute the plan was assigned. per="subscription" never resets while the assignment lasts.
• Length. A plan's duration — PlanDuration(per="month", every=3), PlanDuration(per="year") — is how long it lasts once assigned; without one it lasts until changed. The limits apply inside it: the Plus plan above runs for a year, and allows 250k tokens a day and 2M a week throughout. A plan ends at local midnight: by default its length counts from the start of the day it was assigned (a month from 16 Sep ends on 16 Oct at 00:00); with PlanDuration(per="month", align="calendar") the period it was assigned in counts as the first, so it ends on the 1st of a month, the first day of a week, or 1 January (a month from 16 Sep ends on 1 Oct).
• Subscriptions put a user on a plan from starts_at (now, by default) for the plan's length, keyed by an external_id you choose (an order or enrolment id), so writing the same assignment twice is the same assignment. The end is never sent; it follows from the plan. A user has one at a time: a subscription that is created or changed ends any other it overlaps. One that ends simply stops applying, and the user is back on their credits or the default.
• Nothing renews on its own. Payment happens on your platform, so renewing is yours to say: when the user pays again, write a new subscription under the new order's external_id. To extend without a gap or an overlap, start it where the current one ends (starts_at = its ends_at).
• Credits count one meter — credits_meter="messages" | "tokens" | "cost", tokens by default. They are granted per user from the admin panel or the integration API, or every user starts with default_credits.
• Usage counts up. Moving a user to a bigger plan mid-day gives them the difference; raising a plan's limit gives every subscriber headroom on their next message. Nothing is ever reset and nothing runs on a schedule — a new window is a counter that does not exist yet.

What applies to a user, checked on every turn: your resolve_plan hook if you pass one, then their subscription, then their own credits, then the default access. The default is configured in code (default_plan or default_credits) and can be changed by admins under Settings → Usage & access, along with the credit unit, the quota timezone, the first day of the week, and whether the chat's top bar shows users a meter of what they have left. A refused turn gets 429 quota_exceeded with Retry-After and the reset time, 402 credits_exhausted, or 402 plan_expired when a plan ended and the default credits are gone too.

Admins set one user's access — default, a plan, credits, or unlimited — from the users table, or many users' at once with Change access. Plans declared in code are written to the database at startup and are read-only under Users → Plans, where admins can create more. A plan made there can be archived, so nobody new is put on it while current subscribers keep it, or deleted: deleting a plan anyone is on asks for a plan to move them to. They are put on it as though an admin had assigned it — starting today (or when a booked subscription would have), for its length, with its limits — and the default access follows if it was the deleted plan (DELETE /api/admin/plans/{key}?replace_with=basic, or app.entitlements.delete_plan(key, replace_with=...)). Each user's analytics button in Users opens their usage over time. From your own code, use app.entitlements.set_access(user, {"mode": "plan", "plan": "student"}, source="app") — it starts at the beginning of today in the quota timezone and lasts the plan's duration. Your tools can read what the user is on and meter their own usage:

@agent.tool
async def web_search(ctx: RunContext[AgentDeps[MyContext]], query: str) -> str:
    entitlement = ctx.deps.entitlement
    if not entitlement or not entitlement.features.get("web_search"):
        return "Web search is not included in the user's plan."
    if ctx.deps.usage and not ctx.deps.usage.allows("web_searches"):
        return "The user has no web searches left today."
    results = await search(query)
    if ctx.deps.usage:
        ctx.deps.usage.record("web_searches")
    return results

Sharing a plan between users

Every user belongs to exactly one subscription group, and plans and usage belong to the group. A user nobody put in a group is a group of their own, so all of the above applies to them as written. Users given the same subscription_group — an account, a household, a team — share one plan and one set of counters: if the plan allows 500 messages a day, that is 500 for all of them together, and any one of them can use all 500.

app.entitlements.set_subscription_group(user, "account-42", source="app")  # join, or move
app.entitlements.set_subscription_group(user, None, source="app")  # leave

• A member has no plan of their own. A subscription written, cancelled or set through any member is the group's, from the integration API, the admin panel and app.entitlements alike. Writing a new one for the group ends the group's previous one, as it does for a single user.
• Joining is destructive. It ends the user's own subscriptions and takes away their own credits; leaving restores neither, so a user who leaves is on the default access.
• Groups take plans, not credits. Granting credits to a member is refused with 409 credits_not_for_groups. The default access is the group's too: four members of a group with no plan share one default allowance, not four.
• Usage stays where it was counted. Joining, leaving or moving never moves or resets usage, so adding, removing or re-creating members cannot create allowance.
• Membership is a field on the user, readable as user.subscription_group. A group has no record of its own, and its plan and usage are stored under the key group:<id> (Subscription.group gives the id back). A resolve_plan hook must return the same plan to every member.

Conversations, analytics and the usage ledger stay per user; each usage row records the group that paid for it. Admins set a user's group when editing them under Users, and searching for a group id lists its members.

Linking Users From Another Platform

A platform that signs its own users in — a school portal, a SaaS dashboard — links them to the chat through /api/integration/v1, with a scoped key under the admin panel's Settings → Integrations. The API separates who the user is, what they may use, and whether they are here now, so that every call is safe to repeat:

PUT  /api/integration/v1/users/{external_id}                          profile (never touches usage)
PUT  /api/integration/v1/users/{external_id}/subscriptions/{order_id} assign a plan, idempotent
POST /api/integration/v1/sessions                                     sign the user in

Call POST /sessions each time the user opens the chat:

{
  "user": { "external_id": "stu_48121", "name": "Sara" },
  "subscription": { "external_id": "enrolment-2026-fall", "plan": "student" },
  "redirect_path": "/"
}

The response carries a token for API clients and a single-use, two-minute login_code for browsers: redirect to login_url (or your chat URL with ?code=) and the frontend exchanges it, so no bearer token travels in a URL. Mint a code per page render. A page loaded again with a spent or expired code still signs in, but only if the browser already holds a session for the user that code was issued for; anything else lands on the frontend's RequireAuth fallback (by default /login). The subscription is optional; sending the same one on every sign-in changes nothing, because it is keyed by its external_id and its end is worked out once, from the stored start and the plan's length. Users are found by external_id only — never by email or username — unless you pass link_by to adopt an existing account once.

Also available: GET /users/{external_id}/usage, POST .../subscriptions/{id}/cancel, POST /users/{external_id}/credits with {"amount": n} (additive, so it requires an Idempotency-Key header) and GET /plans. The legacy POST /api/auth/user keeps working and accepts external_id and subscription too.

Accounts with several profiles. When your platform sells a plan to an account and each of the account's profiles is a chat user, make the account a subscription group (see Sharing a plan between users). On every sign-in, send the profile's current account as user.subscription_group, and that account's current order as the subscription:

{
  "user": { "external_id": "profile-17", "name": "Sara", "subscription_group": "account-42" },
  "subscription": { "external_id": "order-981", "plan": "family", "starts_at": "2026-09-01T00:00:00Z" }
}

The group is applied before the subscription, so the order is written for the group, and every profile of the account uses the same plan and the same counters. Sending it again writes nothing.

subscription_group is also accepted by PUT /users/{external_id} and returned by every user response. Group ids are permanent: reusing one inherits that group's plan and usage. Members can see their group's id and subscription (including its metadata) through /api/users/me, so use opaque ids.

🛠️ Advanced Customization

Custom Application Context

You can inject any custom object (database pools, service clients, config) into your agent tools via the context parameter.

@dataclass
class AppContext:
    weather_api_key: str


agent = Agent[AgentDeps[AppContext]](...)


@agent.tool
async def get_weather(ctx: RunContext[AgentDeps[AppContext]], city: str):
    # Access your custom context easily
    api_key = ctx.deps.context.weather_api_key
    return {"temp": 22, "city": city}


app = GentiqApp(agent, context=AppContext(weather_api_key="secret"))

Real-time UI Updates (Streaming)

Gentiq allows you to stream custom events to the frontend while a tool is still running. This is perfect for long-running processes where you want to show progress.

from gentiq import ProgressUpdateEvent


@agent.tool
async def long_task(ctx: RunContext[AgentDeps[AppContext]]):
    await ctx.deps.stream(
        ProgressUpdateEvent(
            tool_name="long_task", status="running", message="Analyzing data... this might take a moment."
        )
    )
    # ... perform work ...
    return "Task completed!"

Accessing Core Stores

Tools have full access to Gentiq's internal stores, enabling agents to perform complex operations like searching through the user's past chat history. Store methods are synchronous — run anything slow through asyncio.to_thread if it would otherwise block the event loop.

@agent.tool
async def search_past_chats(ctx: RunContext[AgentDeps[AppContext]], query: str):
    # Access the ChatStore directly
    threads = ctx.deps.chat_store.list_user_threads(ctx.deps.user.id, limit=20)
    hits = [t for t in threads if query.lower() in (t.get("title") or "").lower()]
    return {"results": hits}

Multi-Agent Transparency

When a tool delegates to another agent, that run happens in its own PydanticAI run and is normally invisible in the admin panel. Wrap it in ctx.deps.capture_subagents(ctx) to record the sub-agent's full transcript (input, output, reasoning, tool calls) into the chat history — shown in the admin panel only, never to the end user. Logging is always on inside the block; passing the tool's ctx also rolls the sub-agent's tokens up into the thread's usage, priced at that sub-agent's own model rates.

@agent.tool
async def detailed_forecast(ctx: RunContext[AgentDeps[AppContext]], city: str) -> str:
    async with ctx.deps.capture_subagents(ctx):
        result = await forecast_agent.run(f"Give a 5-day forecast for {city}.", deps=ctx.deps)
    return result.output

Sub-agents driven via .run_stream() / .iter() are not captured.

Interactive Choice Questions

Let the agent hand the conversation back to the user as a set of buttons instead of guessing at an ambiguous request. choice_questions=True uses Gentiq's default policy on when to ask; passing a string replaces that policy with your own. The wire format the backend parses is appended either way.

app = GentiqApp(
    agent,
    choice_questions=(
        "Ask a choice question only when a request is genuinely ambiguous. "
        "Answer directly otherwise, and never use one just to offer follow-up topics."
    ),
)

Maintenance Operations

Register migrations and one-off fixes as jobs runnable from the admin panel's Operations tab — for the times you cannot get a shell on the production server. The job receives a JobContext exposing every store and the raw DB engine, validated ctx.params, a ctx.dry_run flag, and ctx.log(...) whose output is captured into the run record.

from gentiq import JobContext, ParamSpec


@app.job(
    id="count_users",
    name="Count users",
    description="Reports how many users exist. Safe to run anytime.",
    danger="safe",
    params=[ParamSpec(name="prefix", type="str", required=False, label="Name prefix")],
)
def count_users(ctx: JobContext) -> dict:
    prefix = (ctx.params.get("prefix") or "").strip()
    flt = {"name": {"$regex": f"^{prefix}", "$options": "i"}} if prefix else {}
    count = ctx.engine.count_documents("users", flt)
    ctx.log(f"Matched users: {count}")
    return {"count": count}

Operations are gated behind the admin operations permission. enable_raw_jobs=True additionally allows running arbitrary Python from the panel; since 0.15.0 that is covered by the same operations permission rather than a second one, so leave enable_raw_jobs off unless every Operations admin should have what amounts to shell access.

operations replaced the jobs and dangerous_jobs permissions in 0.15.0. Stored grants for either still work and are rewritten automatically — see Upgrading to 0.15.0.

Returning files

An operation can publish a downloadable artifact by returning a JobFile. The bytes go to the app's storage engine (filesystem or MinIO); only the address travels in the run record.

@app.job(id="export_users", name="Export users", danger="safe")
def export_users(ctx: JobContext) -> dict:
    csv = "id,name\n" + "\n".join(f"{u['id']},{u['name']}" for u in ctx.engine.find_many("users", {}))
    # Alternatives: ctx.save_path("/tmp/report.pdf") for a file on disk, or
    # ctx.file("reports/2026-01.pdf") to point at an object already in storage.
    return {"users": ..., "export": ctx.save_file(csv, filename="users.csv")}

Return one on its own or nested anywhere in the result. On the wire each becomes a Gentiq-native envelope tagged with the reserved __gentiq__ key (gentiq.job_file/1) — dunder-namespaced so an application's own result fields cannot collide with it — and the run lists them under files, which is what the admin panel renders as download buttons.

Artifacts are served from GET /api/admin/jobs/runs/{run_id}/files/{index} by index, so the endpoint can only hand back files a job actually published, never arbitrary objects from the storage backend. It requires the same operations permission as the rest of the tab.

Login Handles

login_fields chooses which of username, email and phone a user can sign in with. Only enabled fields are unique; the rest are ordinary, non-unique profile data. The real identity is always the immutable user id, which is what JWTs carry.

app = GentiqApp(agent, login_fields=["email", "phone"])

🏗️ Pluggable Architecture

Persistence Engines

Gentiq is designed to be storage-agnostic. You can choose from built-in engines or implement your own by subclassing DBEngine or StorageEngine.

# Use MongoDB and MinIO for production scale
app = GentiqApp(
    agent,
    db_engine="mongodb",  # Scales better for message history
    storage_engine="minio",  # Perfect for large file attachments
)

Both parameters also accept an engine instance, so a custom subclass drops straight in.

Extending the API

Since GentiqApp.api is a standard FastAPI instance, you can add your own routes, middleware, and exception handlers while still benefiting from Gentiq's built-in authentication.

from typing import Annotated

from fastapi import APIRouter, Depends
from gentiq import User, get_current_user

router = APIRouter()


@router.get("/profile")
async def get_profile(user: Annotated[User, Depends(get_current_user)]):
    return {"name": user.name, "email": user.email}


app.add_router(router, prefix="/v1")

Other dependencies worth knowing: get_current_admin and require_permission(...) for admin-only routes, and OwnedThreadId / WritableThreadId for any route that takes a thread id — they enforce ownership rather than trusting the client's header.

Adding your own CORSMiddleware replaces Gentiq's rather than stacking a second one, so you never end up emitting duplicate headers.

⬆️ Upgrading to 0.16.0

Balances and recharge policies are gone. Each user now has one kind of access — a plan, credits, or unlimited — and the default for users nobody configured is unlimited. Deploy and restart; the migration runs at startup:

• Every user with a balance or recharge policy moves to the default access, which is unlimited unless you configure one. No plan, subscription or credits are created for them. Put everyone on a plan afterwards in one step: Users → Change access → Everyone in the admin panel, or set Default access under Settings → Usage & access (or default_plan=) so it applies to them and to new users alike. The startup log warns while migrated users are left unlimited.
• balance and recharge_policy are removed from every user row, with a migration ledger entry recording what they held, and initial_balance_tokens / initial_balance_requests are removed from the settings.
• New users no longer start with a balance. INITIAL_BALANCE_TOKENS / INITIAL_BALANCE_REQUESTS are ignored. To keep limiting new users, pass default_plan= or default_credits= to GentiqApp, or set Default access under Settings → Usage & access. The startup log warns when the old settings handed out a starting balance.
• POST /api/auth/user refuses balance and recharge_policy with a 422 that names them, rather than ignoring them and leaving the users you meant to limit unlimited. Send a subscription instead, or grant credits through POST /api/integration/v1/users/{id}/credits.
• Removed endpoints: GET /api/users/balance (the frontend checks the session with /api/users/me), POST /api/admin/users/{id}/balance, /api/admin/users/bulk-balance and /api/admin/users/bulk-renewal. Use PUT /api/admin/users/{id}/access and POST /api/admin/users/bulk-access.
• Refusals changed shape. A spent window is 429 quota_exceeded with Retry-After; spent credits are 402 credits_exhausted (was insufficient_balance); a plan that ended with nothing left to spend is 402 plan_expired.
• BACKEND_API_KEY is now called the root key. It still works everywhere. An invalid key on the server-to-server endpoints now gets 401 rather than 403. Issue scoped keys to partners.
• The primary admin gains the integrations permission if it still holds every other built-in permission.
• The settings dialog's balance field mode is now usage. A stored balance mode is renamed at startup, and a frontend config that still sets balance is honoured.
• Subscription groups need no migration and no setting. Nothing changes until a user is given a subscription_group. From then on Subscription.user_id can hold group:<id>, so code that looks a user up by it should check Subscription.group first; and put_subscription, cancel_subscription, end_subscriptions and set_access called with a member act on the member's group, while the credit methods refuse a member.

⬆️ Upgrading to 0.15.0

The jobs and dangerous_jobs permissions became operations

Maintenance jobs moved out of the settings page into their own Operations admin tab, and the two permissions that gated them merged into one:

Deploy the new version and restart. That is the whole upgrade — there is no script to run, by design, since production deployments cannot always run one-off commands.

The migration is invoked from the application lifespan, so it happens on startup, before the first request is served. Note that this is server startup, not construction: GentiqApp(...) on its own touches nothing. It rewrites jobs / dangerous_jobs to operations on every admin row, and grants operations to the primary admin (the first one created) even if it never held jobs, so the Operations tab is never left unreachable. It is idempotent, re-runs harmlessly on every boot, and can never block startup — a failure is logged and the app comes up anyway.

Two things back it up if that pass does not happen:

• A stored jobs or dangerous_jobs grant still authorizes every Operations route, and is rewritten the first time that admin document is read — so an admin converges their own row by logging in.
• A JWT minted before the upgrade carries the retired value in its claims, where no database write can reach it. Permission checks normalize the token's claims in memory, so it keeps working until it expires.

One behavior change to be aware of. dangerous_jobs used to be a second, separately-granted escalation over jobs; now a single operations grant covers the raw-Python runner too. Any admin who held only jobs gains the ability to reach it. The runner is still gated on the deployment-level enable_raw_jobs switch (off by default) and a typed confirmation phrase — but if you were relying on the two-tier split to keep some Operations admins away from arbitrary code execution, set enable_raw_jobs=False, or review who holds operations after migrating.

Frontend: disabledPages and AdminPage.permission

If you pass disabledPages={['jobs']} to the admin panel, or register a custom AdminPage with permission: 'jobs', both keep working — 'jobs' is accepted as a deprecated alias of 'operations' and normalized at runtime. Prefer 'operations' in new code.

The tab itself moved from /admin/jobs to /admin/operations; the old path redirects, so existing bookmarks and deep links still land. API route paths are unchanged — the backend keeps its job vocabulary, and GET /api/admin/jobs/registered simply gained a raw_enabled field so the panel can tell whether the Raw Python sub-tab is worth showing.

📄 License

Gentiq is open-source software licensed under the Apache 2.0 License.