autoplay-core

name	autoplay-core
description	Sets up the Autoplay SDK for real-time event streaming into any chatbot or AI agent. Covers install, credentials, stream wiring, LLM guardrails, and the two most critical concepts — session scoping and conversation scoping — that must be correct regardless of which chatbot or activity source is used. Use when starting any Autoplay integration, when the user mentions autoplay-sdk, AsyncConnectorClient, ActionsPayload, or asks how to wire real-time user events into a chatbot.
disable-model-invocation	true

Autoplay Core

Install

pip install autoplay-sdk
# Optional: one-line FastAPI bridge
# pip install "autoplay-sdk[serve]"
# Plus your preferred LLM client, e.g.:
#   pip install openai        # OpenAI / Azure OpenAI
#   pip install anthropic     # Anthropic Claude
#   pip install google-generativeai  # Gemini

Upgrading an existing integration? Install the migration helper skill and ask your agent to rewrite deprecated imports:

autoplay-install-skills --migrate

Credentials come from the Autoplay dashboard after running onboard_product:

• CONNECTOR_URL — stream URL: https://your-connector.onrender.com/stream/YOUR_PRODUCT_ID
• API_TOKEN — Bearer token (Unkey key)

---

⚠️ Critical: Session scoping & conversation scoping

This is the most important concept. Get this wrong and context will be empty, mixed between users, or silently dropped.

Recommended path — AutoplayChatbotManager

AutoplayChatbotManager handles all session lifecycle automatically. Use it unless you need custom Redis-backed stores or non-standard delivery logic.

from autoplay_sdk import AutoplayChatbotManager, BaseChatbotWriter

# 1. Implement _post_note — the only platform-specific code
class MyWriter(BaseChatbotWriter):
    async def _post_note(self, conversation_id: str, text: str) -> None:
        # call your chatbot platform's API here
        ...

# 2. Initialize once at server startup
manager = AutoplayChatbotManager(writer=MyWriter())

# 3. Autoplay stream handler — identical for every platform
async def on_actions(payload):
    await manager.on_actions(payload)

# 4. Chatbot webhook handler — only the extraction lines differ per platform
async def on_webhook(data):
    session_id = data["custom_attributes"]["session_id"]  # ← platform-specific
    conversation_id = data["id"]                          # ← platform-specific
    await manager.on_chatbot_event(session_id, conversation_id)

That's it. The manager handles:

• Creating SessionState the moment a session_id is first seen (PostHog is the ground truth)
• Buffering actions before a conversation is linked
• Auto-detecting NEW vs REPLY_EXISTING from the link store
• Flushing buffered actions the moment the link is established
• Routing all future delivery to conversation_id permanently

session_id comes from PostHog. conversation_id comes from the chatbot platform. They are married together by on_chatbot_event. After that call, all proactive triggers, reactive notes, and future actions route to that conversation for the lifetime of the session.

---

Low-level building blocks (advanced use)

Use these directly only when you need custom Redis-backed stores or non-standard delivery.

from autoplay_sdk import (
    InMemorySessionStateStore,
    InMemoryConversationLinkStore,
    link_conversation,
)
from autoplay_sdk.agent_state.v2 import SessionState

# Initialize stores at server startup
session_store = InMemorySessionStateStore()
link_store = InMemoryConversationLinkStore()

# Phase 1 — Autoplay stream: SessionState born here, session_id is the ground truth
async def on_actions(payload):
    if not payload.session_id:
        return
    state = await session_store.get_or_create(payload.session_id)  # ← first, always
    await writer.write_actions(...)

# Phase 2 — Chatbot webhook: look up existing state, add conversation_id
async def on_webhook(data):
    session_id = data["custom_attributes"]["session_id"]  # platform-specific
    conversation_id = data["id"]                          # platform-specific
    state = await session_store.get_or_create(session_id)
    link_conversation(state=state, store=link_store, conversation_id=conversation_id)
    await writer.on_session_linked(session_id, conversation_id)

SessionState fields:

• state.session_id — mandatory required field; SessionState(session_id=...) — construction without it is a TypeError
• state.metadata — optional extras (user_id, email, any future fields)
• state.conversation_linked / state.conversation_id — linked chatbot conversation
• state.current_state — FSM (THINKING / PROACTIVE / REACTIVE)

Production: swap to Redis-backed store

InMemorySessionStateStore loses all state on restart. For production, use RedisSessionStateStore — sessions survive restarts and scale across instances. It is a drop-in replacement:

# Install the redis extra
# pip install "autoplay-sdk[redis]"

from autoplay_sdk import AutoplayChatbotManager, RedisSessionStateStore

# One-line swap at server startup
manager = AutoplayChatbotManager(
    writer=MyWriter(),
    session_store=RedisSessionStateStore(redis_url=os.environ["REDIS_URL"]),
)

