BaseChatbotWriter - Autoplay SDK

BaseChatbotWriter is the delivery engine for any chatbot integration. Subclass it, implement two async methods, and you get the full delivery policy for free. For the complete session lifecycle (state creation, linking, persistence), use AutoplayChatbotManager — it wraps BaseChatbotWriter and handles everything automatically.

AutoplayChatbotManager (recommended)

AutoplayChatbotManager is the fastest path to a working integration. It owns session state, conversation linking, and delivery — you only implement _post_note.

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) -> str | None:
        # call your chatbot platform API here
        ...
    async def _redact_part(self, conversation_id: str, part_id: str) -> None:
        pass  # implement if your platform supports note deletion

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

# 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)

session_id always comes from PostHog (via ActionsPayload.session_id). conversation_id comes from the chatbot platform webhook. AutoplayChatbotManager marries them and routes all future delivery to that conversation permanently.

Production: Redis-backed persistence

By default, AutoplayChatbotManager uses InMemorySessionStateStore — state is lost on restart. For production, pass a RedisSessionStateStore:

# pip install "autoplay-sdk[redis]"
from autoplay_sdk import AutoplayChatbotManager, RedisSessionStateStore
import os

manager = AutoplayChatbotManager(
    writer=MyWriter(product_id="my_product"),
    session_store=RedisSessionStateStore(redis_url=os.environ["REDIS_URL"]),
)

SessionState is the complete source of truth — FSM state, conversation link, identity, and metadata all persist across restarts. Key pattern: autoplay:session_state:{session_id}. Default TTL: 24 h.

What it does

Pre-link buffering (sliding window)

Before a conversation is linked to a PostHog session, actions are buffered in memory rather than discarded. On every new arrival, entries older than pre_link_window_s seconds are trimmed — so the buffer never grows unboundedly.

write_actions() called (no conversation yet)
  → append to _pending[session_id]
  → trim actions older than pre_link_window_s
  → no API call

At-link flush (one API call)

When on_session_linked() fires, the entire buffer is flushed as a single _post_note call. Actions are grouped into bin_seconds-wide visual bins separated by blank lines — making the user’s journey easy to scan. One API call regardless of how many events accumulated before the conversation opened.

on_session_linked("sess1", "conv-123")
  → flush all _pending["sess1"] as one note (binned)
  → clear buffer
  → cancel any in-flight debounce task

Post-link debounce (trailing edge)

After a conversation is linked, each write_actions() call appends to a per-session buffer and (re)schedules a short asyncio.Task. When the timer fires with no new arrivals, one _post_note is made. Rapid event bursts are coalesced; a pause longer than post_link_debounce_s triggers delivery.

write_actions() called (linked)
  → extend _debounce_buffer[session_id]
  → cancel previous debounce task (if any)
  → schedule new task: sleep(post_link_debounce_s) → _post_note()

Session-first linking

BaseChatbotWriter handles delivery. SessionState is the source of truth for session identity and conversation ownership:

session_id — required, always set first. Comes from PostHog (ActionsPayload.session_id), never from the chatbot platform.
metadata — open dict for optional extras (user_id, email, …).
conversation_linked / conversation_id — set by link_conversation; the permanent delivery target once set.
current_state — FSM (THINKING / PROACTIVE / REACTIVE).

Event semantics

Event	Meaning	`SessionState.on_conversation_linked(...)` behavior
`ConversationEventType.NEW`	A newly created conversation for this session	Always set `conversation_linked=True` and overwrite `conversation_id`
`ConversationEventType.REPLY_EXISTING`	Inbound reply on an existing conversation	Set link fields only when currently unlinked; do not overwrite an existing linked `conversation_id`

link_conversation auto-detects the event from the store — you do not need to pass it manually.

Canonical dual-write path (low-level)

When not using AutoplayChatbotManager, manage stores directly:

from autoplay_sdk import (
    InMemoryConversationLinkStore,
    InMemorySessionStateStore,
    link_conversation,
)

session_store = InMemorySessionStateStore()
link_store = InMemoryConversationLinkStore()

# Stream handler
async def on_actions(payload):
    if not payload.session_id:
        return
    state = await session_store.get_or_create(payload.session_id)  # always first
    await writer.write_actions(...)

# Webhook handler
async def on_webhook(data):
    session_id = data["custom_attributes"]["session_id"]
    conversation_id = data["id"]
    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)

link_conversation writes to ConversationLinkStore first, then calls SessionState.on_conversation_linked(...). The event type is inferred automatically: REPLY_EXISTING if the conversation_id is already in the store, NEW otherwise.

