Python SDK

Upload company documents, track processing, and query employee skills with the synchronous Python client.

Every call works in your company workspace, which is identified by your API key. Requires Python 3.9 or later.

Install

Install version 1.0.0 from the workerbee-sdk repository on GitHub:

python -m pip install "workerbee-sdk @ git+https://github.com/Workerbee-Inc/workerbee-sdk.git@sdk-v1.0.0#subdirectory=python"

The package isn’t on PyPI yet. The distribution name is workerbee-sdk and the import is workerbee_sdk. The PyPI package called workerbee is unrelated, so don’t install it. To upgrade, run the same command with a newer release tag and --upgrade.

Configure authentication

Create a key in Settings → Workspaces & SDK → SDK API keys. Upload-and-query examples need files:write, files:read, and graph:read. Restrict the key to Default or allow all company workspaces. Store the full key in your application’s secret manager or environment.

export WORKERBEE_API_KEY="YOUR_API_KEY"

A key restricted to workspaces other than Default can’t use these methods, and its restrictions aren’t widened automatically.

from workerbee_sdk import Workerbee

with Workerbee(timeout=60.0, max_retries=3) as wb:
    for file in wb.list_files():
        print(file.id, file.filename, file.status)

The client reads your key from WORKERBEE_API_KEY. timeout and processing-wait durations are in seconds. Use the context manager so the HTTP client is closed when you’re done.

Upload a resume and query skills

Save as quickstart.py, put a resume at ./resume.pdf, set the API key, and run python quickstart.py.

from workerbee_sdk import Message, Workerbee

with Workerbee() as wb:  # reads WORKERBEE_API_KEY
    resume = wb.upload_file(
        "./resume.pdf",
        document_types=["resume"],
        external_id="employee-123",
        source_system="hris",
        source_document_id="employee-123-resume",
        source_revision="v1",
        idempotency_key="employee-123-resume-v1",
    )
    print(resume.id, resume.status)
    resume.wait_until_ready(timeout=600)

    response = wb.chat(
        messages=[Message(role="user", content="List this employee's skills")],
        subject={"external_id": "employee-123", "source_system": "hris"},
        limit=20,
    )
    if response.refused:
        print(response.refusal_reason)
    elif response.clarification:
        print(response.clarification)
    else:
        print(response.interpretation)
        for row in response.rows:
            print(dict(row))
        print(response.references)

Document types and formats

from workerbee_sdk import Workerbee

with Workerbee() as wb:
    jd = wb.upload_file(
        "./job-description.pdf", document_types=["job_description"],
        external_id="requisition-456", source_system="hris",
    )
    review = wb.upload_file(
        "./review.pdf", document_types=["performance_review"],
        external_id="employee-123", source_system="hris",
    )
    print(jd.id, review.id)

Declare types in document_types, not inside metadata. Supported formats include PDF, DOCX, TXT, Markdown, JSON, and ZIP; SRT and VTT are accepted as text. JSON is document text, not an arbitrary graph-property payload. Legacy DOC files, scanned PDFs, audio and video aren’t supported.

Limits can differ by deployment. Nested archives and unsafe paths are rejected. In a mixed resume/JD document, provide typed subjects for employee and job identities. A mixed document is not a ZIP and does not trigger ZIP matching.

Source revisions and retries

Keep external_id and source_system stable across an employee’s uploads and queries. Names alone are not reliable identities. source_document_id identifies a source document; source_revision identifies an immutable version. Change the revision when bytes change; reusing one for different content returns SOURCE_REVISION_CONFLICT.

Use an idempotency_key for retries of the same upload request. Reusing it for different input is a conflict. The clients retry eligible safe requests and transient failures; unkeyed upload creation is not blindly retried.

ZIP uploads and matching

For mixed-document ZIP uploads, include workerbee-manifest.json at the archive root so Workerbee knows the document type and identity of each included file. Declare every document using its exact archive path, including subdirectories. An entry’s types replace the ZIP default; an entry omitted from the manifest inherits that default, so do not rely on filenames to identify resumes or job descriptions. The manifest itself is control information, not an ingested document.

{
  "files": {
    "resumes/employee-123.pdf": {
      "document_types": ["resume"],
      "external_id": "employee-123",
      "metadata": {"match_job_paths": ["jobs/platform-engineer.pdf"]}
    },
    "jobs/platform-engineer.pdf": {
      "document_types": ["job_description"],
      "external_id": "requisition-456"
    }
  }
}

