Metadata-Version: 2.5
Name: ragnarok-api
Version: 0.1.0
Summary: Python SDK for the RAGnarok HTTP API (/api/v1): auth, streaming queries, ingest, graph and sources.
Author: Laurent-Philippe Albou
License: MIT License
        
        Copyright (c) 2026 Laurent-Philippe Albou
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: graphrag,knowledge-graph,rag,ragnarok,sdk,sse
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2
Provides-Extra: test
Requires-Dist: pytest>=8; extra == 'test'
Description-Content-Type: text/markdown

# ragnarok-api (Python)

The Python SDK for the RAGnarok HTTP API (`/api/v1`). It is a thin layer over the
published contract (`specs/openapi.json` in this repository): typed
models, the SSE event stream of `/query` and of ingest jobs, multipart upload with
progress, binary downloads, typed errors and cursor paging. It holds no business
logic. Caps, roles, concurrency and model allow-lists are enforced by the server
and come back to you as typed errors.

- Runtime dependencies: `httpx>=0.27`, `pydantic>=2`. Python 3.10+.
- Sync `Client` and async `AsyncClient` with the same surface.
- Versioning: the SDK major follows the API major (`1.x` speaks `/api/v1`; `0.x` is the
  pre-release line for v1). `client.check_api_version()` refuses a server whose
  `/api/health` reports another major.

```bash
pip install -e ./sdks/python
```

## Connect

```python
from ragnarok_api import Client, Token

client = Client("https://ragnarok.example", Token("rgk_..."))   # personal API token
print(client.health().status, client.check_api_version().api_version)
print(client.auth.whoami().email)
```

`api_prefix` defaults to `/api/v1`. Pass `api_prefix="/api"` for a server that
predates the versioned mount. `timeout` defaults to 240 s. SSE streams use
a 480 s read backstop (`stream_read_timeout`). Any httpx transport works, including
the in-process `httpx.ASGITransport(app=app)`.

## 1. Query with streaming

```python
from ragnarok_api import AnswerEvent, DoneEvent, ErrorEvent, SourcesEvent, StatusEvent

for event in client.query.stream("What is the approved indication?", workspace="epar", mode="quick"):
    if isinstance(event, StatusEvent):
        print("..", event.message)
    elif isinstance(event, AnswerEvent):
        print(event.text)
    elif isinstance(event, SourcesEvent):
        for src in event.vector_sources:
            print(f"  [{src.citation_label}] {src.chunk_id}")
    elif isinstance(event, ErrorEvent):
        print("error:", event.code, event.message)
    elif isinstance(event, DoneEvent):
        print("done, success =", event.success)

# Or collect everything (raises QueryFailed on an `error` event):
result = client.query.ask("What is the approved indication?", workspace="epar")
print(result.answer, len(result.sources.vector_sources), result.stats.elapsed_ms)
```

Events: `status, answer, answer_delta, citation_map, references, notices, sources,
entities, graph, stats, error, done`. An event the SDK cannot type (a new event name,
or data that is not JSON) arrives as `UnknownEvent`. It is never dropped. If the
stream ends without `done`, the SDK yields a client-generated `error`
(`code="query_network_error"`) and then `done(success=False)`, as the web UI does.

The server records every completed answer in the query history, shared by the web UI,
the `ragnarok` app and the SDKs: `DoneEvent.history_id` (and `result.history_id` from
`ask`) names the entry; it is `None` with `history_error` set when recording failed.
Pass `record_history=False` to keep a query out of the history (evaluations, scripted
batches), or `history_id="..."` to record under your own id (a retry with the same id
updates the entry, never duplicates it).

Async:

```python
import asyncio
from ragnarok_api import AsyncClient, Session

async def main():
    async with AsyncClient("https://ragnarok.example") as client:
        await Session.alogin(client, "me@example.com", "secret")
        async for event in client.query.stream("Summarise the safety profile", workspace="epar"):
            print(event.event)

asyncio.run(main())
```