Hot-path routing helper

For delivery decisions, read from session state (no store lookup on the hot path):

from autoplay_sdk import resolve_linked_conversation_id

conv_id = resolve_linked_conversation_id(state)
if conv_id:
    # linked — send to existing conversation
    ...
else:
    # unlinked — actions buffer until on_session_linked is called
    ...

Note body format

BaseChatbotWriter builds the string passed to _post_note(conversation_id, body) as plain text, line-oriented.

Header (always)

The first lines come from format_chatbot_note_header(session_id, timestamp_unix):

session_id: …
timestamp: … UTC (human-readable from Unix time)
A blank line after the header

For action notes, the timestamp is taken from the earliest timestamp_start among the actions after sorting. If slim_actions is empty, the header uses the current time instead.

Action lines

Actions are sorted by timestamp_start before rendering.
Each line is [n] {description} where n is 1-based and local to that note. It is not the per-batch wire index on each slim action dict.
One action → one line [1] …. Many → [1] through [n]. Zero actions → header only (no action lines).

Binning

Pre-link flush (on_session_linked): uses the constructor’s bin_seconds (default 3). Actions whose timestamp_start values fall in different time bins get a blank line between groups so the note is easier to scan.
Post-link debounced notes: _format_note is called with bin_seconds=0, so no extra blank lines between groups.

Example (actions note)

session_id: abc123
timestamp: 2024-01-15 12:34:50 UTC

[1] User clicked Sign up button on the pricing page
[2] User clicked Confirm plan button on the checkout page

[3] User submitted Payment form on the checkout page

The blank line between [2] and [3] appears when those actions fall in different bin_seconds-wide bins (pre-link flush). Post-link notes omit those separators.

Summary notes (LLM)

_format_note applies only to action timelines. BaseChatbotWriter does not wrap LLM summary text. For overwrite_with_summary, build the body yourself: call format_chatbot_note_header(session_id, time.time()) (or another Unix timestamp), then append your summary prose. The in-repo Intercom integration does this for summary posts.

Import the helper from the package root: from autoplay_sdk import format_chatbot_note_header (also available from autoplay_sdk.chat.chatbot).

Constructor

`BaseChatbotWriter(product_id, pre_link_window_s=120, post_link_debounce_s=0.15, bin_seconds=3)`

product_id

str

required

Product identifier used in logs and metrics.

pre_link_window_s

float

default:"120.0"

How long to retain buffered actions before the session is linked. Actions older than this (measured by timestamp_start) are dropped on each new arrival. Default is 120 seconds (2 minutes).

post_link_debounce_s

float

default:"0.15"

Trailing-edge debounce window in seconds. After the last write_actions() call for a session, the writer waits this long before posting a note. Coalesces rapid event bursts into a single API call. Default is 150 ms.

bin_seconds

int

default:"3"

Width of time bins used to group actions into visual sections in the note body. Actions more than bin_seconds apart get a blank-line separator. Set to 0 to disable binning. Used for the pre-link flush note; post-link notes always use bin_seconds=0.

Building a custom backend

Subclass BaseChatbotWriter and implement these two methods:

`_post_note(conversation_id, body) → str | None`

Post a note to the platform conversation. Return its platform-assigned id (used for later redaction), or None if the platform does not support redaction.

`_redact_part(conversation_id, part_id) → None`

Delete or blank a previously posted note. This is called by AsyncAgentContextWriter’s overwrite_with_summary step when LLM summaries are enabled. Implement as a no-op if the platform does not support redaction.

Example — Zendesk

from autoplay_sdk.chat.chatbot import BaseChatbotWriter

class ZendeskChatbot(BaseChatbotWriter):
    def __init__(self, api_token: str, client, **kwargs):
        super().__init__(**kwargs)
        self._api_token = api_token
        self._client = client

    async def _post_note(self, conversation_id: str, body: str) -> str | None:
        resp = await self._client.post(
            f"https://api.zendesk.com/tickets/{conversation_id}/comments",
            json={"comment": {"body": body, "public": False}},
            headers={"Authorization": f"Bearer {self._api_token}"},
        )
        resp.raise_for_status()
        return str(resp.json()["comment"]["id"])

    async def _redact_part(self, conversation_id: str, part_id: str) -> None:
        pass  # Zendesk comments are immutable; no-op

Example — Generic HTTP endpoint (self-hosted bot bridge)

from autoplay_sdk.chat.chatbot import BaseChatbotWriter

