ContextBase Python SDK

The Python client for ContextBase, a platform for retrieval-augmented generation over your own documents.

Install the preview release from PyPI with

pip install contextbase-sdk==2026.1.0rc1

Once the stable release is on PyPI, install it with

pip install contextbase-sdk

The distribution is named contextbase-sdk; the package you import is contextbase. Python 3.11 or newer.

The source repository and its issue tracker require GitHub access during this preview.

Quick start

A key is minted by a person in the web interface (or through POST /api/v1/auth/api-keys/). A key that creates a space is granted manage on it, so one key is enough for everything below. Set the key and the server's address, then run this as it stands:

export CONTEXTBASE_API_KEY=cb_...
export CONTEXTBASE_BASE_URL=https://contextbase.example.com

from contextbase import ContextBase, Done, ErrorEvent, Token

with ContextBase() as client:
    client.check_compatibility()

    space = client.spaces.create(
        'Quick start',
        llm_config={'model': 'openai/gpt-4o'},
    )
    try:
        document = space.documents.upload(
            b'The capital of France is Paris.',
            filename='facts.txt',
        )
        space.documents.wait_indexed([document], timeout=120)

        answer = space.query('What is the capital of France?')
        print('answer:', answer.answer)
        print('sources:', [source.text for source in answer.sources])

        for event in space.query_stream('What is the capital of France?'):
            if isinstance(event, Token):
                print(event.delta, end='', flush=True)
            elif isinstance(event, Done):
                print('\nstreamed:', len(event.answer.sources), 'sources')
            elif isinstance(event, ErrorEvent):
                print('\nstream failed:', event.detail)

        for source in space.search('capital', top_k=3):
            print('found:', source.text)
    finally:
        space.delete()

The same with asyncio: AsyncContextBase has every call above, awaited, and async for over list() and query_stream().

import asyncio

from contextbase import AsyncContextBase


async def main() -> None:
    async with AsyncContextBase() as client:
        async for space in client.spaces.list():
            print(space.id, space.name)


asyncio.run(main())

Configuring a space and evaluating it

Reading a space's pipeline needs any grant on it; checking, applying and rebuilding it, and evaluating the space, need manage, which a key that created the space has. A key granted less is refused with PermissionDeniedError. The catalogue and the presets are readable with any key, and evaluation datasets belong to the key's owner.

from contextbase import ContextBase
from contextbase.types import PipelineConfigStatus


def show(status: PipelineConfigStatus) -> None:
    states = ', '.join(f'{i.kind} {i.status}' for i in status.representations)
    print('rebuild:', states)


with ContextBase() as client:
    print('presets:', [preset.id for preset in client.pipeline.presets()])
    space = client.spaces.create(
        'Evaluated',
        llm_config={'model': 'openai/gpt-4o'},
    )
    try:
        space.documents.upload(
            b'The capital of France is Paris.',
            filename='facts.txt',
        )
        hybrid = {
            **space.config.get().to_dict(),
            'retrievers': [{'node': 'bm25'}, {'node': 'dense'}],
            'fusion': {'node': 'rrf'},
        }
        report = space.config.validate(hybrid)
        print('preflight:', report.status)

        change = space.config.apply(hybrid, on_progress=show)
        print('applied:', change.result.scope, 'stale:', change.stale)

        dataset = client.eval_datasets.create(
            'Capitals',
            corpus=[{'id': 'fr', 'text': 'The capital of France is Paris.'}],
            queries=[
                {
                    'question': 'What is the capital of France?',
                    'relevant_ids': ['fr'],
                },
            ],
        )
        run = space.evaluations.run(
            dataset,
            on_progress=lambda run: print(
                'run:', run.status, run.progress.stage
            ),
        )
        print('metrics:', run.metrics)
        for outcome in space.evaluations.queries(run):
            print(outcome.position, outcome.question, outcome.answer)
    finally:
        space.delete()

config.apply and config.reindex wait for the space's indexes to be rebuilt, unless wait=False. A worker takes the rebuild up some time after the change is accepted, and until then the space's status still describes the indexes as they were, so the wait is in two steps: up to start_timeout seconds (60 by default) for the rebuild to begin -- RebuildNotStartedError if it does not -- and then up to timeout for nothing to be building -- RebuildTimeoutError if something still is. Indexes left out of date with no worker on them do not hold the wait up; change.stale names them, and a config.reindex() rebuilds them. A change over storage this platform did not make starts a pass over that storage, reported as change.storage_pass and not waited for; the space's storage index, through client.raw, follows it.

