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
| Event | Trigger |
|---|---|
| Conversation Created | Fires when a new conversation is created |
| Conversation Closed | Fires when a conversation is closed (by agent, bot, or automation) |
| Conversation Handed Off | Fires when the bot hands off a conversation to a human agent |
| CSAT Sent | Fires after a CSAT request email is successfully sent; context["args"]["csat"]["channel"] identifies the channel |
| Conversation Rated | Fires when the customer submits a CSAT score; context["args"]["rating"] contains the score and optional comment |
| Bot Message Sent | Fires each time the bot sends a message |
| Customer Message Sent | Fires each time the customer sends a message |
| Agent Message Sent | Fires each time a human agent sends a message |
| Campaign Recipient Unreachable | Fires when a phone-campaign recipient goes terminally failed — dial attempts exhausted, or the bot reached voicemail |
| Campaign Call Completed | Fires when a connected phone-campaign call finishes and its transcript is available. Voicemail outcomes fire Campaign Recipient Unreachable instead |
| Campaign Pre Call | Fires before each phone-campaign dial attempt. Can skip the recipient or override its prompt context |
| Rate Limit Hit | Fires when a rate limit is exceeded and incoming traffic is being dropped |
| Bot QA Result | Fires when an automated QA review of a bot-handled conversation finishes |
| Tag Modified | Fires 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:
| Field | Description |
|---|---|
tag | The tag's title |
change | added or removed |
source | Where 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
- You create an event handler — give it a name, select the event type, and write your Python code
- When the event occurs on any conversation in your organization, the handler runs automatically
- The handler receives full conversation context — messages, metadata, and the event type
- 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"]:
| Field | Description |
|---|---|
source | Which channel was being protected — see the table below |
scope | What was being counted: email, domain, business, sender, conversation, browser_session, ip |
limit | How many were allowed in the window |
window_seconds | Length of the window, in seconds |
details | Identifying details for the specific counter that overflowed. Contents vary by source |
Supported source values and what details contains for each:
source | Fires when | details |
|---|---|---|
inbound_email | An incoming email was dropped | email, domain |
contact_form | A contact form submission was dropped | email, domain |
meta_inbound_message | A Messenger / Instagram / WhatsApp message was dropped | channel, senderId |
web_chat_send_message | A web chat message was rejected | conversationId, browserSessionId |
web_chat_new_conversation | A new web chat was rejected before it started | browserSessionId, 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"]:
| Field | Description |
|---|---|
overall | The grade: clean, minor-only, major, or critical |
summary | A short written summary of the review |
issues | List of everything that went wrong — empty when overall is clean. See below |
critical_count | How many issues are critical. Also major_count and minor_count |
trigger | What queued the review: handoff, closed, or backfill (a re-review of older conversations) |
checks_presented | How many QA checks were applied to this conversation |
checks_passed | How many of them passed. Also checks_not_applicable for checks that didn't apply |
judgment_id | Stable ID for this review — useful as a deduplication key |
Each entry in issues has:
| Field | Description |
|---|---|
kind | violation (a specific QA check failed) or finding (a problem outside your checks) |
severity | critical, major, or minor |
check_title | The check that failed. None for findings |
check_key | The check's short key, e.g. U3. None for findings |
category | For findings, what kind of problem it was. None for violations |
evidence | The quoted bot output that triggered the issue |
justification | Why 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.
- Open your event handler in the dashboard
- Enter a Conversation ID — this loads the real conversation data as context. Use any conversation ID or public ID from your dashboard
- Click Run Test
- 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, readcontext["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_metadatato 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.