AI Knowledge & Logic

Event Handlers

Run custom Python code automatically when conversation events occur — close, handoff, and more. Use event handlers for CRM updates, notifications, analytics, and post-conversation workflows.

Event handlers let you run Python code automatically when specific events happen in your conversations. Unlike custom actions (which the AI calls during a conversation), event handlers trigger after key moments — when a conversation is closed, when it's handed off to a human, and more in the future.

Common uses:

  • Update your CRM when a conversation closes
  • Send a Slack notification when a conversation is handed off
  • Log conversation data to an external analytics system
  • Trigger a follow-up email or survey after resolution
  • Alert your team when a rate limit starts dropping incoming traffic
  • Ping a Slack channel when an automated QA review flags a bot reply as critical

An event handler can be Python or a webhook. Settings → Automation → Event Handlers covers both: Python (this page) and webhooks. You choose which when you create one. They react to the same events. Reach for Python when the reaction needs logic — deciding, transforming, calling several systems, writing back to the conversation. Reach for a webhook when all you need is "send this event somewhere": it's pure configuration, and comes with signing, automatic retries, a delivery log, and filtering on channel, tags, metadata and handoff state that you'd otherwise write by hand.


Supported events

EventTrigger
Conversation CreatedFires when a new conversation is created
Conversation ClosedFires when a conversation is closed (by agent, bot, or automation)
Conversation Handed OffFires when the bot hands off a conversation to a human agent
CSAT SentFires after a CSAT request email is successfully sent; context["args"]["csat"]["channel"] identifies the channel
Conversation RatedFires when the customer submits a CSAT score; context["args"]["rating"] contains the score and optional comment
Bot Message SentFires each time the bot sends a message
Customer Message SentFires each time the customer sends a message
Agent Message SentFires each time a human agent sends a message
Campaign Recipient UnreachableFires when a phone-campaign recipient goes terminally failed — dial attempts exhausted, or the bot reached voicemail
Campaign Call CompletedFires when a connected phone-campaign call finishes and its transcript is available. Voicemail outcomes fire Campaign Recipient Unreachable instead
Campaign Pre CallFires before each phone-campaign dial attempt. Can skip the recipient or override its prompt context
Rate Limit HitFires when a rate limit is exceeded and incoming traffic is being dropped
Bot QA ResultFires when an automated QA review of a bot-handled conversation finishes
Tag ModifiedFires when a tag is added to or removed from a conversation

Each event handler is configured for exactly one event type. The per-message events (Bot / Customer / Agent Message Sent) fire once for every matching message, so keep their handlers lightweight.

Rate Limit Hit is never tied to a conversation — the guards run before one exists. See Reacting to rate limits below.

Tag Modified fires for tag changes made by an agent in the dashboard and through the REST API. context["args"] carries:

FieldDescription
tagThe tag's title
changeadded or removed
sourceWhere the change came from (agent, api, ...)

One handler covers both directions — branch on change rather than writing a pair.

def handle_event(context):
    args = context["args"]

    if args["change"] != "added" or args["tag"] != "needs-refund-review":
        return None

    add_conversation_note(context, "Queued for refund review.")
    return {"queued": True}

Tag changes made by Python — add_conversation_tag and remove_conversation_tag — deliberately do not fire this event. Without that, a handler that tags on Tag Modified would trigger itself indefinitely, and each pass costs a real sandbox execution. The trade-off is that one handler cannot chain into another via tags; call the second handler's logic directly, or move the shared part into a Python module.

The phone-campaign events carry a conversation only when a dial actually connected, so Campaign Recipient Unreachable and Campaign Pre Call handlers must tolerate context["conversation"] being None. The campaign and recipient details arrive in context["args"]["campaign"] regardless.


How it works

  1. You create an event handler — give it a name, select the event type, and write your Python code
  2. When the event occurs on any conversation in your organization, the handler runs automatically
  3. The handler receives full conversation context — messages, metadata, and the event type
  4. Results are logged for debugging and monitoring

You can create multiple handlers for the same event. They all run independently.


Scoping a handler to specific businesses

By default a handler runs for every business in your organization. If you only want it to run for some businesses, set its Scope to "Only specific businesses" and pick them. Businesses that aren't in the list are a no-op — the event fires, but the handler's Python code never runs for them.

This is the same business-whitelist mechanism used by custom actions, and it's the recommended way to limit a handler to a single brand (e.g. send Slack notifications for one business only).


Writing an event handler

An event handler implements a handle_event function. It receives the same context object as custom actions.

import requests

def handle_event(context):
    event_type = context["args"]["event_type"]
    conversation = context["conversation"]

    # Post to your webhook
    requests.post("https://api.example.com/webhook", json={
        "event": event_type,
        "conversationId": conversation["publicId"],
        "subject": conversation["subject"],
        "messageCount": len(conversation["messages"]),
    }, timeout=30)

    return {"status": "notified"}

Context structure

The context object contains conversation data, customer profile, business info, and the event type via context["args"]["event_type"]. See Python Context for the full reference.

