DocForge SDK — INGESTION, FORGED

The typed Python client for DocForge — async and sync, fully type-hinted, zero server-tree dependency.

docforge-sdk is a typed Python client for the DocForge REST API — a document-intelligence platform that melts any document (PDF, Office, images…) into a canonical intermediate representation, enriches and chunks it, embeds it, and serves hybrid retrieval over it.

The SDK ships both an asynchronous and a synchronous client with an identical surface, is fully type-hinted (py.typed, Pydantic v2 models), and has zero dependency on the DocForge server tree — it is a clean-room client that talks to the API over HTTP only (httpx + pydantic + hand-written models mirroring the public REST contract), so it can be vendored or published independently.

✅ Async + sync — the same methods with and without await.
✅ Typed end to end — request/response Pydantic models, no dict-spelunking.
✅ Context-managed — connection pooling via async with / with.
✅ Typed errors — one exception hierarchy for connection, auth, 404, 409, 422…
✅ Tiny footprint — only httpx and pydantic.

Install
Requirements
Quickstart
- Async
- Sync
Authentication
Resources & methods
Recipes
Error handling
Type hints & discoverability
Versioning & compatibility
License
Publishing (maintainers)

Install

pip install docforge-sdk

Requirements

Python 3.12+
A running DocForge API (self-hosted). The examples assume http://localhost:10040.
An API token when the server has auth enabled (see Authentication).

Quickstart

Async

import asyncio

from docforge_sdk import AsyncClient, SearchRequest


async def main() -> None:
    async with AsyncClient("http://localhost:10040", api_token="df_root_...") as client:
        # Liveness
        print(await client.health.ping())

        # List collections
        collections = await client.collections.list()
        for c in collections:
            print(c.id, c.name)

        # Search the first one
        hits = await client.search.search(
            collections[0].id,
            SearchRequest(query="quarterly revenue", limit=5),
        )
        for hit in hits.hits:
            print(f"{hit.score:.3f}  {hit.text[:80]}")


asyncio.run(main())

Sync

The synchronous client is the async one without await — same resources, same signatures.

from docforge_sdk import Client, SearchRequest

with Client("http://localhost:10040", api_token="df_root_...") as client:
    collections = client.collections.list()
    hits = client.search.search(
        collections[0].id,
        SearchRequest(query="quarterly revenue", limit=5),
    )
    for hit in hits.hits:
        print(hit.score, hit.text)

Both clients accept the same constructor:

AsyncClient(base_url: str, timeout: float = 30.0, api_token: str = "")
Client(base_url: str, timeout: float = 30.0, api_token: str = "")

Argument	Default	Meaning
`base_url`	—	API origin, e.g. `"http://localhost:10040"`.
`timeout`	`30.0`	Per-request timeout in seconds.
`api_token`	`""`	Bearer token; empty means unauthenticated requests.

Always use the client as a context manager (async with / with) so the underlying HTTP connection pool is opened and closed cleanly. You can construct it directly, but then you own await client.aclose() / client.close().

Authentication

When the DocForge server runs with auth enabled, pass a bearer token as api_token. The token is sent as Authorization: Bearer <token> on every request.

client = AsyncClient("https://docforge.example.com", api_token="df_...")

Keys are scoped: a key carries a set of capabilities (read, write, search, create, admin) and an optional allow-list of collection ids. The root token created at server bootstrap has full access; mint narrower keys with client.auth (see Manage API keys).

A key holding create may create new collections, and is auto-granted ownership of what it creates: the new collection's id is appended to that key's own scope, so a single key can be given "may create collections + full power over the ones it creates" without knowing ids in advance — the natural setup for driving DocForge from an agent (e.g. over MCP). Pair it with a narrow, search-only key scoped to one collection for your app's runtime.

Resources & methods

Every resource hangs off the client (client.<resource>.<method>(...)). The async and sync surfaces are identical.

`health`

Method	Returns	Purpose
`ping()`	`HealthStatus`	Liveness probe.

`collections`