## 2. Ingest a folder

```python
from ragnarok_api import collect_folder

folder = collect_folder("./dossiers")            # .pdf .txt .md .rmd, relative upload names
print(len(folder), "files;", len(folder.skipped), "skipped:", folder.skipped)

estimate = client.ingest.estimate(folder, workspace="my-workspace", level="BALANCED")
print(estimate.full_ingest["totals"])

def progress(sent: int, total: int | None) -> None:
    print(f"\rupload {sent}/{total or '?'} bytes", end="")

job = client.ingest.submit(folder, workspace="my-workspace", level="BALANCED",
                           run_postprocess=True, doc_workers=3, progress=progress).job
for ev in client.ingest.jobs.events(job.id):     # replays history, then tails until done
    print(ev.event, ev.progress, ev.message)
print(client.ingest.jobs.get(job.id).status)
```

Uploads are sent as `multipart/form-data` with the same field names the web client
uses: `workspace` and a repeated `files`, plus `level`, `run_postprocess`,
`doc_workers`, `document_prefix` and the rest of the documented options. An option
name that the route does not accept raises `TypeError` instead of being ignored.
Ingest requires the `data_manager` role on the workspace (or `admin`).

Concurrency of a run: `doc_workers` (documents in parallel, 1-64), `llm_concurrency` and
`vlm_concurrency` (calls per document, 1-1000); without them the server uses the Settings
`ingest.concurrency.*` (3 documents x 10 / 10). `client.ingest.plan(ws, doc_workers=…,
llm_concurrency=…)` answers `plan.concurrency`: per role the calls in flight (`effective`) and the
limit that binds (`run`, or the endpoint's `endpoint_ceiling`).

### One job for a large folder (uploads in several requests)

A long ingest is ONE server job: created with every run option (frozen on the job),
filled by as many upload requests as the server's limits need, then submitted with the
full manifest. Each file travels with its sha256 (computed locally); staging is
idempotent per name + sha256, so a retried or resumed upload never duplicates a file.

```python
from ragnarok_api import UploadInterrupted, ManifestMismatch

ref = load_or_create_my_ref()                     # e.g. ragnarok_api.new_client_ref(), persisted by you
try:
    result = client.ingest.upload_job(folder, workspace="my-workspace", level="BALANCED", client_ref=ref,
                                      progress=lambda p: print(p.batch, p.batches, p.staged_files))
except UploadInterrupted as exc:                 # a request kept failing: the job keeps what it staged
    result = client.ingest.upload_job(folder, workspace="my-workspace", client_ref=exc.client_ref)
for ev in client.ingest.events(result.job_id, after=0):   # after=<last seq> on a re-attach
    print(ev.step, ev.accounting)                # store_done / doc_skipped / doc_failed carry the counts
```

`client_ref` (generated when not given; on the result and on `UploadInterrupted`) names the
job for the caller: a later call with the same ref finds it (`job_by_client_ref`) and
continues it, and a create repeated after a lost response answers the same job
(`existing`), never a second one. With `job_id` (or a ref) of a job already submitted it
only re-attaches (`already_submitted=True`, nothing sent). The low-level calls are `limits()`,
`create_job()` (`submit=false`, `client_ref=`), `upload_files()`, `remove_files()`, `submit_job()` and
`uploading_jobs()`. Typed refusals: `ManifestMismatch` (`missing`, `unexpected`,
`sha256_differs`), `StagedNameConflict`, `ChecksumMismatch`, `StagingQuotaExceeded`
(`used_bytes`, `limit_bytes`), `FilesTooLarge` (a file no request can carry; nothing sent).
Counts come from the server only: `job.accounting` (pages, chunks, tables, figures,
documents) and `IngestDoneEvent.summary.accounting`.

A job found that already ended `failed`/`cancelled` (e.g. `upload_abandoned`) raises
`UploadJobFinished` (start over with a NEW `client_ref`); a `paused` one is re-attached
(`paused=True`). A repeated create with other options raises `ClientRefConflict`. Against a
server before resumable uploads, `upload_job` sends ONE one-shot request
(`legacy_single_upload=True`, `message`), or raises `ServerTooOldForUpload` when the files
cannot fit in one request.

### Persistent jobs

Jobs and their events live in MongoDB, so they outlive an API restart. Admins see every
job; other callers see the jobs they submitted.

```python
from ragnarok_api import Conflict, IngestDoneEvent, IngestErrorEvent

for job in client.ingest.jobs.list(workspace="my-workspace", status="running", limit=20):
    print(job.id, job.status, job.progress, job.owner_email)

client.ingest.jobs.cancel(job.id)                # cooperative: the job ends `cancelled`
try:
    client.ingest.jobs.cancel(job.id)
except Conflict as exc:                          # 409 job_not_running once it has finished
    print(exc.code, exc.detail["status"])

for ev in client.ingest.jobs.events(job.id):
    if isinstance(ev, IngestDoneEvent):          # compact: counts + the job path, not the result
        print(ev.success, ev.cancelled, ev.summary.documents_committed, ev.job_path)
    elif isinstance(ev, IngestErrorEvent):
        print(ev.code, ev.error_message)         # e.g. code="access_revoked"
full = client.ingest.jobs.get(job.id).result     # the full per-document result

report = client.workspaces.repair("my-workspace")              # admin; dry run
print(report.rollback_count, [(f.document_name, f.kind) for f in report.findings])
client.workspaces.repair("my-workspace", apply=True)           # roll back the leftovers
```

`client.ingest.job(id)` and `client.ingest.events(id)` remain as aliases.

### Ingest models

Per role (extraction, vision) a run uses the request's choice, else the model the workspace
remembers from its last run, else the Settings ingest profile. Ask the server before uploading:

```python
from ragnarok_api import BadRequest, Conflict

plan = client.ingest.plan("my-workspace", level="BALANCED",
                          semantic_extraction_llm_model="lm::qwen3",   # endpoint::model, optional
                          llm_reasoning_effort="low")                  # also vlm_model, vlm_reasoning_effort
x = plan.extraction                               # roles[extraction_role]; plan.vlm is roles[vision_role]
print(x.model, x.source, x.reasoning_effort, x.status)   # source: request | remembered | profile
print(plan.embedding.model, plan.embedding.dimensions, plan.embedding.locked)
for notice in plan.run_notices:                   # the notices about this run's roles
    print("!", notice)

try:   # the same four fields are form options of estimate() and submit()
    client.ingest.submit(folder, workspace="my-workspace", vlm_model="gw::vision")
except BadRequest as exc:                        # 422 ingest_model_invalid: the server's reason
    print(exc.code, exc.message)

try:   # the embedder changes only while the workspace holds no document
    client.workspaces.set_embedding("new-workspace", "lm::embed-2", 768)
except Conflict as exc:                          # 409 embedding_locked
    print(exc.message)
```

`estimate.models`, the `data.models` of a job's `init` and `run_start` status events and
`job.result["models"]` have the same shape (`IngestPlan.model_validate(...)` reads them); a
`model_notice` status event announces a remembered model the server replaced.

## 3. Read the knowledge graph

```python
hits = client.graph.search("pembrolizumab", workspace="epar").results
entity = client.graph.entity(hits[0].id, workspace="epar", include_description=True, include_neighbors=True)
sub = client.graph.neighborhood(hits[0].id, workspace="epar", hops=2)
print(entity.entity["display_name"], len(sub.entities), len(sub.relationships))

chunks = client.search.chunks("overall survival", workspace="epar", limit=20).chunks
texts = client.sources.batch([c.chunk_id for c in chunks], workspace="epar").contents
pdf = client.sources.file(chunks[0].document_name, workspace="epar")   # bytes
```

Cursor-paged routes, such as the planned KG API, can use
`ragnarok_api.iterate(fetch_page, cursor_field="next_cursor")`.

## 4. Manage access

```python
from ragnarok_api import Client, Session, TermsRequired

client = Client("https://ragnarok.example")
try:
    session = Session.login(client, "admin@example.com", "secret")
except TermsRequired as exc:              # first login, or terms older than the re-accept window
    session = exc.session.accept_terms()  # explicit user act; the server issues a fresh token

me = client.auth.whoami()
print(me.email, me.account_type, me.is_admin, me.role_on("epar"))   # "admin", "owner", "viewer"... or None
print([(g.workspace, g.role) for g in me.grants])
print([(w.name, w.role) for w in client.workspaces.list()])          # your role on each workspace

# Admin: accounts with per-workspace roles (viewer < data_manager < owner)
client.admin.create_user("analyst@example.com", "a-long-password", grants=[("epar", "viewer")])
client.admin.update_user("analyst@example.com", grants=[("epar", "viewer"), ("has", "data_manager")])

# Owner session, or admin credential allowed by the server privileged-write policy
grants = client.workspaces.grants.list("epar")
print(grants.your_role, [(g.email, g.role) for g in grants.grants], grants.admins)
change = client.workspaces.grants.put("epar", "analyst@example.com", "data_manager")
print(change.previous_role, "->", change.role, change.changed)
client.workspaces.grants.delete("epar", "analyst@example.com")
print(client.workspaces.meta("epar").owners)                           # data_manager or more

client.admin.deactivate_user("analyst@example.com")
session.logout()                          # revokes the token server-side
```

Roles are per workspace. An admin holds every role everywhere and has
no grants. An owner may grant `viewer` or `data_manager` only, so granting `owner` raises
`Forbidden` with `code="role_ceiling"`. Other refusals are `admin_only_workspace`,
`self_grant` and `owner_grant`. A route above your role raises
`Forbidden(code="workspace_role_required")`, whose `detail` holds `required` and `role`.
A concurrent grant change raises `Conflict(code="concurrent_update")`.
`allowed_workspaces` is deprecated: pass `grants`, and never both (422
`grants_conflict`).

`Token` and `Session` never show their secret in `repr()`. The password is sent once
and is not stored.

## Errors

All errors subclass `RagnarokError` and carry `status`, `code`, `detail` and
`request_id` (from `x-request-id`, else the error body's `request_id`):

| Error | When |
|---|---|
| `Unauthorized` | 401 |
| `Forbidden` | 403. `code` gives the reason, e.g. `model_not_allowed`, `workspace_role_required`, `role_ceiling`, `admin_only_workspace`, `session_required` |
| `TermsRequired` | 403 `terms_required`, or a login that needs terms acceptance. Carries `.session` |
| `EmbeddingMismatch` | 403 `embedding_mismatch` |
| `NotFound` / `Conflict` / `BadRequest` | 404 / 409 (e.g. `job_not_running`, `concurrent_update`) / 400 and 422 |
| `RateLimited` | 429. `retry_after` is in seconds |
| `ServerError` | 5xx |
| `TransportError` | no HTTP response (connection refused or reset, timeout) |
| `ApiVersionMismatch` | the server's API major is not 1 |
| `QueryFailed` | the query stream emitted `error`. Carries `.error` and `.result` |

Retries: only idempotent GETs, and only on connection failures (at most 2 retries,
with jitter). The SDK does no caching.

## Development tests

From the repository root:

```sh
python -m pip install -e './sdks/python[test]'
python -m pytest -c sdks/python/pyproject.toml sdks/python/tests -q
```

These tests use recorded fixtures and mocked HTTP transports; no server or database is needed.

## License

MIT

Query modes are `talk`, `quick`, `balanced` and `deepdive`; availability and authorization are enforced by the server.