context["conversation"] is present for every conversation event. It is None for Rate Limit Hit, which fires before any conversation exists, and for phone-campaign events where no dial connected — always guard before using it if your handler covers one of those events.


Available helper functions

Event handlers have access to all the same built-in helper functions as custom actions, condition providers, and sidebar widgets — including conversation data, LLM classification, integrations, and more.

See Python Helpers for the full list with documentation links.


Example: CRM update on conversation close

Update your CRM with conversation summary data when a conversation is resolved.

import requests

def handle_event(context):
    conversation = context["conversation"]
    messages = conversation["messages"]

    customer_messages = [m for m in messages if m["sender"] == "customer"]

    requests.post("https://api.example.com/crm/conversations", json={
        "externalId": conversation["publicId"],
        "subject": conversation["subject"],
        "messageCount": len(messages),
        "customerMessageCount": len(customer_messages),
        "resolvedAt": messages[-1]["timestamp"] if messages else None,
    }, timeout=30)

    return {"status": "synced"}

Example: Slack notification on handoff

Alert your team when a conversation needs human attention.

import requests

def handle_event(context):
    conversation = context["conversation"]
    messages = conversation["messages"]

    # Get last few customer messages for context
    recent = [m["content"] for m in messages if m["sender"] == "customer"][-3:]

    requests.post("https://hooks.slack.com/services/YOUR/WEBHOOK/URL", json={
        "text": f"Conversation {conversation['publicId']} handed off",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Handoff:* {conversation['subject']}\n*ID:* {conversation['publicId']}\n*Recent messages:*\n" + "\n".join(f"> {m}" for m in recent)
                }
            }
        ]
    }, timeout=10)

    return {"status": "notified"}

Example: Google Sheets logging

Log every closed conversation to a Google Sheet for reporting.

def handle_event(context):
    conversation = context["conversation"]
    messages = conversation["messages"]

    add_google_sheets_row("YOUR_SPREADSHEET_ID", {
        "Conversation ID": conversation["publicId"],
        "Subject": conversation["subject"],
        "Messages": len(messages),
        "Closed At": messages[-1]["timestamp"] if messages else "",
        "Business": conversation["businessSlug"],
    })

    return {"status": "logged"}

Reacting to rate limits

Octocom applies rate limits to incoming traffic to protect your inbox from floods, mailing-list loops, and abuse. When one of these limits is exceeded, the offending message is dropped or rejected — it never becomes a conversation.

The Rate Limit Hit event lets you react to that: page an on-call channel, open a ticket in your own system, or record it for reporting.

The rate_limit object

Everything about the tripped limit arrives under context["args"]["rate_limit"]:

FieldDescription
sourceWhich channel was being protected — see the table below
scopeWhat was being counted: email, domain, business, sender, conversation, browser_session, ip
limitHow many were allowed in the window
window_secondsLength of the window, in seconds
detailsIdentifying details for the specific counter that overflowed. Contents vary by source

Supported source values and what details contains for each:

sourceFires whendetails
inbound_emailAn incoming email was droppedemail, domain
contact_formA contact form submission was droppedemail, domain
meta_inbound_messageA Messenger / Instagram / WhatsApp message was droppedchannel, senderId
web_chat_send_messageA web chat message was rejectedconversationId, browserSessionId
web_chat_new_conversationA new web chat was rejected before it startedbrowserSessionId, ip

context["business"] tells you which business was affected. context["conversation"] is always None for this event.

How often it fires

The event is deduplicated per counter and per window, so a sustained flood produces a steady trickle rather than one event per blocked message. There is also a five-minute floor between repeat events for the same counter, so short-window limits can't fire more than about twelve times an hour each.

Some scopes are noisier than others by design. business scope means real volume is being turned away and is almost always worth acting on. The per-visitor scopes (ip, browser_session, conversation) trip during ordinary use — one shopper opening several chats in a minute is enough — so branch on scope rather than alerting on everything.

Example: alert your team when email is being dropped

import requests

def handle_event(context):
    rate_limit = context["args"]["rate_limit"]

    # Only care about email, and only when it's an org-wide flood
    if rate_limit["source"] not in ("inbound_email", "contact_form"):
        return {"status": "ignored"}
    if rate_limit["scope"] not in ("domain", "business"):
        return {"status": "ignored"}

    business = context["business"]
    details = rate_limit["details"]
    hours = rate_limit["window_seconds"] / 3600

    requests.post("https://hooks.slack.com/services/YOUR/WEBHOOK/URL", json={
        "text": (
            f"Incoming email is being dropped for {business['slug']}\n"
            f"Scope: {rate_limit['scope']} ({details.get('domain', 'n/a')})\n"
            f"Limit: {rate_limit['limit']} per {hours:g}h"
        )
    }, timeout=10)

    return {"status": "notified"}

Example: log every trip to a Google Sheet

from datetime import datetime, timezone

