Manage Libraries: create, upload, and verify ingestion

Create a Library, upload a document, and wait until it is readable. Task attachments are also stored in a Library, under a path for that Task.

Before you manage a Library

Complete Install the SDK before you run this workflow.

Before the first call:

• Set DUALEAI_TOKEN to an API token for an Agent Identity, and DUALEAI_AGENT_ID to that Agent Identity’s public identifier.
• Set DUALEAI_TENANT_ID to your Tenant’s identifier.
• The example creates its Library under agent/<agent_id>/. There, the supplied Agent default lets every Agent Identity start uploads and create, write, and read Libraries. For another path, the Agent Identity needs library:write on it to create the Library and upload, plus library:read to poll. Access and isolation explains how policies and Grants supply them.
• Put a non-empty, synthetic contract.pdf file beside the script below.
• Check Library limits for the current accepted file types and upload limits.
• Assign a Human Identity with library:delete and valid aal3 authentication in the Dashboard before creating the Library. An API token cannot delete the whole Library.

This workflow can create persistent storage and recorded usage under your agreement; see Pricing and access.

Create a Library and upload a document

create() needs library:write on the requested path. If an active Library already uses that path, it returns that Library; otherwise, it creates one. It adds no Grant: the policies and Grants that already cover the path decide who reaches the new Library.

More than one Library can have the same path, and they share that path’s access. Use the Library ID to identify one. Deleted Libraries do not reserve a path. The SDK refuses a path that contains {, }, or *.

import asyncio
from pathlib import Path

from dualeai import (
    LibraryCreateRequest,
    LibraryDocumentGetRequest,
    LibraryResponseDocumentStatus,
    create_sdk,
)


async def main() -> None:
    async with create_sdk() as sdk:
        if sdk.agent_id is None:
            raise SystemExit("Set DUALEAI_AGENT_ID to the token's Agent Identity ID.")
        create_request = LibraryCreateRequest(path=f"agent/{sdk.agent_id}/contracts")
        library = await sdk.libraries.create(create_request)
        print(f"library_id={library.id}")

        attachments = sdk.prepare_attachments(
            [(Path("contract.pdf"), "Synthetic example contract")],
        )
        receipts = await sdk.libraries.upload(library.id, attachments)
        receipt = receipts[attachments[0].key]
        print(f"document_id={receipt.document_id}")

        document = await sdk.libraries.wait_for_document(
            LibraryDocumentGetRequest(
                library_id=library.id,
                document_id=receipt.document_id,
            ),
        )
        if document.status is LibraryResponseDocumentStatus.failed:
            print(document.failure)
            return

        print(f"status={document.status.value}")


asyncio.run(main())

Save the example as manage_library.py, keep contract.pdf beside it, then run:

python manage_library.py

Verify document ingestion

A successful run prints the Library id, document id, and terminal status:

library_id=0192f7c1-6e2a-7c31-9a4d-2b6f8c1d4e90
document_id=0192f7d9-712b-7c31-8f08-845c30a42a31
status=ready

The script prints both IDs before polling. Save them so you can resume checking the same document after a timeout.

PublicIndexedDocument includes progress_pct, page_count, tags, and optional failure details.

The create request sends only path; Duale AI returns the Library ID as library.id. Use that ID for later operations.

A lost response or a 5xx during creation can leave the Library unfinished. An unfinished Library is absent from list() and cannot be read yet. Repeat sdk.libraries.create(create_request) with the same path: any caller with library:write on that path can complete it. The call returns the Library at that path, or creates one if none remains there.

Read problem_details and correct invalid input or missing access before retrying. A rename, deletion, or access change between attempts can change which Library create() returns.

upload() keys receipts by PreparedAttachment.key. Each receipt includes library_id, a new document_id, and queued status. Use those IDs with SDK methods; ignore the receipt’s location field.

wait_for_document() polls until ready or failed. Its default timeout covers the whole polling operation; wait_for_document gives the interval and the default. Set a positive timeout= to change it.

• Allow more time when needed. Processing can take longer than the default timeout; Duale AI does not guarantee when it will finish. See Library limits.
• A TimeoutError stops the client wait. It neither cancels processing nor predicts success.
• Resume with get_document(). Poll the same document to a terminal status. Another upload() creates a duplicate.

Handle ingestion failure

Upload success does not prove that the document becomes readable. A file that fails the integrity, malware, or extraction checks makes wait_for_document() return a failed document, whose failure details can be None. See Document ingestion failures for codes and recovery.

Recover a partial upload

A batch can partly succeed. On LibraryUploadError, save completed_receipts and the failed attachment_key, then poll the existing documents. For other errors, list the Library’s documents to identify missing files before deciding what to upload again. Repeating the batch creates duplicates.

Read every page of documents

list_documents() returns one page. Pass next_cursor unchanged into the next request until it is None. This excerpt uses the open sdk and library from the earlier example:

from dualeai import LibraryDocumentListRequest

cursor = None
while True:
    page = await sdk.libraries.list_documents(
        LibraryDocumentListRequest(library_id=library.id, limit=100, cursor=cursor),
    )
    for document in page.documents:
        print(document.document_id, document.status.value)
    cursor = page.next_cursor
    if cursor is None:
        break

Remove the example data

Have the cleanup owner delete the example Library in the Dashboard when the files are no longer needed. Whole-Library deletion makes its documents unrecoverable, and an API token cannot perform it. To remove only one document while keeping the Library, use sdk.libraries.delete_document() with library:delete; the document enters Recently deleted. Security and data handling explains the effects of each deletion and the restore window.

Continue with other Library operations

Troubleshoot the SDK covers authorization and connection failures.