Method	Returns	Purpose
`list()`	`list[CollectionListItem]`	Every collection (schema + pipelines), each with its server-computed health summary.
`get(collection_id)`	`CollectionModel`	One collection (schema + pipelines).
`create(CreateCollectionRequest)`	`CollectionModel`	Create a collection (contract).
`update(collection_id, UpdateCollectionRequest)`	`CollectionModel`	Patch name / formats / fields / pipelines / cost-estimate overrides.
`delete(collection_id)`	`None`	Delete a collection.
`health(collection_id)`	`CollectionHealthResponse`	On-demand operational health — 5-state verdict, provider reachability sweep, index stats. No job enqueued, no spend.
`storage(collection_id)`	`CollectionStorageResponse`	Material storage footprint per store (S3 exact, Postgres/Qdrant estimated) + per-document breakdown.
`estimate(collection_id, scope="pending", document_ids=None, filter=None)`	`CostEstimate`	Pre-hoc cost + volume dry-run over the whole collection (`scope`), a selected subset (`document_ids`), or a corpus `filter`. No spend. Per-collection rate/assumption overrides apply.
`contract_schema()`	`CollectionContractSchemaResponse`	The full contract vocabulary to build a valid create/update: identity/limits JSON Schema, the metadata FieldSpec schema (field_type/origin/scope enums), and the accepted `supported_formats` tokens.

`documents`

Method	Returns	Purpose
`upload(collection_id, file, metadata=None, filename=None)`	`UploadAccepted`	Admit a document; returns the ingestion `job_id`. `file` is a path, `bytes`, or a `Path`.
`set_enabled(document_id, enabled)`	`DocumentEnabledResponse`	Reversibly hide/show a document from search.
`get_markdown(document_id, download=False)`	`DocumentView`	The document rendered as Markdown from the canonical IR (a generated view).
`get_html(document_id, download=False)`	`DocumentView`	The document rendered as HTML from the canonical IR (a generated view).
`reingest(document_id, force=False)`	`UploadAccepted`	Re-run the full ingestion of a single document.

`search`

Method	Returns	Purpose
`search(collection_id, SearchRequest)`	`SearchResponse`	Hybrid (dense + sparse RRF) search, optional rerank.

`explorer`

Method	Returns	Purpose
`list_documents(collection_id)`	`list[DocumentListItem]`	Documents in a collection.
`get_document(document_id)`	`DocumentDetail`	One document's detail.
`get_pages(document_id)`	`list[PageInfo]`	Page-level info.
`get_ir(document_id)`	`DocumentIRModel`	The canonical IR (blocks, tables, figures, enrichments).
`get_chunks(document_id)`	`list[ChunkInfo]`	The document's chunks.
`delete_document(document_id)`	`None`	Delete a document.
`set_chunk_enabled(chunk_id, enabled)`	`ChunkEnabledResult`	Toggle one chunk in/out of search.
`set_chunks_enabled(BulkChunkEnabledPatch)`	`BulkChunkEnabledResponse`	Toggle many chunks at once.

`jobs`

Method	Returns	Purpose
`list(collection_id)`	`list[JobStatus]`	Ingestion jobs for a collection.
`get(job_id)`	`JobStatus`	One job's status + progress.
`get_events(job_id)`	`JobTrace`	Per-stage event trace.
`live_workers()`	`WorkersLive`	Currently active workers.
`cancel(job_id, force=False)`	`CancelResult`	Stop a job — cooperative by default, immediate with `force=True`.
`cost(collection_id)`	`CollectionCost`	Paid text-gen roll-up (tokens + USD).
`queue(collection_id=None)`	`QueueDepth`	Pending/running backlog (fleet-wide or per-collection).
`stage_durations(collection_id)`	`StageDurations`	Average per-stage wall-clock (ETA basis).

`blobs`

Method	Returns	Purpose
`get(content_hash)`	`BlobContent`	Fetch a content-addressed blob (bytes + media type).

`corpus`

Server-side document grid + bulk operations at scale.

Method	Returns	Purpose
`query(collection_id, DocumentQueryRequest)`	`DocumentQueryResponse`	Filtered/sorted/paginated page + total match count.
`bulk_delete(collection_id, DocumentSelector)`	`BulkDeleteResponse`	Delete by id-set XOR filter-minus-excludes.
`bulk_set_enabled(collection_id, DocumentSelector, enabled)`	`BulkEnabledResponse`	Bulk enable/disable searchability.
`bulk_reingest(collection_id, DocumentSelector, force=False)`	`BulkReingestResponse`	Bulk full re-ingest (capped fan-out).

`snippets`

Granular, config-only export/import of ONE collection slice — synchronous, secret-masked, versioned (saved as *.dfsnippet, distinct from the whole-collection .dcexport bundle).

Method	Returns	Purpose
`export(collection_id, kind)`	`CollectionSnippet`	Export the ingestion `pipeline`, the `search` graph, or the metadata `schema` as a portable snippet.
`apply(collection_id, kind, snippet)`	`SnippetImportResult`	Apply a snippet of that kind onto an existing collection (healed/validated like a PATCH).

