AsyncConnectorClient

AsyncConnectorClient is the async-native client. Use it when your pipeline uses asyncio — LangChain, LlamaIndex, FastAPI, or any framework where embedding and vector store calls are async. Callbacks are async def coroutines, so you can await any async operation directly inside them.

Constructor

from autoplay_sdk import AsyncConnectorClient

client = AsyncConnectorClient(
    url="https://your-connector.onrender.com/stream/YOUR_PRODUCT_ID",
    token="uk_live_...",
    session_concurrency=4,
)

string

required

Full URL to GET /stream/{product_id} on the connector.

string

default:"\"\""

Unkey API key (uk_live_...). Leave empty to connect without authentication.

int

default:"4"

Maximum number of concurrent in-flight callback tasks per session. Each session’s callbacks run as independent asyncio.Task instances — a slow LLM or vector store call for one session never delays delivery for another. Increase this if a single session can generate bursts of events faster than your callback can process them.

Callbacks

All registration methods return self for chaining.

`.on_actions(fn)`

from autoplay_sdk import ActionsPayload

async def handle_actions(payload: ActionsPayload):
    text = payload.to_text()
    embedding = await embed_api.create(input=text)
    await vector_store.upsert(id=payload.session_id, vector=embedding)

client.on_actions(handle_actions)

async Callable[[ActionsPayload], None]

required

Async coroutine called every time a batch of UI actions arrives. await any async I/O directly — it won’t block other coroutines.

`.on_summary(fn)`

from autoplay_sdk import SummaryPayload

async def handle_summary(payload: SummaryPayload):
    text = payload.to_text()
    await vector_store.upsert(id=payload.session_id, vector=await embed(text))

client.on_summary(handle_summary)

async Callable[[SummaryPayload], None]

required

Async coroutine called every time a session summary arrives.

Per-session task isolation

Each incoming payload is dispatched as an independent asyncio.Task keyed by session_id. This means:

A slow embedding call for session A never delays delivery for session B.
Each session has its own asyncio.Semaphore capped at session_concurrency concurrent tasks.
The SSE reader loop is never blocked by your callback logic.

This mirrors the event connector’s own per-session asyncio.Task architecture.

Lifecycle

`await .run()`

Connect to the SSE stream and process events. Reconnects with exponential backoff (1 s → 30 s).

async def main():
    async with AsyncConnectorClient(url=URL, token=TOKEN) as client:
        client.on_actions(handle_actions)
        await client.run()

asyncio.run(main())

HTTP 401, 403, and 404 are not retried — they raise immediately. Check your url and token.

`.run_in_background()`

Schedule run() as an asyncio.Task and return immediately.

task = client.run_in_background()

await do_other_async_work()

client.stop()
await task

Returns an asyncio.Task.

`.stop()`

Signal the client to stop after the current event.

Async context manager

async with AsyncConnectorClient(url=URL, token=TOKEN) as client:
    client.on_actions(handle_actions).on_summary(handle_summary)
    await client.run()
# stop() called automatically on exit

FastAPI integration

from contextlib import asynccontextmanager
from fastapi import FastAPI
from autoplay_sdk import AsyncConnectorClient, ActionsPayload

async def handle_actions(payload: ActionsPayload):
    await embed_and_store(payload.to_text(), session_id=payload.session_id)

client = AsyncConnectorClient(url=URL, token=TOKEN).on_actions(handle_actions)

@asynccontextmanager
async def lifespan(app: FastAPI):
    task = client.run_in_background()
    yield
    client.stop()
    await task

app = FastAPI(lifespan=lifespan)

Chained setup

await (
    AsyncConnectorClient(url=URL, token=TOKEN)
    .on_actions(handle_actions)
    .on_summary(handle_summary)
    .run()
)

Who We Are

Get Started

Live User Activity

AI Support Agents

Visual User Tours

Receive events

🔌 Live activity via MCP

Build with events

Integration helpers

Context sources

Reference

Constructor

Callbacks

`.on_actions(fn)`

`.on_summary(fn)`

Per-session task isolation

Lifecycle

`await .run()`

`.run_in_background()`

`.stop()`

Async context manager

FastAPI integration

Chained setup

​Constructor

​Callbacks

​.on_actions(fn)

​.on_summary(fn)

​Per-session task isolation

​Lifecycle

​await .run()

​.run_in_background()

​.stop()

​Async context manager

​FastAPI integration

​Chained setup

Constructor

Callbacks

`.on_actions(fn)`

`.on_summary(fn)`

Per-session task isolation

Lifecycle

`await .run()`

`.run_in_background()`

`.stop()`

Async context manager

FastAPI integration

Chained setup