class HttpBridgeWriter(BaseChatbotWriter):
    def __init__(self, endpoint: str, client, **kwargs):
        super().__init__(**kwargs)
        self._endpoint = endpoint
        self._client = client

    async def _post_note(self, conversation_id: str, body: str) -> str | None:
        resp = await self._client.post(
            f"{self._endpoint}/conversations/{conversation_id}/note",
            json={"body": body},
        )
        resp.raise_for_status()
        return resp.json().get("id")

    async def _redact_part(self, conversation_id: str, part_id: str) -> None:
        resp = await self._client.delete(
            f"{self._endpoint}/conversations/{conversation_id}/note/{part_id}"
        )
        resp.raise_for_status()

Using with `AsyncAgentContextWriter`

BaseChatbotWriter already debounces write_actions() calls internally via post_link_debounce_s. When wiring an AsyncAgentContextWriter to a BaseChatbotWriter subclass, keep debounce_ms=0 (the default) — the base class debounce is sufficient, and stacking both windows only adds latency.

from autoplay_sdk.context.agent_context import AsyncAgentContextWriter
from autoplay_sdk import AsyncSessionSummarizer

chatbot = ZendeskChatbot(
    product_id="my_product",
    api_token="...",
    client=http_client,
    post_link_debounce_s=0.15,
)

# Link sessions when a conversation opens:
# await chatbot.on_session_linked(session_id, conversation_id)

summarizer = AsyncSessionSummarizer(llm=my_llm, threshold=20)

writer = AsyncAgentContextWriter(
    summarizer=summarizer,
    write_actions=lambda sid, text: chatbot.write_actions("", sid, _parse(text)),
    overwrite_with_summary=my_overwrite_cb,
    debounce_ms=0,  # BaseChatbotWriter already debounces — keep this at 0
)

Avoid double-debouncing. Setting debounce_ms > 0 on AsyncAgentContextWriter while using a BaseChatbotWriter subclass stacks two debounce windows and adds unnecessary latency with no additional reduction in API calls.

API reference

Method	Description
`write_actions(conversation_id, session_id, slim_actions, ...)`	Route actions to pre-link buffer or post-link debounce pipeline
`on_session_linked(session_id, conversation_id)`	Store the session→conversation mapping and flush pre-link buffer
`_post_note(conversation_id, body)`	Subclass contract — post a note; return its id or `None`
`_redact_part(conversation_id, part_id)`	Subclass contract — delete/blank a posted note (best-effort)
`_format_note(session_id, slim_actions, bin_seconds=None)`	Builds the action-note body described in Note body format above (header, sorted 1-based lines, optional binning)

AgentContextWriter — LLM summarisation and push delivery; pairs with BaseChatbotWriter for the full pipeline
Typed payloads — ActionsPayload and SlimAction — the typed models your callbacks receive
Intercom integration — the built-in BaseChatbotWriter subclass for Intercom
Agent states — SessionState FSM reference — THINKING, PROACTIVE, REACTIVE

Documentation Index

​AutoplayChatbotManager (recommended)

​Production: Redis-backed persistence

​What it does

​Pre-link buffering (sliding window)

​At-link flush (one API call)

​Post-link debounce (trailing edge)

​Session-first linking

​Event semantics

​Canonical dual-write path (low-level)

​Hot-path routing helper

​Note body format

​Header (always)

​Action lines

​Binning

​Example (actions note)

​Summary notes (LLM)

​Constructor

​BaseChatbotWriter(product_id, pre_link_window_s=120, post_link_debounce_s=0.15, bin_seconds=3)

​Building a custom backend

​_post_note(conversation_id, body) → str | None

​_redact_part(conversation_id, part_id) → None

​Example — Zendesk

​Example — Generic HTTP endpoint (self-hosted bot bridge)

​Using with AsyncAgentContextWriter

​API reference

​Related

AutoplayChatbotManager (recommended)

Production: Redis-backed persistence

What it does

Pre-link buffering (sliding window)

At-link flush (one API call)

Post-link debounce (trailing edge)

Session-first linking

Event semantics

Canonical dual-write path (low-level)

Hot-path routing helper

Note body format

Header (always)

Action lines

Binning

Example (actions note)

Summary notes (LLM)

Constructor

`BaseChatbotWriter(product_id, pre_link_window_s=120, post_link_debounce_s=0.15, bin_seconds=3)`

Building a custom backend

`_post_note(conversation_id, body) → str | None`

`_redact_part(conversation_id, part_id) → None`

Example — Zendesk

Example — Generic HTTP endpoint (self-hosted bot bridge)

Using with `AsyncAgentContextWriter`

API reference

Related