`auth`

Method	Returns	Purpose
`create_key(name, permissions=None, expires_at=None)`	`CreatedKey`	Mint a key. The plaintext token is returned once.
`list_keys()`	`list[KeyInfo]`	Every key (metadata only, never the secret).
`rotate_key(key_id, ...)`	`CreatedKey`	Roll a key's secret.
`revoke_key(key_id)`	`None`	Revoke a key.
`whoami()`	`WhoAmI`	THIS token's own capabilities + collection scope.

`pipelines`

Discovery + design surface for the ingestion / search graphs (advanced).

Method	Returns	Purpose
`list_surfaces()`	`PipelineIndexResponse`	Available pipeline designs.
`get_design(key, full=True)`	`PipelineDesignResponse`	A design's palette + default blob.
`inspect(key, blob)`	`InspectResponse`	Validate a graph blob.
`edit(key, ...)`	`EditResponse`	Apply a structural edit to a blob.
`view_stages(key, blob)`	`StageViewResponse`	Compile a blob into the stage-rail view.
`apply_stage(key, ...)`	`StageApplyResponse`	Apply one stage-rail action.

Recipes

Create a collection

A collection is a contract: a metadata schema + an ingestion pipeline + a search pipeline. The pipelines default to the stock graphs when omitted.

from docforge_sdk import Client, CreateCollectionRequest, FieldSpec, FieldType

with Client("http://localhost:10040", api_token="df_...") as client:
    collection = client.collections.create(
        CreateCollectionRequest(
            name="reports",
            supported_formats=["pdf", "docx"],
            max_file_size_bytes=50 * 1024 * 1024,
            fields=[
                FieldSpec(field_name="year", field_type=FieldType.INTEGER, filterable=True),
                FieldSpec(field_name="team", field_type=FieldType.KEYWORD_LIST, filterable=True),
            ],
        )
    )
    print(collection.id)

Key CreateCollectionRequest fields beyond the schema: max_file_size_bytes (bytes) and job_timeout_seconds (float | None, seconds) — the whole-ingest-job wall-clock job timeout for that collection; None (the default) inherits the worker's global job-timeout default. Same field, same semantics on CollectionModel (read) and UpdateCollectionRequest (write; there, omitting it leaves the current value unchanged, a set value overrides it).

Upload a document and wait for ingestion

upload returns immediately with a job_id; ingestion runs asynchronously on the worker. Poll the job until it reaches a terminal state.

import time
from docforge_sdk import Client

with Client("http://localhost:10040", api_token="df_...") as client:
    accepted = client.documents.upload(
        collection_id,
        file="report-2024.pdf",
        metadata={"year": 2024, "team": ["finance"]},
    )

    while True:
        job = client.jobs.get(accepted.job_id)
        print(job.status, job.progress, job.current_stage)
        if job.status in {"done", "failed"}:
            break
        time.sleep(2)

    if job.status == "failed":
        raise RuntimeError(job.error)

The async variant is the same with await and asyncio.sleep.

Hybrid search with filters

SearchRequest drives dense + sparse fusion (RRF), optional late-interaction rerank, and metadata filtering.

from docforge_sdk import Client, SearchRequest

with Client("http://localhost:10040", api_token="df_...") as client:
    resp = client.search.search(
        collection_id,
        SearchRequest(
            query="revenue guidance for next fiscal year",
            limit=10,
            filters={"year": 2024, "team": "finance"},
        ),
    )
    for hit in resp.hits:
        print(f"{hit.score:.3f}  doc={hit.document_id}  {hit.text[:100]}")

Key SearchRequest fields: query (required), limit (1–100, default 10), filters (dict[str, Any]), search_in (list[SearchTarget] to pick which vectors to query).

Explore a document (pages, IR, chunks)

with Client("http://localhost:10040", api_token="df_...") as client:
    docs = client.explorer.list_documents(collection_id)
    doc = client.explorer.get_document(docs[0].document_id)

    ir = client.explorer.get_ir(doc.document_id)  # canonical IR
    chunks = client.explorer.get_chunks(doc.document_id)

    # Reversibly drop a noisy chunk out of search:
    client.explorer.set_chunk_enabled(chunks[0].chunk_id, enabled=False)

Measure a collection's storage footprint

storage reports the material footprint per store — S3 bytes are exact (deduped), Postgres/Qdrant bytes are estimates (each section flags this via its own estimated) — plus a per-document breakdown sorted heaviest first.

