🚀 Quickstart

import posthog from 'posthog-js'

posthog.init('YOUR_POSTHOG_PROJECT_API_KEY', {
    api_host: 'https://us.i.posthog.com',
    person_profiles: 'identified_only',
    // This stops recording if a user walks away for 2 minutes, keeping your data clean:
    session_idle_timeout_seconds: 120, 
    loaded: (posthog) => {
        posthog.identify(posthog.get_distinct_id(), {
            // Autoplay-specific field. Must match onboard_product(product_id=...).
            product_id: 'YOUR_PRODUCT_ID', 
        });
    },
})

posthog.identify(user.id, {
    product_id: 'YOUR_AUTOPLAY_PRODUCT_ID',
    email: user.email, // Added to the person profile and the $identify event.
})

// If you want payload.email on ongoing autocapture events, also register email as a super-property.
posthog.register({ email: user.email })

pip install autoplay-sdk
# OR
uv add autoplay-sdk

import asyncio
from autoplay_sdk.admin import onboard_product

async def main() -> None:
    result = await onboard_product(
        "YOUR_AUTOPLAY_PRODUCT_ID",
        contact_email="you@yourcompany.com",
        print_operator_summary=True,
    )
    # result still has the same full fields if you need them in code

asyncio.run(main())

fun extractFromElementsChain(str, pattern) {
    try {
        if (empty(str)) {
            return ''
        }
        let startIdx := position(str, pattern)
        if (startIdx <= 0) {
            return ''
        }
        let sub := substring(str, startIdx + length(pattern), length(str) - startIdx - length(pattern) + 1)
        let endIdx := position(sub, '"')
        if (endIdx > 0) {
            return substring(sub, 1, endIdx - 1)
        }
        return ''
    } catch (err) {
        print(f'extractFromElementsChain error for pattern {pattern}:', err)
        return ''
    }
}


let elements_chain := event.elements_chain ?? ''

let element_id := ''
let input_field_name := ''
let link_destination := ''
let button_or_link_text := ''

try {
    element_id := extractFromElementsChain(elements_chain, 'attr__id="')
} catch (err) {
    print('Error extracting element_id:', err)
    element_id := ''
}
try {
    input_field_name := extractFromElementsChain(elements_chain, 'attr__name="')
} catch (err) {
    print('Error extracting input_field_name:', err)
    input_field_name := ''
}
try {
    link_destination := extractFromElementsChain(elements_chain, 'attr__href="')
} catch (err) {
    print('Error extracting link_destination:', err)
    link_destination := ''
}

try {
    button_or_link_text := extractFromElementsChain(elements_chain, 'text="')
} catch (err) {
    print('Error extracting button_or_link_text:', err)
    button_or_link_text := ''
}

let payload := {
    'event': event.event,
    'referrer': event.properties?.$referrer ?? '',
    'email': event.properties?.email ?? event.person?.properties?.email ?? '',
    'timestamp': event.timestamp ?? '',
    'element_id': element_id,
    'event_type': event.properties?.$event_type ?? '',
    'session_id': event.properties?.$session_id ?? '',
    'current_url': event.properties?.$current_url ?? '',
    'distinct_id': event.distinct_id ?? '',
    'elements_chain': elements_chain,
    'input_field_name': input_field_name,
    'link_destination': link_destination,
    'button_or_link_text': button_or_link_text
}


let headers := {
    'Content-Type': 'application/json',
    'x-posthog-secret': '{webhook_secret_from_step_3}'
}

let req := {
    'headers': headers,
    'body': jsonStringify(payload),
    'method': 'POST'
}

let url := '{webhook_url_from_step_3}'

if (inputs.debug) {
    print('Request payload', payload)
    print('Request', url, req)
}

let res := fetch(url, req)


if (res.status >= 400) {
    print('Webhook error response', res.status, res.body)
    throw Error(f'Webhook returned {res.status}: {res.body}')
}
if (inputs.debug) {
    print('Response', res.status, res.body)
}

import asyncio
from autoplay_sdk import AsyncConnectorClient

# Paste your Stream URL and Token here!
STREAM_URL = "https://your-connector.onrender.com/stream/YOUR_PRODUCT_ID"
API_TOKEN  = "unkey_xxxx..."
# You can also set AUTOPLAY_APP_UNKEY_TOKEN in your environment.

def on_actions(p):
    print("\n=== LIVE USER ACTIONS ===")
    print(f"  Session ID   : {p.session_id}")
    print(f"  User ID      : {p.user_id}")
    print(f"  Product ID   : {p.product_id}")
    print(f"  Total Clicks : {p.count}")
    print(f"  Time Received: {p.forwarded_at}")
    for i, action in enumerate(p.actions):
        print(f"  [{i}] Action: {action.title}")
        print(f"       Details: {action.description}")
        print(f"       Page:    {action.canonical_url}")

def on_summary(p):
    print("\n=== AI SUMMARY ===")
    print(f"  Session ID   : {p.session_id}")
    print(f"  Product ID   : {p.product_id}")
    print(f"  Summarizes   : {p.replaces} actions")
    print(f"  Summary text : {p.summary}")

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

asyncio.run(main())

=== ACTIONS ===
  session_id   : ps_abc123
  user_id      : user_xyz
  product_id   : acme-corp
  count        : 3
  forwarded_at : 1736940685.103
  [0] title        : Page Load: Dashboard
       description  : User landed on the main Dashboard page
       canonical_url: <https://app.example.com/dashboard>
  [1] title        : Click Export CSV
       description  : User clicked the Export CSV button
       canonical_url: <https://app.example.com/dashboard>
  [2] title        : Click Settings
       description  : User clicked the Settings link in the sidebar
       canonical_url: <https://app.example.com/dashboard>

=== SUMMARY ===
  session_id   : ps_abc123
  product_id   : acme-corp
  replaces     : 12 actions
  forwarded_at : 1736940750.881
  summary      : User explored the Dashboard, exported a CSV, and navigated to billing settings.