> ## Documentation Index
> Fetch the complete documentation index at: https://developers.autoplay.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# BaseChatbotWriter

> Base class for delivering session events to any support AI agent platform — handles pre-link buffering, at-link flush, and post-link debouncing so you only implement the API call.

`BaseChatbotWriter` is the delivery engine for any support AI agent 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`](#autoplaychatbotmanager-recommended) — 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`.

```python theme={null}
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 support AI agent 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. Support AI agent 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 support AI agent 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`:

```python theme={null}
# 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 support AI agent 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:

```python theme={null}
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):

```python theme={null}
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](/integrations/intercom) does this for summary posts.

<Tip>
  Import the helper from the package root: `from autoplay_sdk import format_chatbot_note_header` (also available from `autoplay_sdk.chat.chatbot`).
</Tip>

***

## Constructor

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

<ParamField path="product_id" type="str" required>
  Product identifier used in logs and metrics.
</ParamField>

<ParamField path="pre_link_window_s" type="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).
</ParamField>

<ParamField path="post_link_debounce_s" type="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.
</ParamField>

<ParamField path="bin_seconds" type="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`.
</ParamField>

***

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

```python theme={null}
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)

```python theme={null}
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.

```python theme={null}
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
)
```

<Warning>
  **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.
</Warning>

***

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

***

## Related

* [AgentContextWriter](/sdk/agent-context) — LLM summarisation and push delivery; pairs with `BaseChatbotWriter` for the full pipeline
* [Typed payloads](/sdk/typed-payloads) — `ActionsPayload` and `SlimAction` — the typed models your callbacks receive
* [Intercom integration](/integrations/intercom) — the built-in `BaseChatbotWriter` subclass for Intercom
* [Agent states](/sdk/agent-states) — `SessionState` FSM reference — THINKING, PROACTIVE, REACTIVE