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.

pip install -e ./sdks/python

Connect

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

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:

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

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.

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.

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:

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

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

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:

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.