Python SDK

cortina-sdk ist der offizielle async Python-Client für Cortina. Auf httpx und pydantic aufgebaut, asyncio-native, mit Built-in Retry-with-Backoff.

import asyncio
from cortina import AsyncCortinaClient

async def main() -> None:
    async with AsyncCortinaClient(api_key="tup_...") as cortina:
        masked = await cortina.anonymize(
            "Hans Müller, IBAN DE89 3704 0044 0532 0130 00, ruft +49 89 1234567 an."
        )
        # masked.anonymized_text:
        # "[PERSON_1], IBAN [DE_IBAN_1], ruft [DE_PHONE_NUMBER_1] an."

        llm_reply = await call_my_llm(masked.anonymized_text)

        restored = await cortina.deanonymize(
            text=llm_reply, session_id=masked.session_id
        )

asyncio.run(main())

Install

🚧 Noch nicht auf PyPI publiziert

cortina-sdk ist funktional fertig (0.1.0, Phase 4) — async-only in dieser Version (AsyncCortinaClient). Sync-Wrapper sind als Follow-up geplant. Aktueller Pfad: aus dem Source unter anonymizer/sdks/python bauen — uv pip install -e ./sdks/python. PyPI-Publish kommt in einem separaten Phase-4-Sub-Commit (Trusted Publishing).

pip install cortina-sdk
# oder
uv add cortina-sdk

Voraussetzungen: Python ≥ 3.10. Runtime-Dependencies: httpx, pydantic.

Quickstart

Convenience: cortina.anonymize(text)

Der kürzeste Pfad. Die SDK erstellt automatisch eine Session und gibt ihre ID auf der Response mit zurück.

async with AsyncCortinaClient(api_key=api_key) as cortina:
    masked = await cortina.anonymize("Anna Müller wohnt in Köln.")
    print(masked.anonymized_text)  # "[PERSON_1] wohnt in [LOCATION_1]."

    # Round-Trip später, im selben oder einem anderen Prozess:
    restored = await cortina.deanonymize(
        text=masked.anonymized_text, session_id=masked.session_id
    )

Sessions: cortina.create_session()

Wenn du mehrere Texte mit gemeinsamem Pseudonym-Vokabular anonymisieren willst (Multi-Turn-Conversation, Dokument in Chunks), nutze eine explizite Session. Innerhalb derselben Session bekommt „Anna” jedes Mal denselben [PERSON_1]-Platzhalter.

async with AsyncCortinaClient(api_key=api_key) as cortina:
    async with await cortina.create_session(ttl_seconds=3600) as session:
        turn1 = await session.anonymize("Hans rief seine Frau Anna an.")
        turn2 = await session.anonymize("Anna ist Ärztin in Frankfurt.")
        # Über turn1 und turn2 hinweg ist „Anna" derselbe [PERSON_2].

        joined = "\n".join(t.anonymized_text for t in (turn1, turn2))
        llm_reply = await call_my_llm(joined)
        restored = await session.deanonymize(llm_reply)
    # Session wird beim Exit automatisch destroyed (Redis-Mapping wiped).

ttl_seconds defaults zu 3600 (1 h), Server-Bounds [60, 86400].

Batch: cortina.anonymize_batch(inputs)

Parallele Anonymisierung mit asyncio.Semaphore-beschränkter Concurrency. Per-Element-Ergebnisse kommen in der Input-Reihenfolge zurück.

batch = await cortina.anonymize_batch(
    [
        "Anna lebt in Köln.",
        "Hans Müller, IBAN DE89 3704 0044 0532 0130 00.",
        "Ruf mich an unter +49 89 1234567",
    ],
    concurrency=5,
)

print(batch.success_count, batch.failure_count)
for r in batch.results:
    if r.status == "fulfilled":
        print(r.index, r.value.anonymized_text)
    else:
        print(r.index, repr(r.reason))

Wenn ein Input keine session_id mitbringt, erstellt der Batch eine gemeinsame Session vorab und nutzt sie für alle solchen Inputs — ein 100-Element-Batch kostet so eine Session-Create-Round- Trip, nicht 100.

stop_on_first_error=True raised den ersten Fehler direkt; pending Tasks werden gecanceled.

Stage-Filter

Cortina läuft per Default mit allen vier Stages. Wenn die LLM-Latenz nicht zum Workflow passt, einzelne Stages skippen:

await cortina.anonymize(text, stages=["stage1", "stage2"])

→ Pipeline-Trade-offs.

Configuration

AsyncCortinaClient(
    *,
    api_key: str,                                   # required, "tup_..."
    base_url: str = "https://api.tup-ai.de/v1/cortina",
    timeout: float = 30.0,                          # Sekunden, pro Call
    retry: RetryOptions | None = DEFAULT_RETRY,     # None = Retry aus
    default_headers: Mapping[str, str] | None = None,
    http_client: httpx.AsyncClient | None = None,   # Inject your own
)

Retry-with-Backoff

Transiente Transport-Fehler werden automatisch retried. Defaults: 3 Retries, 0,5 s Base-Delay, 30 s Cap, ±0,1 s Jitter. Der Retry-After- Header bei 429 wird respektiert (gecappt auf max_delay).

Error	Retried?
RateLimitError (429)	✅
ServerError 502 / 503 / 504	✅
ServerError andere 5xx	❌
NetworkError	✅
RequestTimeoutError	✅
AuthenticationError (401 / 403)	❌
ValidationError (400 / 422)	❌
NotFoundError (404)	❌
asyncio.CancelledError	❌

Override oder ausschalten:

from cortina import AsyncCortinaClient, RetryOptions

AsyncCortinaClient(
    api_key=api_key,
    retry=RetryOptions(max_retries=5, base_delay=1.0, max_delay=60.0, jitter=0.2),
)

# Komplett aus — genau ein Versuch:
AsyncCortinaClient(api_key=api_key, retry=None)

Cancellation

Die SDK respektiert standardmäßiges asyncio-Cancellation. Per-Call- Timeout via asyncio.wait_for(...) oder den umgebenden Task canceln. Cancellation propagiert durch die Retry-Loop.

import asyncio

try:
    masked = await asyncio.wait_for(cortina.anonymize(text), timeout=5.0)
except asyncio.TimeoutError:
    # Call wurde gecanceled; nichts wurde mehr auf die Wire geschickt
    ...

Zusätzlich Built-in Per-Call-Timeout (default 30 s, konfigurierbar via Constructor) der als RequestTimeoutError surfaced.

Error-Handling

Alle SDK-Errors erben von CortinaError. Die Klassen-Hierarchie spiegelt das HTTP-Status-Set des Service:

from cortina import (
    AsyncCortinaClient,
    CortinaError,
    AuthenticationError,    # 401 / 403
    ValidationError,        # 400 / 422
    NotFoundError,          # 404
    RateLimitError,         # 429 — hat `retry_after` (Sekunden)
    ServerError,            # 5xx
    NetworkError,           # Transport failed (DNS, Connection-Refused, TLS)
    RequestTimeoutError,    # Per-Call-Timeout überschritten
)

try:
    await cortina.anonymize(text)
except RateLimitError as err:
    await asyncio.sleep(err.retry_after or 1)
except AuthenticationError:
    # Key rotieren, Ops alerten
    raise
except CortinaError as err:
    print(err.status, err.details and err.details.code, err)

Hinweis: Die SDK-Timeout-Klasse heißt RequestTimeoutError, nicht TimeoutError — das vermeidet ein Shadowing des Python-Builtins, das seit 3.11 auch von asyncio bei Cancellation geraised wird. Wer jeden SDK-Error uniform fangen will: CortinaError.

Public Response-Shape

Wire ist snake_case und die SDK-Pydantic-Models matchen das 1:1 — keine Transform-Layer.

class AnonymizeResponse(BaseModel):
    anonymized_text: str
    entities: list[PiiEntity]
    session_id: UUID
    audit_flags: list[str]

class PiiEntity(BaseModel):
    entity_type: str
    start: int                # UTF-16 offset im Originaltext
    end: int
    placeholder: str
    confidence: float

re_id_score, Adversarial-Iteration-Counts und LLM-Token-Counts werden absichtlich nicht in der Response surfaced — sie leben nur im Audit-Log.

Roadmap

• Sync-CortinaClient-Wrapper (geplant, additiv — kein Breaking Change für AsyncCortinaClient-User).
• PyPI-Publish via Trusted Publishing.

Weiterführend

→ API-Reference: Anonymize → Sessions & Mappings