Python SDK

pip install nusomi

from nusomi import Nusomi

nusomi = Nusomi(
    api_key=os.environ["NUSOMI_API_KEY"],
    base_url="https://api.nusomi.com",  # override for self-hosted
    timeout=30.0,
    max_retries=3,
)

from nusomi import AsyncNusomi

async with AsyncNusomi(api_key=...) as nusomi:
    session = await nusomi.sessions.create(workflow="...")
    ...

nusomi.sessions
nusomi.events
nusomi.frames
nusomi.replay
nusomi.recovery
nusomi.memory
nusomi.exports
nusomi.workflows
nusomi.webhooks

nusomi.sessions.create(workflow, metadata=None, actor=None) -> Session
nusomi.sessions.get(id) -> Session
nusomi.sessions.list(**filters) -> Iterator[Session]
nusomi.sessions.start(id) -> Session
nusomi.sessions.stop(id) -> Session
nusomi.sessions.tag(id, name, **data) -> None
nusomi.sessions.delete(id) -> None

with nusomi.sessions.create(workflow="...") as session:
    # session.start() called on enter, session.stop() on exit
    do_work()

nusomi.events.query(session_id, **filters) -> list[Event]
nusomi.events.stream(session_id, **opts) -> Iterator[Event]
nusomi.events.emit(session_id, type, **payload) -> None

nusomi.frames.get(frame_id) -> Frame
nusomi.frames.image(frame_id) -> bytes
nusomi.frames.list(session_id, **filters) -> Iterator[Frame]

nusomi.replay.exact(session_id, **opts) -> Replay
nusomi.replay.guided(session_id, **opts) -> Replay
nusomi.replay.partial(session_id, from_frame=..., **opts) -> Replay

nusomi.recovery.from_session(session_id, **opts) -> Replay

nusomi.memory.paths(**filters) -> list[Path]
nusomi.memory.outliers(**filters) -> list[Session]
nusomi.memory.similar(session_id, limit=10) -> list[Session]
nusomi.memory.recovery_points(**filters) -> list[Frame]
nusomi.memory.on_drift(workflow, handler) -> Subscription

nusomi.exports.create(workflow, filter=None, format="parquet", destination=None) -> Export
nusomi.exports.get(id) -> Export
nusomi.exports.list(**filters) -> Iterator[Export]

nusomi.webhooks.list() -> list[Webhook]
nusomi.webhooks.create(**input) -> Webhook
nusomi.webhooks.delete(id) -> None

for s in nusomi.sessions.list(workflow="process_invoice", since="30d"):
    print(s.id, s.duration_ms)

from nusomi import NusomiError

try:
    nusomi.sessions.create(workflow="...")
except NusomiError as e:
    print(e.code, e.message, e.request_id)
    if e.code == "rate_limited":
        time.sleep(e.retry_after_ms / 1000)

for ev in nusomi.events.stream(
    session_id,
    types=["click", "validation_error"],
):
    print(ev.t_ms, ev.type, ev.payload)

async with AsyncNusomi(api_key=...) as nusomi:
    async for ev in nusomi.events.stream(session_id):
        print(ev)

from nusomi.types import Session, Event, Frame, ClickEvent

match ev.type:
    case "click":
        print(ev.payload.x, ev.payload.y)
    case "validation_error":
        print(ev.payload.field, ev.payload.message)

from playwright.sync_api import sync_playwright
from nusomi import Nusomi

nusomi = Nusomi(api_key=...)

with sync_playwright() as p:
    browser = p.chromium.launch()
    context = browser.new_context()

    with nusomi.sessions.create(workflow="login_test") as session:
        session.attach_playwright(context)
        page = context.new_page()
        page.goto("https://app.example.com/login")
        ...

    browser.close()

import logging
logging.getLogger("nusomi").setLevel(logging.DEBUG)