Key pattern: autoplay:session_state:{session_id}. Default TTL: 24 h (configurable via ttl_s). On Redis errors, falls back gracefully and logs a warning — never crashes.

What breaks without correct scoping:

• Wrong user's events delivered to wrong conversation
• Empty context because retrieval key doesn't match storage key
• Pre-link actions silently dropped

---

Fastest SDK-native path (recommended)

import asyncio
from autoplay_sdk import AsyncConnectorClient, compose_chat_pipeline
from autoplay_sdk.context.user_index import UserSessionIndex

CONNECTOR_URL = "https://your-connector.onrender.com/stream/YOUR_PRODUCT_ID"
API_TOKEN = "your-api-token"

async def llm(prompt: str) -> str:
    """Wire your preferred LLM here.

    The SDK only needs an async callable: (str) -> str.
    Any provider works — OpenAI, Anthropic, Gemini, Mistral, a local model, etc.

    OpenAI example:
        from openai import AsyncOpenAI
        _client = AsyncOpenAI()
        r = await _client.chat.completions.create(model="gpt-4o-mini", messages=[{"role":"user","content":prompt}])
        return r.choices[0].message.content
    """
    raise NotImplementedError("Replace with your LLM client")

# Compose summarizer + context store + writer safely (no callback clobbering)
pipeline = compose_chat_pipeline(
    llm=llm,
    threshold=20,
    lookback_seconds=300,
    max_actions=20,
    write_actions=None,           # optional push callback for chatbot notes
    overwrite_with_summary=None,  # optional summary overwrite callback
)
user_index = UserSessionIndex(pipeline.context_store, lookback_seconds=300)

async def run():
    async with AsyncConnectorClient(url=CONNECTOR_URL, token=API_TOKEN) as client:
        async def on_actions(payload):
            await pipeline.on_actions(payload)
            user_index.add(payload)

        client.on_actions(on_actions)
        await client.run()

asyncio.run(run())

If you skip compose_chat_pipeline, summaries are never produced. The on_summary / overwrite_with_summary callback requires the pipeline's summarizer to be wired into the stream. There is no implicit fallback — omitting this call means overwrite_with_summary never fires, regardless of the threshold value.

When your chatbot gets a user_id, call:

activity = user_index.get_user_activity(user_id)

This avoids hand-rolled user_id -> session_id indexing and keeps product_id-aware lookups correct.

---

Analytics-only recipe (no chatbot)

If you only need to receive enriched events and forward them to your own pipeline — no chatbot, no notes, no session linking — you need none of the chatbot-shaped machinery above. Just the client:

import asyncio
from autoplay_sdk import AsyncConnectorClient

CONNECTOR_URL = "https://your-connector.onrender.com/stream/YOUR_PRODUCT_ID"
API_TOKEN = "your-api-token"

async def run():
    async with AsyncConnectorClient(url=CONNECTOR_URL, token=API_TOKEN) as client:
        async def on_actions(payload):
            if not payload.session_id:
                return
            # payload.slim_actions  — list of enriched action dicts
            # payload.user_id       — set if posthog.identify() was called
            # payload.email         — set if posthog.identify() was called
            your_downstream(payload)  # push to PostHog, Segment, your DB, etc.

        client.on_actions(on_actions)
        await client.run()

asyncio.run(run())

No BaseChatbotWriter, no compose_chat_pipeline, no session store required.

---

Optional one-line HTTP bridge

If your chatbot runtime is in another process (Rasa/Botpress/Twilio/custom webhook), prefer the built-in FastAPI factory:

from autoplay_sdk.api import build_copilot_app

app = build_copilot_app(
    stream_url=CONNECTOR_URL,
    token=API_TOKEN,
    llm=llm,
    summary_threshold=20,
    lookback_seconds=300,
)

Default endpoints:

• GET /healthz
• GET /context/{user_id}?query=...
• GET /reply/{user_id}?query=...
• POST /admin/reset/{user_id}

---

LLM guardrails

Use versioned prompt metadata with your provider client. The SDK does not currently ship a call_llm(...) helper.

# GOOD — use prompt metadata alongside your provider call
MY_PROMPT = {
    "name": "Support Answer Prompt",
    "version": "0.1",
    "description": "Answer user questions with product-aware context",
    "content": "You are a helpful assistant...\n\n{context}",
}

messages = [{"role": "system", "content": MY_PROMPT["content"].format(context=context_text)}]
response = await openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
)

# BAD — inline prompt string with no version metadata
response = await openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "system", "content": "You are a helpful assistant..."}],
)

Prompt files live in src/llm/ only. Each exports:

MY_PROMPT = {
    "name": "My Prompt",
    "version": "0.1",
    "description": "...",
    "content": "...",
}

