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.
• Atomic Balance Tracking: Integrated per-user token and request balance management.
• 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).

🛠️ 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.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.