A configuration its preflight checks refuse raises PreflightBlockedError carrying their report -- as PreflightConflictError (a ConflictError) for 409 or PreflightUnprocessableError (an UnprocessableEntityError) for 422. evaluations.run waits for the run: a failed one raises EvaluationFailedError with the server's reason, one still going at timeout raises EvaluationTimeoutError naming its stage and how far it got. The timeouts derive from WaitTimeoutError, as does IndexingTimeoutError from documents.wait_indexed.

Anything else the server offers -- the knowledge graph, a space's metadata schema, its storage index, the chunk map, the agent graph -- is reached through the generated operations with client.raw:

from uuid import UUID

from contextbase import ContextBase
from contextbase._generated.api.graph import get_graph_status

with ContextBase() as client:
    space = next(iter(client.spaces.list()))
    print(get_graph_status.sync(UUID(space.id), client=client.raw))

Downloading an original

Use the generated operation through client.raw. A successful response is a File; read its payload to get the original bytes. The detailed variant also provides the response headers and status. Error responses retain their generated JSON models.

from uuid import UUID

from contextbase._generated.api.documents import download_document
from contextbase._generated.types import File

original = download_document.sync(
    UUID(space.id),
    UUID(document.id),
    client=client.raw,
)
if isinstance(original, File):
    original_bytes = original.payload.read()

For an asynchronous client, await download_document.asyncio with the same arguments. The SDK does not write a path based on the attachment filename.

Typed inputs, or plain ones

A filter or a configuration is either a model from contextbase.types, which an editor completes field by field, or a mapping of the same shape; both send the same request. A model carries its defaults, so it may send a field the mapping leaves to the server.

from contextbase import ContextBase
from contextbase.types import (
    MetadataCondition,
    MetadataFilter,
    MetadataFilterJoin,
    MetadataFilterOperator,
    SpaceLLMConfig,
)

with ContextBase() as client:
    space = client.spaces.create(
        'Typed',
        llm_config=SpaceLLMConfig(model='openai/gpt-4o', temperature=0.2),
    )
    try:
        support_only = MetadataFilter(
            join=MetadataFilterJoin.ALL,
            conditions=[
                MetadataCondition(
                    field='team',
                    operator=MetadataFilterOperator.EQ,
                    value='support',
                ),
            ],
        )
        same_as_a_mapping = {
            'join': 'all',
            'conditions': [
                {'field': 'team', 'operator': 'eq', 'value': 'support'},
            ],
        }
        for narrowing in (support_only, same_as_a_mapping):
            print(len(space.search('refunds', metadata_filter=narrowing)))
    finally:
        space.delete()

An id is taken as a string or a uuid.UUID, wherever a call takes one.

What the client does for you

