Intercom

​

Module: autoplay_sdk.integrations.intercom

​

Webhook topics

​

intercom_chatbot_webhook_url(connector_host, product_id) -> str

• connector_host: hostname (event-connector-xxxx.onrender.com) or full origin (https://…). Trailing slashes stripped; bare host gets https:// prepended.
• product_id: non-empty, no / (single path segment).

​

format_reactive_session_link_script(connector_host, product_id) -> str

​

Proactive Messenger quick replies

​

Prerequisites

• conversation_id — From Intercom webhooks, SlimAction conversation_id, or after POST /sessions/link links the browser session.
• Access token — Bearer token for the Intercom API (app allowed to reply as admin).
• admin_id — Workspace/agent id Intercom expects on quick_reply payloads (from Intercom app settings).

​

Recipe: end-to-end

1. Build context — Instantiate ProactiveTriggerContext with chronological canonical_urls (and session_id, conversation_id, action_count, product_id when you have them).
2. Evaluate — result = registry.evaluate_first(ctx) using default_proactive_trigger_registry() or your own ProactiveTriggerRegistry([...]). If result is None, do not send.
3. Gate (recommended) — Cooldown: skip if now - last_fired_at < result.cooldown_s for (conversation_id, result.trigger_id) (you store last_fired_at). FSM: with Agent session states, call can_show_proactive_with_reason() before sending; after a successful send, move to proactive_assistance. When the user never engages past result.interaction_timeout_s, use run_proactive_idle_expiry for chat surfaces (not expire_proactive_to_thinking_if_idle alone): implement ProactiveIdleExpiryHooks — delete_remote_chat_thread performs DELETE /conversations/{id} via build_intercom_delete_conversation_request, returns True only on 2xx or 404; then the orchestrator runs expire_proactive_to_thinking_if_idle and clear_local_chat_thread_state (session↔thread + persisted FSM — your host). See Delete conversation (idle teardown).
4. Build the POST — headers = intercom_quick_reply_http_headers(access_token). payload = build_intercom_quick_reply_reply_payload(admin_id=..., body=result.body, prompt_labels=list(result.reply_option_labels)). Empty prompt_labels is fine (intro-only).
5. Send — POST {INTERCOM_REST_API_BASE}/conversations/{conversation_id}/reply with json=payload and headers.

​

Delete conversation (idle teardown)

• build_intercom_delete_conversation_request(access_token, conversation_id, *, retain_metrics=True) -> tuple[str, dict[str, str]] — returns (url, headers) for client.delete(url, headers=headers).
• Lower-level: intercom_delete_conversation_url, intercom_delete_conversation_headers.

​

Example (Python)

from autoplay_sdk.integrations.intercom import (
    INTERCOM_PROACTIVE_QUICK_REPLY_DEFAULT_BODY,
    INTERCOM_REST_API_BASE,
    build_intercom_quick_reply_reply_payload,
    intercom_quick_reply_http_headers,
    proactive_trigger_canonical_url_ping_pong,
)
from autoplay_sdk.proactive_triggers import (
    ProactiveTriggerContext,
    default_proactive_trigger_registry,
)

registry = default_proactive_trigger_registry()

def maybe_send_proactive_quick_reply(
    *,
    access_token: str,
    admin_id: str,
    conversation_id: str,
    canonical_urls: list[str | None],
    session_id: str = "",
) -> bool:
    """Returns True if a quick_reply was sent (after your cooldown/FSM checks)."""
    ctx = ProactiveTriggerContext(
        canonical_urls=canonical_urls,
        conversation_id=conversation_id,
        session_id=session_id,
    )
    result = registry.evaluate_first(ctx)
    if result is None:
        return False
    # Apply cooldown using result.cooldown_s and result.trigger_id.
    # Optionally gate with AgentStateMachine.can_show_proactive_with_reason().
    payload = build_intercom_quick_reply_reply_payload(
        admin_id=admin_id,
        body=result.body,
        prompt_labels=list(result.reply_option_labels),
    )
    headers = intercom_quick_reply_http_headers(access_token)
    url = f"{INTERCOM_REST_API_BASE}/conversations/{conversation_id}/reply"
    # requests.post(url, json=payload, headers=headers, timeout=30)
    return True


def minimal_ping_pong_only(
    *,
    access_token: str,
    admin_id: str,
    conversation_id: str,
    urls: list[str | None],
) -> bool:
    """Same POST shape without the proactive registry — predicate + default body only."""
    if not proactive_trigger_canonical_url_ping_pong(urls):
        return False
    payload = build_intercom_quick_reply_reply_payload(
        admin_id=admin_id,
        body=INTERCOM_PROACTIVE_QUICK_REPLY_DEFAULT_BODY,
        prompt_labels=[],
    )
    headers = intercom_quick_reply_http_headers(access_token)
    _url = f"{INTERCOM_REST_API_BASE}/conversations/{conversation_id}/reply"
    return True

​

Helpers reference (delivery)

​

Package: autoplay_sdk.proactive_triggers

​

Adding your own trigger

1. Implement ProactiveTrigger: trigger_id and evaluate(ctx) -> ProactiveTriggerResult | None.
2. Optionally wrap with ProactiveTriggerEntity(inner, ProactiveTriggerTimings(...)) for custom timeouts / cooldown.
3. Register on ProactiveTriggerRegistry in priority order; call evaluate_first in production.

​

Optional: connector LLM-generated button labels

​

Connector endpoints (reference)

​

Delivery stack in this repo

• BaseChatbotWriter (Support agent writer) — pre-link buffer, post-link debounce, shared note body format (format_chatbot_note_header, numbered action lines, binning).
• IntercomChatbot (event connector flows/chatbot/intercom.py) — subclass; implements Intercom REST admin notes (_post_note, _redact_part) and webhook payload parsing for linking.

​

Related

• Support agent writer — BaseChatbotWriter contract and note format.
• Agent context — summariser + overwrite_with_summary flow.
• Agent session states — FSM gates when using the optional proactive connector endpoint with conversation_id.