Step 1 — Connect real-time events

title: Click element
description: User clicked element on the pricing page
canonical_url: https://yourapp.com/pricing
user_id: 019dd835-3f68-7e78
session_id: 019dd958-7eea-75a2

The user navigated to the pricing page and interacted with several
diffusion model options, suggesting they are configuring an image
generation workflow.

const filter = { id: 1 } // filter for the single summary row
const summaryRecord = await AutoPlayEventsSummaryTable.findRecords({ filter })
console.log('summary', summaryRecord)
const summary = summaryRecord[0].summary

// Events from the last 2 minutes
const twoMinutesAgo = new Date(Date.now() - 2 * 60 * 1000).toISOString()
const recentEvents = await AutoPlayEventsTable.findRecords({
  filter: { createdAt: { $gte: twoMinutesAgo } }
})
console.log('recentEvents', JSON.stringify(recentEvents))

const result = {
  recentEvents,
  summary: summary && summary.length > 0 ? summary[0] : null
}
console.log('result', result)

workflow.agentContext = result

const body = event.payload.body
console.log('jsonBody:', JSON.stringify(body))

const records = body.actions.map((action) => ({
  title: action.title,
  description: action.description,
  canonical_url: action.canonical_url,
  user_id: body.user_id,
  product_id: body.product_id,
  session_id: body.session_id,
}))

await AutoPlayEventsTable.createRecords(records)
console.log('Records created:', records.length)

You are a proactive software adoption assistant. Your role is to help users
succeed with the product by understanding exactly where they are in their
journey — based on their live session activity and a rolling summary of
their recent behaviour.

You have access to two data sources:
- **recentEvents**: A list of actions the user has taken in the last 2 minutes
  (page visits, clicks, feature interactions, form submissions, etc.)
- **summary**: A rolling narrative of what the user has been doing this session
  (may be null if the session is new)

Here is the user's current session context:
<PLACEHOLDER: PLACE MOUSE CURSOR HERE>

When responding:
- Identify which part of the product the user is currently working in, and
  what they appear to be trying to accomplish
- Spot signs of confusion or friction (e.g. repeated visits to the same page,
  abandoning a flow, clicking around without completing an action)
- Offer specific, actionable guidance that meets them exactly where they are —
  not generic help documentation
- If they are stuck, proactively suggest the next step or the feature that
  would unblock them
- Keep responses concise and practical — users are mid-task, not reading docs
- If no session data is available, ask a clarifying question to understand
  what they are trying to do

Your north star is reducing time-to-value: every response should move the
user one step closer to completing their goal inside the product.

import asyncio
from autoplay_sdk import (
    ActionsPayload,
    AsyncAgentContextWriter,
    AsyncConnectorClient,
    AsyncSessionSummarizer,
)
import httpx
import requests
import json
import openai

# ── Autoplay ──────────────────────────────────────────────────────────────
STREAM_URL = "https://your-connector.onrender.com/stream/<your-product-id>"
UNKEY_API_KEY = "your-unkey-api-key"

# ── Botpress ──────────────────────────────────────────────────────────────
BOTPRESS_ACCESS_TOKEN = "your-botpress-access-token"
BOT_ID = "your-bot-id"
BOTPRESS_WEBHOOK_URL = "your-botpress-webhook-endpoint-url"

botpress_client = httpx.AsyncClient(
    base_url="https://api.botpress.cloud/v1/tables",
    headers={
        "Authorization": f"Bearer {BOTPRESS_ACCESS_TOKEN}",
        "Accept": "application/json",
        "x-bot-id": BOT_ID,
    },
)

# ── OpenAI (used by the summarizer) ───────────────────────────────────────
async_openai = openai.AsyncOpenAI()


def api_call(func):
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            print(f"Error in API call: {e}")
    return wrapper


@api_call
async def llm(prompt: str) -> str:
    r = await async_openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
        max_tokens=256,
    )
    return r.choices[0].message.content

@api_call
async def post_actions_to_webhook(payload: ActionsPayload) -> None:
    print("\n=== POSTING ACTIONS TO WEBHOOK ===")
    data = {
        "session_id": payload.session_id,
        "user_id": payload.user_id,
        "product_id": payload.product_id,
        "actions": [
            {
                "title": action.title,
                "description": action.description,
                "canonical_url": action.canonical_url,
            }
            for action in payload.actions
        ],
    }
    response = requests.post(
        BOTPRESS_WEBHOOK_URL,
        data=json.dumps(data),
        headers={"Content-Type": "application/json"},
    )
    print(f"Webhook response status: {response.status_code}")


async def dummy_write_actions(session_id: str, text: str) -> None:
    pass  # raw payload is already sent via post_actions_to_webhook


@api_call
async def overwrite_summary(session_id: str, summary: str) -> None:
    print("\n=== OVERWRITING SUMMARY IN BOTPRESS TABLE ===")
    payload = {
        "rows": [{"id": 1, "session_id": session_id, "summary": summary}],
        "keyColumn": "id",
        "waitComputed": True,
    }
    response = await botpress_client.post(
        "/AutoPlayEventsSummaryTable/rows/upsert",
        json=payload,
        headers={
            "x-bot-id": BOT_ID,
            "Authorization": f"Bearer {BOTPRESS_ACCESS_TOKEN}",
            "Content-Type": "application/json",
        },
    )
    print(f"Summary upsert response status: {response.status_code}")

summarizer = AsyncSessionSummarizer(llm=llm, threshold=10)

agent_writer = AsyncAgentContextWriter(
    summarizer=summarizer,
    write_actions=dummy_write_actions,
    overwrite_with_summary=overwrite_summary,
    debounce_ms=0,
)


async def handle_actions_interceptor(payload: ActionsPayload):
    # 1. Send the raw structured JSON to the Botpress Webhook
    await post_actions_to_webhook(payload)
    # 2. Forward to the agent_writer so it counts toward the summarizer threshold
    await agent_writer.add(payload)

async def main():
    async with AsyncConnectorClient(url=STREAM_URL, token=UNKEY_API_KEY) as client:
        client.on_actions(handle_actions_interceptor)
        print("Listening for live website clicks… (Press Ctrl+C to stop)")
        await client.run()


asyncio.run(main())