Python SDK cookbook

Concrete recipes for middleware, batch, retries, and async concurrency.

Middleware around an AI call

import time
from cohesion import Client

cohesion = Client(api_key=os.environ["COHESION_API_KEY"])

def ai_with_cohesion(prompt, operator_id, session_id):
    started = time.time()
    ai_out = call_your_ai_provider(prompt)
    elapsed_ms = int((time.time() - started) * 1000)

    decision = capture_operator_decision(ai_out)

    cohesion.score(
        session_id=session_id,
        operator_id=operator_id,
        domain="financial",
        interaction={
            "ai_recommendation_presented": True,
            "time_to_decision_ms": elapsed_ms,
            "decision": decision.kind,
            "modification_extent": decision.edit_ratio,
            "ai_available": True,
            "scenario_type": "standard",
            "outcome_correct": None,
            "hover_events": decision.hovers,
            "scroll_depth": decision.scroll,
            "alternative_views_checked": decision.alt_views,
        },
    )
    return ai_out

Batch backfill from a telemetry store

from itertools import islice

def chunk(iterable, n):
    it = iter(iterable)
    while batch := list(islice(it, n)):
        yield batch

for interactions in chunk(telemetry.stream(), 100):
    cohesion.score_batch(operator_id="analyst-42", domain="financial", interactions=interactions)

Retry on 429

The SDK retries automatically with exponential backoff. To customize:

client = Client(api_key="...", max_retries=5)

Async concurrency

import asyncio
from cohesion import AsyncClient

async def score_many(items):
    async with AsyncClient(api_key=os.environ["COHESION_API_KEY"]) as c:
        return await asyncio.gather(*(c.score(**item) for item in items))

Idempotency keys

import uuid
cohesion.score(..., idempotency_key=str(uuid.uuid4()))

Client-side generation today. Server-side deduplication lands in v1.2.

Next step