hiring-bundle.zip
├── workerbee-manifest.json
├── resumes/employee-123.pdf
└── jobs/platform-engineer.pdf

The example below packages local ./resume.pdf and ./job-description.pdf under those paths, writes the manifest, then uploads the ZIP with the generic other default and hris source system. Per-file declarations select the resume and job-description processing. One invalid child does not discard successful siblings.

from workerbee_sdk import Workerbee
from workerbee_sdk.errors import FilePartialSuccessError
import json
from zipfile import ZipFile, ZIP_DEFLATED

manifest = {
    "files": {
        "resumes/employee-123.pdf": {
            "document_types": ["resume"],
            "external_id": "employee-123",
            "metadata": {"match_job_paths": ["jobs/platform-engineer.pdf"]},
        },
        "jobs/platform-engineer.pdf": {
            "document_types": ["job_description"],
            "external_id": "requisition-456",
        },
    }
}
with ZipFile("hiring-bundle.zip", "w", compression=ZIP_DEFLATED) as bundle:
    bundle.writestr("workerbee-manifest.json", json.dumps(manifest, indent=2))
    bundle.write("./resume.pdf", arcname="resumes/employee-123.pdf")
    bundle.write("./job-description.pdf", arcname="jobs/platform-engineer.pdf")

with Workerbee() as wb:
    archive = wb.upload_file(
        "./hiring-bundle.zip", document_types=["other"], source_system="hris",
    )
    try:
        archive.wait_until_ready(timeout=600)
    except FilePartialSuccessError as error:
        archive = error.file
    for child in archive.children_files():
        print(child.archive_path, child.id, child.status)
    for adapter in archive.adapters:
        print(adapter.document_type, adapter.status, adapter.error)

Other failures propagate; see error handling below.

With ZIP auto-matching enabled, a ZIP containing one JD and resumes adds successfully processed employees to that JD’s Internal Talent pool. Multiple JDs require explicit metadata.match_job_paths on each resume; paths must identify JDs in the same ZIP. Resume-only archives do not assign employees to a job.

Matching runs separately from ingestion. Inspect the parent’s archive_matching adapter: PENDING/RUNNING, NEEDS_REVIEW, or SUCCEEDED. Retryable scoring failures are recovered by the backend. A processed ZIP can still have matching in progress.

For individual uploads, open Workspaces & SDK → Employees, choose an employee and job, then select Match and open ranking. The SDK doesn’t expose employee-profile or direct job-matching methods; chat is not a replacement for a match operation.

Manage files

from workerbee_sdk import Workerbee

with Workerbee() as wb:
    for file in wb.list_files(status="PROCESSED", page_size=50):
        print(file.id, file.filename)
    file = wb.get_file("YOUR_FILE_ID")
    file.refresh()
    file.download("./downloaded-source.pdf")
    # Only when intended:
    # file.retry()
    # file.delete()

Listing paginates automatically. Use children_files() for ZIP entries. Retry failed or partially successful processing; successful adapters are preserved. Deleting a source removes access to its bytes but does not remove an employee from an existing job pool.

Chat responses

Check refusal and clarification before consuming rows. interpretation describes the query; row fields depend on the question. references retain available source identities, and audit provides query context. Not every result or aggregate has a reference.

A subject selects the employee using the same external ID and source system used for ingestion. Without one, supported inventory queries may be company-wide. Workspace restrictions control resource access; do not treat workspace selection or a chat subject as a department-level privacy boundary.

Handle errors and processing states

from workerbee_sdk import Workerbee
from workerbee_sdk.errors import (
    WorkerbeeError, ReadyTimeoutError, FileProcessingError,
)

try:
    with Workerbee() as wb:
        file = wb.get_file("YOUR_FILE_ID")
        file.wait_until_ready(timeout=600)
except ReadyTimeoutError as error:
    print("Still processing:", error.file.id, error.file.status)
except FileProcessingError as error:
    print(error.code, error.file.status)
    for adapter in error.file.adapters:
        print(adapter.document_type, adapter.status, adapter.error)
except WorkerbeeError as error:
    print(error.code, error.status, error.request_id, error.retryable)

Settled outcomes raise FileNeedsReviewError, FilePartialSuccessError, FileFailedError, or FileDeletedError. Their file contains the latest snapshot. Authentication, permission, validation, rate-limit, connection, and server errors also derive from WorkerbeeError. Keep request IDs when reporting failures; never log the API key.

Last updated October 2, 2026