with Client("http://localhost:10040", api_token="df_...") as client:
    footprint = client.collections.storage(collection_id)
    print(footprint.grand_total_bytes, footprint.s3.physical_unique_bytes)
    for doc in footprint.documents[:5]:
        print(doc.filename, doc.total_bytes)

Manage API keys

Mint a scoped, expiring key (the plaintext is shown once, on creation):

from datetime import datetime, timedelta, timezone
from docforge_sdk import Client, Capability, KeyPermissions

with Client("http://localhost:10040", api_token="df_root_...") as client:
    created = client.auth.create_key(
        name="reporting-bot",
        permissions=KeyPermissions(
            capabilities=[Capability.READ, Capability.SEARCH],
            collections=[collection_id],  # empty list = all collections
        ),
        expires_at=datetime.now(timezone.utc) + timedelta(days=90),
    )
    print("SAVE THIS NOW:", created.key)  # plaintext, only returned once

    for key in client.auth.list_keys():
        print(key.id, key.name, key.last_used_at)

Error handling

Every failure raises a subclass of DocForgeError, so you can catch broadly or precisely.

from docforge_sdk import (
    Client,
    DocForgeError,
    APIConnectionError,
    APITimeoutError,
    APIStatusError,
    AuthError,
    NotFoundError,
    ConflictError,
    UnprocessableError,
)

with Client("http://localhost:10040", api_token="df_...") as client:
    try:
        client.collections.get("does-not-exist")
    except NotFoundError:
        ...  # 404
    except AuthError:
        ...  # 401 / 403 — bad or unscoped token
    except UnprocessableError as e:
        ...  # 422 — validation errors (see e.status / e.detail)
    except APITimeoutError:
        ...  # request exceeded `timeout`
    except APIConnectionError:
        ...  # server unreachable
    except DocForgeError:
        ...  # catch-all

Hierarchy:

DocForgeError
├── APIConnectionError
│   └── APITimeoutError
└── APIStatusError            # any non-2xx (carries .status)
    ├── AuthError             # 401 / 403
    ├── NotFoundError         # 404
    ├── ConflictError         # 409
    └── UnprocessableError    # 422

Type hints & discoverability

The package ships py.typed, so editors and type-checkers see every request/response type. All public models are re-exported from the top level:

from docforge_sdk import (
    # clients
    AsyncClient,
    Client,
    # collections
    CollectionModel,
    CreateCollectionRequest,
    UpdateCollectionRequest,
    FieldSpec,
    FieldType,
    CollectionStorageResponse,
    # documents / explorer
    UploadAccepted,
    DocumentDetail,
    DocumentListItem,
    ChunkInfo,
    PageInfo,
    DocumentIRModel,
    # search
    SearchRequest,
    SearchResponse,
    SearchHit,
    SearchTarget,
    # jobs
    JobStatus,
    JobTrace,
    JobEvent,
    WorkersLive,
    # auth
    Capability,
    KeyPermissions,
    CreateKeyRequest,
    CreatedKey,
    KeyInfo,
    # errors
    DocForgeError,
    AuthError,
    NotFoundError,
    ConflictError,
    UnprocessableError,
)

from docforge_sdk import * also works and pulls the full public surface.

Versioning & compatibility

The SDK tracks the DocForge REST contract; a CI parity gate diffs the SDK models against the live server's OpenAPI on every change, so a published version is coherent with the API it targets. Pin a version in production:

pip install "docforge-sdk==0.3.0"

License

MIT — see LICENSE. This SDK is deliberately licensed MIT even though the parent DocForge repository is GPLv3: it is a standalone, clean-room client (HTTP models only, no server code), so a permissive per-directory license is intentional and lets any project depend on it freely.

Publishing (maintainers)

Releases publish to PyPI via Trusted Publishing (OIDC) — there is no API token stored in the repo. To cut a release, tag a commit with the sdk-v<version> prefix (the version must match docforge_sdk/_version.py) and push the tag:

git tag sdk-v0.1.0
git push origin sdk-v0.1.0

The .github/workflows/release-sdk.yml workflow then builds and uploads the sdist + wheel.

One-time PyPI setup (done once by the maintainer, before the first release):

Reserve the project name docforge-sdk on PyPI.
Under the project's Publishing settings, add a GitHub trusted publisher with:
- Owner: Florian-BARRE
- Repository: docforge
- Workflow name: release-sdk.yml
- Environment: pypi

Development Status
- 4 - Beta
Intended Audience
- Developers
Operating System
- OS Independent
Programming Language
- Python :: 3
- Python :: 3.12
Topic
- Software Development :: Libraries
Typing
- Typed