Log: prompt name + version, model, token usage, latency, errors. Never log full session payloads.

---

Integration Self-Reasoning Checklist

Run through this checklist before considering any chatbot integration complete. Reason through each question — do not skip.

1. Session identity

• Is session_id the first thing set in SessionState? state = SessionState(session_id=...) — SessionState() without session_id is a TypeError.
• Is PostHog the ground truth for session_id? It must come from ActionsPayload.session_id, never from the chatbot platform.
• Is await session_store.get_or_create(session_id) the very first await in every handler — before any processing, delivery, or API call?
• Does state.metadata carry all optional identity context (user_id, email, etc.)? Nothing important should live outside SessionState.

2. Race conditions

• Webhook fires before stream: get_or_create handles this — it creates the state with session_id set. Actions that arrive later from the stream route correctly because the state already exists.
• Stream fires before webhook: Actions buffer in BaseChatbotWriter._pending until on_session_linked is called. Nothing is dropped.
• Concurrent webhooks for the same session: InMemorySessionStateStore.get_or_create is a synchronous dict operation — race-safe. RedisSessionStateStore does a GET-before-SET — acceptable for session context (last writer wins on the rare collision).
• Session state missing after restart: If using InMemorySessionStateStore in production, state is lost. Switch to RedisSessionStateStore.

3. Chatbot linking

• Is link_conversation(state=state, store=store, conversation_id=conv_id) called with no event= argument? Let it auto-detect NEW vs REPLY_EXISTING from the store.
• Is await writer.on_session_linked(session_id, conversation_id) called immediately after link_conversation? This is the flush trigger — without it, buffered actions are never delivered.
• Is the link sticky? Once state.conversation_linked == True, the same session must not be re-linked to a different conversation_id. The REPLY_EXISTING semantics handle this automatically — but verify the webhook handler is not calling link_conversation again on subsequent replies.
• If using RedisSessionStateStore: is await session_store.save(state) called after linking? AutoplayChatbotManager.on_chatbot_event does this automatically — verify if using the low-level path.

4. Proactive triggers

• Are proactive triggers reading state.conversation_id as the delivery target?
• Is there a guard: if not state.conversation_linked: return before calling the chatbot API? A trigger that fires before the conversation is linked has nowhere to deliver.
• Does the FSM gate the trigger correctly? Call state.tick() first, then allowed, reason = state.can_deliver_proactive(). This is the canonical check — it verifies both THINKING state and cooldown simultaneously. Do not replicate this logic inline.
• After delivering a proactive trigger, is the FSM transitioned? state.transition_to_proactive(trigger_id) must be called or the session stays in THINKING and the trigger may re-fire.
• After any FSM transition (to_proactive, to_reactive, tick, start_cooldown), is await manager.save_state(state) called? on_actions and on_chatbot_event auto-save — but external FSM mutations require an explicit save_state call:

  state.to_proactive(session_id=state.session_id, ...)
  await manager.save_state(state)   # persist PROACTIVE FSM state to Redis
  

5. Persistence (production)

• On first session_id arrival, get_or_create creates and immediately persists SessionState to Redis. This is the ground truth — everything else builds on it.
• on_actions does NOT save state — it only reads state.conversation_id for routing and passes actions to the writer. SessionState is not mutated by action delivery.
• on_chatbot_event always saves after linking — conversation_linked and conversation_id are the mutations that need persistence there.
• For any FSM mutation that happens outside on_chatbot_event, call await manager.save_state(state). This covers: to_proactive, to_reactive, tick, start_cooldown, record_user_interaction, and any metadata update.
• Is InMemorySessionStateStore replaced with RedisSessionStateStore in production?

  # pip install "autoplay-sdk[redis]"
  manager = AutoplayChatbotManager(
      writer=MyWriter(),
      session_store=RedisSessionStateStore(redis_url=os.environ["REDIS_URL"]),
  )
  

• Is REDIS_URL set in the production environment?
• Is the TTL appropriate? Default is 24 h (ttl_s=86400). Sessions longer than the TTL will lose state.

Smoke test — verify the full hop before going to production

# Run from your project root after setting CONNECTOR_URL and API_TOKEN:
python -m autoplay_sdk.smoke_test \
    --url "$CONNECTOR_URL" \
    --token "$API_TOKEN"
# Expected output:
#   ✓ connection  — stream connected
#   ✓ heartbeat   — first ActionsPayload received within 10 s
#   ✓ session_id  — payload.session_id is non-empty

If session_id is empty at this stage, PostHog identify() has not fired yet — see activity-posthog Step 2.

---

Reference

• Quickstart: https://developers.autoplay.ai/quickstart
• SDK overview: https://developers.autoplay.ai/sdk/overview
• Chatbot tutorials: https://developers.autoplay.ai/recipes/intercom-tutorial