Metadata-Version: 2.4
Name: tagnos
Version: 0.3.1rc1
Summary: Official Python SDK for Tagnos — AI-first HITL annotation studio
Project-URL: Homepage, https://tagnos.app
Project-URL: Documentation, https://docs.tagnos.app
Project-URL: Source, https://github.com/ellzxperso/tagnos-sdk
Author-email: Tagnos <support@tagnos.app>
License: Apache-2.0
Keywords: annotation,dataset,fine-tuning,hitl,labeling,llm
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2.6
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pandas>=2.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-httpx>=0.30; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: respx>=0.21; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Provides-Extra: hf
Requires-Dist: datasets>=2.14; extra == 'hf'
Provides-Extra: pandas
Requires-Dist: pandas>=2.0; extra == 'pandas'
Provides-Extra: webhooks
Requires-Dist: starlette>=0.37; extra == 'webhooks'
Requires-Dist: uvicorn>=0.30; extra == 'webhooks'
Description-Content-Type: text/markdown

# tagnos-sdk

Official Python SDK for [Tagnos](https://tagnos.app) — 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+.

```bash
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`](./docs/DEPLOYMENT.md).

## Authentication

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

```python
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

```python
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

```python
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:

```python
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:

```python
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:

```python
# 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.

```python
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, …)

```python
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:

```python
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

```python
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

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

## Releasing

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

## License

Apache-2.0