• Keys and addresses. ContextBase(api_key=..., base_url=...), or CONTEXTBASE_API_KEY / CONTEXTBASE_BASE_URL. Without a key only meta() and check_compatibility() work; everything else raises MissingAPIKeyError before sending anything.
• Timeouts. 60 seconds a request unless timeout= says otherwise. A streamed answer is held to it only while connecting.
• Repeats. A GET, PUT or DELETE the server answers with 429 or 503 is asked again, up to max_retries=2 times, waiting as Retry-After asks or doubling from half a second. A question (POST) is never asked twice.
• Listings. client.spaces.list() and space.documents.list() walk every page, fetching the next only when it is read.
• Streams. space.query_stream() yields Stage, Check, Token, then Done or ErrorEvent. Stage and check events retain attempt_id; checks also identify node_id, phase, verdict and action. Unknown event names are ignored for compatibility with newer servers.
• Publication. answer.outcome is accepted or blocked, for buffered queries and Done.answer alike. Check it before treating an empty answer as success. query_id locates the authorized inspection record. postgeneration carries the server's redacted execution summary; rejected drafts and judge explanations remain behind the server's content-access controls. Old servers that omit these fields default to accepted, None and None.
• Errors. What the server or the connection causes derives from ContextBaseError; arguments no request can be made from (no address, bytes with no filename) raise ValueError. An error status is an APIError of its own class -- AuthenticationError (401), PermissionDeniedError (403), NotFoundError (404), ConflictError (409), PayloadTooLargeError (413), RateLimitError (429), ServerError (5xx) and the rest -- carrying status_code, detail (the server's sentences), reason (its machine name for the refusal) and names. A 404 for a path the server has no route for is an UnknownEndpointError, whose message suggests the server is older than this client. Not reaching the server is an APIConnectionError; wait_indexed running out of time is an IndexingTimeoutError holding each document's last state.
• Compatibility. client.check_compatibility() raises IncompatibleServerError for a server older than MINIMUM_SERVER_VERSION or speaking another API version. That version is carried in the package rather than in the distribution's metadata, which has no field for it.
• Escape hatch. client.raw is the generated client underneath, sharing the client's connection, key and timeout, for an operation the client does not cover. It is typed as the generated operations take it, so list_spaces.sync(client=client.raw) type-checks. Without a key there is no authenticated client to hand over, and asking for it raises MissingAPIKeyError.
• Changing a space's settings is left to a person: the server refuses it to every key, so the client has no call for it. The generated operation is contextbase._generated.api.spaces.update_space, through client.raw.

Developing the SDK

Everything below runs from sdk/python/ and goes through one entry point, scripts/sdk.py. The SDK has its own environment, managed by uv (0.5.18 or newer), and shares nothing with the server's Poetry environment.

uv sync                               # create the environment from uv.lock
uv run scripts/sdk.py gen             # regenerate the derived sources
uv run scripts/sdk.py check-stale     # refuse derived sources that are out of date
uv run scripts/sdk.py lint            # ruff, the formatter's check, mypy
uv run scripts/sdk.py test            # the unit suite, 100 % coverage
uv run scripts/sdk.py build           # sdist and wheel into dist/
uv run scripts/sdk.py smoke-wheel     # build, install into a fresh environment, call meta()
uv run scripts/sdk.py bump-check      # refuse a version that understates a breaking change
uv run scripts/sdk.py release         # see below

What is written and what is derived

• src/contextbase/_generated/ is generated by openapi-python-client from openapi/openapi.json at the repository root, the same snapshot the web client is generated from. Never edit it by hand. Binary response construction is customised by the file-property template in templates/, so a binary schema uses the HTTP body even when its media type includes JSON. The template preserves the generator's file input handling. The generator is pinned in pyproject.toml; its configuration is openapi-python-client.json.
• src/contextbase/_async/ is the client, written by hand.
• src/contextbase/_sync/ is produced from _async/ by unasync. Edit _async/ and run gen; the names renamed beyond unasync's own list are in SYNC_NAMES in scripts/sdk.py. Docstrings are copied across unchanged, so prose in _async/ is written to be true of both clients.

Both derived directories are committed. check-stale refuses a package whose derived directories are not what a regeneration writes; how it compares is described at command_check_stale in scripts/sdk.py. The repository's gate (docker/django/ci.sh) runs it together with lint and test. When the server's API changes, the refusal of scripts/check_openapi_snapshot.py at the repository root says how to refresh the snapshot and what to regenerate with it.

oasdiff

bump-check judges schema changes with oasdiff, and only with release 1.32.1: another release may classify a change differently, so the command refuses any other. Install it with either of

brew install oasdiff                                  # while Homebrew ships 1.32.1
go install github.com/oasdiff/oasdiff@v1.32.1

Versions and releases

The SDK is versioned independently of the server. The version the next release will carry is version in pyproject.toml; a release is the commit tagged sdk-py/v<that version>.

bump-check compares openapi/openapi.json with the snapshot committed at the newest final release tag older than the current version (pre-releases and development releases promise nothing, so they are passed over). If oasdiff reports a breaking change, the version must raise the major number -- before 1.0, the minor. With no earlier final release there is nothing to compare, and any version passes.

release runs, in order and stopping at the first refusal:

1. the tag at HEAD (or the one named by --tag) points at HEAD and names the version in pyproject.toml;
2. sdk/python/ and openapi/ hold no change git status reports, since the build and bump-check read the working tree and the release is HEAD;
3. bump-check;
4. test;
5. build;
6. uv publish.

The upload goes where --to says (testpypi or pypi); without it, to UV_PUBLISH_URL if that is set, and otherwise to TestPyPI. --to wins over UV_PUBLISH_URL, and the release says so when the two disagree. The token is read from UV_PUBLISH_TOKEN, and a release without one is refused before anything runs. --dry-run runs every step but the upload.

UV_PUBLISH_TOKEN=pypi-... uv run scripts/sdk.py release    # to TestPyPI
uv run scripts/sdk.py release --dry-run                    # stop before uploading

Record every release in CHANGELOG.md.

Supported Python versions

3.11 through 3.14. The gate runs the suite under the version .python-version names; the oldest and the newest supported are checked by hand, each in an environment kept outside the checkout, and by installing the wheel under it:

UV_PROJECT_ENVIRONMENT=/tmp/sdk-py311 uv run --frozen --python 3.11 scripts/sdk.py test
UV_PROJECT_ENVIRONMENT=/tmp/sdk-py314 uv run --frozen --python 3.14 scripts/sdk.py test
uv run scripts/sdk.py smoke-wheel --python 3.11
uv run scripts/sdk.py smoke-wheel --python 3.14

Licence

Apache License 2.0; see LICENSE.

Applications and reference

See examples/README.md in the source repository for the executable notebook, document ingestion workflow, FastAPI service, and LangChain/LlamaIndex retrievers. The examples' dependencies live in the pinned development group; they do not add framework dependencies to the SDK runtime. The SDK test command runs their offline HTTP smoke tests as part of the repository gate.

The Sphinx site's SDK guide includes these source files directly. Its Python reference is generated from the public SDK and its REST reference renders the committed OpenAPI snapshot. Regenerate the Python reference with uv --directory sdk/python run --frozen scripts/examples_reference.py --write. The SDK suite refuses a stale committed reference. Build the site from the repository root with poetry run sphinx-build -W -b html docs /tmp/contextbase-docs.

Canonical extraction volume

Document.extracted_char_count is the server's exact Unicode codepoint count of canonical extracted text, including whitespace before chunking. It is independent of the original byte size, chunk overlap, and indexing status. A confirmed empty extraction has 0; unknown legacy extraction has null. Python represents an absent field from an older server as UNSET (from contextbase._generated.types), preserving it separately from None. Neither unknown nor absent values should be treated as zero. The field is server-owned: owner metadata with the same key does not change it.

A search or query source exposes document_id independently of locator. It is the canonical document UUID when the server verified the returned leaf text against its Space-scoped canonical chunk. Missing offsets do not prevent known identity, and known identity does not manufacture a locator, page or section. Unknown provenance remains null (older servers may omit the field). Do not parse document identity from source id, metadata or text. Parent text and graph context remain separate evidence; this UUID identifies the leaf.

Sources may carry source.bibliography and source.parent_bibliography even when no verified locator is available. Their optional title is the canonical contextual document title; author, year and url are document-owner metadata, not extractor-verified facts or model attributions:

from contextbase import Answer


def show_bibliography(answer: Answer) -> None:
    for source in answer.sources:
        citation = source.bibliography
        if citation is not None:
            print(citation.title, citation.author, citation.year, citation.url)

The URL is an owner-supplied bibliographic display reference. Syntax checks cannot guarantee public access or absence of secrets in its contents, and known original/media routes are recognised only on configured backend/storage origins. Authenticated originals remain on the generated download_document API described above; bibliography does not grant file access. Older responses without bibliography decode as None, and unknown metadata fields are not included in source bibliography.

Tracking document deletion

documents.delete accepts deletion; its return does not confirm physical cleanup. For completion tracking, keep the operation ID returned by the explicit request method. The operation remains readable by its space and ID after the document is absent. An optional command UUID identifies a repeated deletion command.

operation_id = space.documents.request_delete(
    document.id, command_id=command_id
)
status = space.documents.get_operation(operation_id)
print(status.state)

A missing or malformed operation header raises APIError after acceptance; the client does not repeat DELETE to obtain another ID.

Document lifecycle commands

activate, deactivate and reindex accept a required caller-owned command UUID and return DocumentOperationStatus. Keep that UUID for an explicit replay of the same command; using it for another action is a conflict. POST commands are never retried automatically, including after a 503 response.

from uuid import uuid4

command_id = uuid4()
operation = space.documents.deactivate(document.id, command_id=command_id)
status = space.documents.get_operation(operation.id)

The asynchronous methods take the same arguments and are awaited. Commands need write access; reading retained status needs read access. An attached storage document is refused with a conflict; another space's document is masked as not found. succeeded, failed and cancelled are terminal status values. A document's required is_active describes settled managed visibility; accepting a queued command does not flip it. indexed describes ingestion status independently. Existing delete and request_delete remain available.