tagnos 0.3.1rc1

Official Python SDK for Tagnos — AI-first HITL annotation studio

tagnos-sdk

Official Python SDK for Tagnos — the AI-first human-in-the-loop annotation studio. Programmatic access to projects, schemas (TAM), items, annotations, consensus, exports, and integrations (Langfuse, …), plus a webhook receiver for real-time events.

Status: alpha. API surface may change before 1.0. Pin tagnos<0.3 in production.

Install

Requires Python 3.10+.

pip install tagnos                    # core SDK (sync + async)
pip install "tagnos[webhooks]"        # + starlette webhook receiver
pip install "tagnos[pandas]"          # + DataFrame → items adapter
pip install "tagnos[hf]"              # + 🤗 datasets → items adapter

The SDK is distributed on PyPI (public) and mirrored on Google Artifact Registry (europe-west1-python.pkg.dev/ellzx-491606/tagnos-python/) for internal GCP consumers. Both are updated on every vX.Y.Z tag push.

To install from the private GAR mirror instead (e.g. from a Cloud Run service in the same project), see docs/DEPLOYMENT.md.

Authentication

Generate a personal API key from https://<your-org>.tagnos.app/settings → API keys.

from tagnos import Tagnos

client = Tagnos(
    org_slug="acme",            # your tenant subdomain
    api_key="tk_live_...",      # or env TAGNOS_API_KEY
)

The SDK reads TAGNOS_API_KEY, TAGNOS_ORG_SLUG, and TAGNOS_BASE_DOMAIN (default tagnos.app) from the environment when unset.

Quickstart — sync

from tagnos import Tagnos, TamBuilder

with Tagnos() as client:
    # 1. Create a project from a starter template
    tam = client.schema.from_template("medical_ner")
    project = client.projects.create(name="Consultations Q2", tam=tam)

    # 2. Push items in bulk
    client.items.import_jsonl(
        project.id,
        [
            {"externalId": "case-001", "content": "Le patient présente…"},
            {"externalId": "case-002", "content": "Pas de plainte notable."},
        ],
    )

    # 3. Iterate the full annotated queue (cursor-paginated)
    for item in client.items.iter(project.id, limit=100):
        print(item.id, item.status)

    # 4. Export once annotations are in
    client.exports.download_to(project.id, "dataset.jsonl", format="jsonl")

Quickstart — async

import asyncio
from tagnos import AsyncTagnos


async def main() -> None:
    async with AsyncTagnos() as client:
        # Fan-out: fetch first page of items for every project in parallel
        projects = await client.projects.list()
        pages = await asyncio.gather(
            *(client.items.list(p.id, limit=20) for p in projects)
        )
        for project, page in zip(projects, pages):
            print(project.name, len(page.items), "items")


asyncio.run(main())

Build a schema with TamBuilder

Tagnos projects are configured with a TAM (Tagnos Annotation Manifest) — a JSON schema that declares modality, tools (NER, classification, rating, bbox, …), label sets, relations, and the AI pipeline. TamBuilder is a fluent builder that mirrors the server-side Zod schema:

from tagnos import TamBuilder

tam = (
    TamBuilder.text(name="Sentiment clients", locale="fr-FR")
    .add_classification(
        id="sentiment",
        label="Sentiment",
        values=[
            {"id": "pos", "name": "Positif", "color": "#10b981"},
            {"id": "neu", "name": "Neutre",  "color": "#6b7280"},
            {"id": "neg", "name": "Négatif", "color": "#ef4444"},
        ],
    )
    .add_rating(id="utility", label="Utilité", scale_min=1, scale_max=5)
    .with_consensus(auto_accept_threshold=0.85)
    .build()
)

project = client.projects.create(name="Avis clients", tam=tam)

Inspect the schema of an existing project:

from tagnos import TamInspector

payload = client.schema.get(project.id)
insp = TamInspector(payload.tam)

insp.modality                            # "text"
insp.tools_by_type("classification")     # [{'id': 'sentiment', …}]
insp.label_ids(tool_id="sentiment")      # ['pos', 'neu', 'neg']

Import data

Method	When to use
client.items.create_bulk(project_id, items)	You already have a Python list of small dicts
client.items.import_jsonl(project_id, rows)	Stream many rows as JSONL (1 object / line)
client.items.import_csv(project_id, rows)	CSV — must have a content or text column
client.items.import_file(project_id, path)	Upload a .jsonl or .csv file (format auto-detected)
dataframe_to_rows(df) (extra: pandas)	Adapter: DataFrame → item iterator
dataset_to_rows(ds) (extra: hf)	Adapter: 🤗 datasets.Dataset → item iterator

Deduplication: rows carrying an externalId are deduped server-side against already-imported items. Anonymous rows (no externalId) are always inserted.

Pagination

Every list endpoint supports cursor-based paging. Both sync and async clients expose a convenience .iter() generator that follows the cursor for you:

# sync
for annotation in client.annotations.iter(project_id, limit=200):
    ...

# async
async for annotation in client.annotations.iter(project_id, limit=200):
    ...

Webhooks

Tagnos posts webhook events (annotation.created, review.approved, item.completed, consensus.disagreement, …) to URLs you register in Settings → Webhooks. The receiver verifies the HMAC signature and dispatches to typed handlers.

from starlette.applications import Starlette
from starlette.routing import Route
from tagnos.webhooks import WebhookReceiver, AnnotationCreated

receiver = WebhookReceiver(secret="whk_...")

@receiver.on(AnnotationCreated)
async def on_annotation(evt: AnnotationCreated) -> None:
    print(f"new annotation {evt.annotation_id} on item {evt.item_id}")

app = Starlette(routes=[Route("/webhooks/tagnos", receiver.asgi, methods=["POST"])])

Integrations (Langfuse, …)

from tagnos import IntegrationType  # re-exported from tagnos.models

config = client.integrations.create(
    type=IntegrationType.LANGFUSE,
    name="Production traces",
    public_key="pk-lf-…",
    secret_key="sk-lf-…",
    api_url="https://cloud.langfuse.com",
    target_project_id=project.id,
)
client.integrations.test(config.id)      # dry-run connectivity

A Cloud Scheduler job on the Tagnos side pulls new traces hourly and turns them into annotatable items in target_project_id.

Assignments (task management)

Distribute items across your team using one of three strategies (round_robin, workload_balanced, random) and an optional due date:

members = client.members.list()
item_ids = [it.id for it in client.items.iter(project.id, limit=500)]

result = client.assignments.bulk_assign(
    project.id,
    item_ids=item_ids,
    user_ids=[m.user_id for m in members],
    strategy="workload_balanced",
    due_date="2026-05-15T17:00:00Z",
)
print(f"created {result.created}/{result.total} assignments")

# iterate outstanding work for a given user
for a in client.assignments.iter(project.id, user_id="u_123", status="PENDING"):
    print(a.item_id, a.due_date)

Errors

from tagnos import (
    TagnosAuthError,        # 401 / 403
    TagnosNotFoundError,    # 404
    TagnosConflictError,    # 409 — concurrent schema update
    TagnosRateLimitError,   # 429 — check .retry_after_seconds
    TagnosValidationError,  # 400 / 422
    TagnosAPIError,         # base class for all HTTP errors
)

The client auto-retries idempotent failures (429, 500, 502, 503, 504) with exponential backoff (honors Retry-After). Pass retries=0 to opt out.

Development

pip install -e ".[dev,webhooks,pandas]"
pytest
ruff check .
mypy src

Releasing

Tag-driven via Cloud Build (v*.*.*); see docs/DEPLOYMENT.md.

License

Apache-2.0