Metadata

Development Status
- 4 - Beta
Intended Audience
- Developers
Operating System
- OS Independent
Programming Language
- Python :: 3
- Python :: 3.12
Topic
- Software Development :: Libraries
Typing
- Typed

Release files for docforge-sdk 0.31.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for docforge-sdk 0.31.0
File	Size	Uploaded	Provenance
docforge_sdk-0.31.0.tar.gz	251.3 kB	Oct 8, 2026	With provenance	Details

Built distribution (wheel)

Table of built distributions (wheels) for docforge-sdk 0.31.0
File	Interpreter	ABI	Platform	Reset
docforge_sdk-0.31.0-py3-none-any.whl 126.7 kB Oct 8, 2026 With provenance	Python 3	none	any	Details

Total release size: 377.9 kB

Release files / docforge_sdk-0.31.0.tar.gz

Download URL	docforge_sdk-0.31.0.tar.gz
Size	251.3 kB
Tags	Source
SHA-256 checksum How to use checksums	`8e3bb1fcf5b8dde0fdd4f127bf5c7ea2917f6b7694602ff18ca46ec490e469d1`
BLAKE2b-256 checksum How to use checksums	`e723ef842e958b29fea60a6ca334de32f28a458b7393dac3f284b687cb686d69`
Upload date	Oct 8, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	Yes
Uploaded via	`twine/7.0.0 CPython/3.13.14`

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

Identity

Publishing platform: GitHub Actions
Publishing repository: github.com/Florian-BARRE/DocForge
Publishing commit: github.com/Florian-BARRE/DocForge/tree/37e777a27d013a3b94ea33c7477ede251b82d375

Workflow

Publishing configuration: github.com/Florian-BARRE/DocForge/blob/37e777a27d013a3b94ea33c7477ede251b82d375/.github/workflows/release-sdk.yml
Publishing logs: github.com/Florian-BARRE/DocForge/actions/runs/37747691189/attempts/1

Artifact

Subject: docforge_sdk-0.31.0.tar.gz
SHA-256 checksum: 8e3bb1fcf5b8dde0fdd4f127bf5c7ea2917f6b7694602ff18ca46ec490e469d1 Verifying attestations

Release files / docforge_sdk-0.31.0-py3-none-any.whl

Download URL	docforge_sdk-0.31.0-py3-none-any.whl
Size	126.7 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`c885f9d94d4553b84dbd4cfaf164d0953132bfbbd1e771dcb46d73f055125dc2`
BLAKE2b-256 checksum How to use checksums	`cea8f7730a7d481f7f1ed29d484ff4cd9ddb44d785f403b35a73faa2d0f71fda`
Upload date	Oct 8, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	Yes
Uploaded via	`twine/7.0.0 CPython/3.13.14`

Provenance

Identity

Publishing platform: GitHub Actions
Publishing repository: github.com/Florian-BARRE/DocForge
Publishing commit: github.com/Florian-BARRE/DocForge/tree/37e777a27d013a3b94ea33c7477ede251b82d375

Workflow

Publishing configuration: github.com/Florian-BARRE/DocForge/blob/37e777a27d013a3b94ea33c7477ede251b82d375/.github/workflows/release-sdk.yml
Publishing logs: github.com/Florian-BARRE/DocForge/actions/runs/37747691189/attempts/1

Artifact

Subject: docforge_sdk-0.31.0-py3-none-any.whl
SHA-256 checksum: c885f9d94d4553b84dbd4cfaf164d0953132bfbbd1e771dcb46d73f055125dc2 Verifying attestations

Security for docforge-sdk 0.31.0

Report project as malware

Provenance

Full provenance

All 2 files in this release have been uploaded with provenance metadata.

Attestation for docforge_sdk-0.31.0.tar.gz

Identity

Publishing platform: GitHub Actions
Publishing repository: github.com/Florian-BARRE/DocForge
Publishing commit: github.com/Florian-BARRE/DocForge/tree/37e777a27d013a3b94ea33c7477ede251b82d375

Workflow

Publishing configuration: github.com/Florian-BARRE/DocForge/blob/37e777a27d013a3b94ea33c7477ede251b82d375/.github/workflows/release-sdk.yml
Publishing logs: github.com/Florian-BARRE/DocForge/actions/runs/37747691189/attempts/1

Artifact

Subject: docforge_sdk-0.31.0.tar.gz
SHA-256 checksum: 8e3bb1fcf5b8dde0fdd4f127bf5c7ea2917f6b7694602ff18ca46ec490e469d1 Verifying attestations

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.