def handle_event(context):
    rate_limit = context["args"]["rate_limit"]
    details = rate_limit["details"]

    add_google_sheets_row("YOUR_SPREADSHEET_ID", {
        "Time": datetime.now(timezone.utc).isoformat(),
        "Business": context["business"]["slug"],
        "Source": rate_limit["source"],
        "Scope": rate_limit["scope"],
        "Limit": rate_limit["limit"],
        "Window (s)": rate_limit["window_seconds"],
        "Who": details.get("email") or details.get("senderId") or details.get("ip") or "",
    })

    return {"status": "logged"}

Reacting to bot QA results

Octocom automatically reviews bot-handled conversations against your QA checks and grades each one. The Bot QA Result event fires once per completed review, so you can alert on failures, log every grade for reporting, or open a ticket when the bot gets something seriously wrong.

Reviews run after a conversation is handed off or closed, not in real time — expect the event minutes to days after the conversation itself, depending on the trigger. It fires for every completed review, including clean ones, so branch on overall rather than assuming a problem.

The bot_qa object

Everything about the review arrives under context["args"]["bot_qa"]:

FieldDescription
overallThe grade: clean, minor-only, major, or critical
summaryA short written summary of the review
issuesList of everything that went wrong — empty when overall is clean. See below
critical_countHow many issues are critical. Also major_count and minor_count
triggerWhat queued the review: handoff, closed, or backfill (a re-review of older conversations)
checks_presentedHow many QA checks were applied to this conversation
checks_passedHow many of them passed. Also checks_not_applicable for checks that didn't apply
judgment_idStable ID for this review — useful as a deduplication key

Each entry in issues has:

FieldDescription
kindviolation (a specific QA check failed) or finding (a problem outside your checks)
severitycritical, major, or minor
check_titleThe check that failed. None for findings
check_keyThe check's short key, e.g. U3. None for findings
categoryFor findings, what kind of problem it was. None for violations
evidenceThe quoted bot output that triggered the issue
justificationWhy it was graded that way

context["conversation"] is always present for this event, so conversation["url"] links straight to the reviewed conversation in the dashboard.

Example: Slack alert on critical QA failures

import requests

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"

def handle_event(context):
    bot_qa = context["args"]["bot_qa"]

    # Only page on critical — everything else is visible in the dashboard
    if bot_qa["overall"] != "critical":
        return {"status": "ignored", "overall": bot_qa["overall"]}

    conversation = context["conversation"]
    critical = [i for i in bot_qa["issues"] if i["severity"] == "critical"]

    lines = []
    for issue in critical:
        title = issue["check_title"] or issue["category"] or "Finding"
        lines.append(f"> *{title}* — {issue['justification']}")

    send_slack_notification(SLACK_WEBHOOK_URL, "\n".join([
        f":rotating_light: Critical bot QA failure — {context['business']['name']}",
        f"<{conversation['url']}|{conversation['subject'] or conversation['publicId']}>",
        *lines,
    ]))

    return {"status": "notified", "criticalCount": len(critical)}

send_slack_notification is a built-in helper — see Python Helpers. You can also post to Slack with requests directly if you want full Block Kit control.

Example: log every review to a Google Sheet

from datetime import datetime, timezone

def handle_event(context):
    bot_qa = context["args"]["bot_qa"]
    conversation = context["conversation"]

    add_google_sheets_row("YOUR_SPREADSHEET_ID", {
        "Time": datetime.now(timezone.utc).isoformat(),
        "Business": context["business"]["slug"],
        "Conversation": conversation["url"],
        "Overall": bot_qa["overall"],
        "Critical": bot_qa["critical_count"],
        "Major": bot_qa["major_count"],
        "Minor": bot_qa["minor_count"],
        "Checks Passed": f"{bot_qa['checks_passed']}/{bot_qa['checks_presented']}",
        "Summary": bot_qa["summary"],
    })

    return {"status": "logged"}

Testing

Event handlers can be tested from the dashboard before they go live.

  1. Open your event handler in the dashboard
  2. Enter a Conversation ID — this loads the real conversation data as context. Use any conversation ID or public ID from your dashboard
  3. Click Run Test
  4. The dashboard executes your code and shows the result, stdout, and stderr

The test simulates the event without actually triggering it. The selected event type from the form is passed as context["args"]["event_type"].

Note: Testing gives your code a real conversation but no event-specific args, so it can't reproduce a Rate Limit Hit or Bot QA Result payload. To exercise those handlers, read context["args"].get("rate_limit") / context["args"].get("bot_qa") and fall back to a sample value when it's missing.

Tip: Test with conversations that represent different scenarios — short conversations, long ones, ones with file attachments — to make sure your handler is robust.


Best practices

  • Keep handlers fast. Event handlers run asynchronously but have a 60-second timeout. Avoid heavy processing — offload to external systems if needed.
  • Handle errors gracefully. If your external API is down, catch the exception and return an error status rather than crashing. Failed executions are logged for debugging.
  • Use metadata for state. If you need to track whether a handler has already processed a conversation, use set_conversation_metadata / get_conversation_metadata to avoid duplicate processing.
  • Be idempotent. Events may occasionally trigger more than once. Design your handlers so running them twice on the same conversation produces the same result.
  • Check the event type. If you plan to reuse similar logic across events, check context["args"]["event_type"] to branch behavior.

On this page