Metadata-Version: 2.5
Name: gentiq
Version: 0.15.0b2
Summary: A world-class, modular framework for building production-ready AI chatbots.
Project-URL: Homepage, https://github.com/arxyzan/gentiq
Project-URL: Repository, https://github.com/arxyzan/gentiq
Project-URL: Documentation, https://github.com/arxyzan/gentiq#readme
Project-URL: Issues, https://github.com/arxyzan/gentiq/issues
Author-email: Aryan Shekarlaban <arxyzan@gmail.com>
License: Apache-2.0
Keywords: ai,chatbot,fastapi,framework,pydantic-ai
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.12
Requires-Dist: argon2-cffi>=23.1.0
Requires-Dist: email-validator>=2.0.0
Requires-Dist: fastapi>=0.129.0
Requires-Dist: genai-prices>=0.1.4
Requires-Dist: httpx2>=2.12.0
Requires-Dist: pydantic-ai-slim[anthropic,google,logfire,mcp,openai,ui]<3,>=2.34.0
Requires-Dist: pyjwt>=2.11.0
Requires-Dist: python-dotenv>=1.2.1
Requires-Dist: python-multipart>=0.0.22
Requires-Dist: sse-starlette>=3.2.0
Requires-Dist: starlette>=0.52.1
Requires-Dist: uvicorn>=0.40.0
Provides-Extra: engines
Requires-Dist: minio>=7.2.20; extra == 'engines'
Requires-Dist: pymongo>=4.16.0; extra == 'engines'
Description-Content-Type: text/markdown

# 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](https://fastapi.tiangolo.com/) and [PydanticAI](https://ai.pydantic.dev/). 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

```bash
pip install gentiq
```

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

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

*For monorepo development, install in editable mode:*

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

---

## 💡 Quick Start

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

| Variable | Notes |
| --- | --- |
| `JWT_SECRET_KEY` | **Required.** Signs user tokens. |
| `BACKEND_API_KEY` | **Required.** Guards server-to-server endpoints such as `POST /api/auth/user`; they reject every request while it is unset. |
| `ADMIN_JWT_SECRET_KEY` | Optional — derived from `JWT_SECRET_KEY` when unset, so the two privilege domains stay distinct. |
| `JWT_EXPIRATION_HOURS` / `ADMIN_JWT_EXPIRATION_HOURS` | Default 24 h and 8 h. |
| `LOGIN_FIELDS` | Comma-separated subset of `username`, `email`, `phone`. Defaults to `username`. |
| `INITIAL_BALANCE_TOKENS` / `INITIAL_BALANCE_REQUESTS` | Starting balance for each new user. |
| `MAX_ATTACHMENT_SIZE` | Default 10 MiB. `MAX_REQUEST_BODY_SIZE` is derived from it to allow for base64 inflation. |
| `ARGON2_TIME_COST` / `ARGON2_MEMORY_COST` / `ARGON2_PARALLELISM` | Password-hashing cost, sized for a small container by default. |
| `LOGFIRE_TOKEN` | Enables tracing when `send_to_logfire=True`. |
| `MONGODB_*` / `MINIO_*` | Only for `db_engine="mongodb"` / `storage_engine="minio"`. |

### Usage Cost Tracking

Usage is priced and snapshotted automatically, per request, against
[`genai-prices`](https://github.com/pydantic/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:

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

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

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

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

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

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

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

```python
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](#upgrading-to-0150).

#### 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.

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

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

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

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

| Removed from the permission picker | Replaced by |
| --- | --- |
| `jobs` | `operations` |
| `dangerous_jobs` | `operations` |

**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](LICENSE).
