Bot Lifecycle Hooks

Observe and control your messaging bot’s gateway, session, and schedule lifecycle

Bot platform adapters now ship in the praisonai-bot package. praisonai bot serve still works exactly as documented here; for a standalone install see praisonai-bot Migration.

Bot lifecycle hooks let you watch your bot start and stop, see every user session begin and end, and react when scheduled jobs fire — without changing your agent code.

from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="You are a helpful messaging bot.")
agent.start("Log when sessions start and end.")

This page covers in-process outbound lifecycle hooks (GATEWAY_START, SESSION_START, SCHEDULE_TRIGGER) that fire inside the running gateway. For HTTP inbound triggers that let external services start agent runs via POST /hooks/<path>, see Gateway Inbound Hooks.

The user opens a chat session; lifecycle hooks fire on gateway start, session boundaries, and scheduled jobs.

Quick Start

1. Register a SESSION_START hook

   import os
   from praisonaiagents import Agent
   from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult
   
   registry = HookRegistry()
   
   @registry.on(HookEvent.SESSION_START)
   def on_session_start(event_data):
       print(f"New session: {event_data.session_id} on {event_data.session_name}")
       return HookResult.allow()
   
2. Pass the registry to your agent and bot

   from praisonai.bots import TelegramBot
   
   agent = Agent(
       name="MyBot",
       instructions="You are a helpful assistant.",
       hooks=registry
   )
   
   bot = TelegramBot(
       token=os.getenv("TELEGRAM_BOT_TOKEN"),
       agent=agent
   )
   
   import asyncio
   asyncio.run(bot.start())
   

   Running this bot now prints a line whenever a user opens a new session.

---

How It Works

BEFORE_AGENT and AFTER_AGENT are fired by agent.chat() itself — the gateway does not re-fire them to avoid double-dispatch to plugins.

---

Events Fired by the Bot Runtime

Event	When	Input Type	Key Fields
GATEWAY_START	BotOS.start()	GatewayStartInput	platforms, bot_count
GATEWAY_STOP	BotOS.stop()	GatewayStopInput	platforms, bot_count, reason
SESSION_START	First message per user (once per session lifetime)	SessionStartInput	session_id, agent_name, source, session_name
SESSION_END	/new reset, policy auto-reset, stale reap, reset_all	SessionEndInput	session_id, agent_name, reason
SCHEDULE_TRIGGER	Scheduled job runs in BotOS._execute_schedule_job. Fires exactly once per job even when multiple BotOS processes share the same store (see BotOS → Multi-Process / HA Deployments).	ScheduleTriggerInput	job_name, job_id, message
MESSAGE_RECEIVED	Incoming message from platform, before agent dispatch — inbound gate: deny drops the message, modified_input["content"] rewrites it	MessageReceivedInput	platform, content, sender_id
MESSAGE_SENDING	Before bot sends a reply — outbound gate: deny cancels sending, modified_input["content"] rewrites it	MessageSendingInput	platform, content, channel_id
MESSAGE_SENT	After bot successfully sends a reply	MessageSentInput	platform, content, message_id

SESSION_END reason values:

• clear — user sent /new
• policy — policy auto-reset triggered
• stale — session reaped due to inactivity
• clear_all — reset_all called (clears all sessions)

---

Common Patterns

Audit Log

Log every gateway and session event to a file with timestamps.

import os
import datetime
from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult
from praisonai.bots import TelegramBot

registry = HookRegistry()
LOG_FILE = "bot_audit.log"

def _log(msg: str) -> None:
    ts = datetime.datetime.utcnow().isoformat()
    with open(LOG_FILE, "a") as f:
        f.write(f"[{ts}] {msg}\n")

@registry.on(HookEvent.GATEWAY_START)
def audit_gateway_start(event_data):
    _log(f"GATEWAY_START platforms={event_data.platforms} bots={event_data.bot_count}")
    return HookResult.allow()

@registry.on(HookEvent.SESSION_START)
def audit_session_start(event_data):
    _log(f"SESSION_START session={event_data.session_id} platform={event_data.session_name}")
    return HookResult.allow()

@registry.on(HookEvent.SESSION_END)
def audit_session_end(event_data):
    _log(f"SESSION_END session={event_data.session_id} reason={event_data.reason}")
    return HookResult.allow()

@registry.on(HookEvent.GATEWAY_STOP)
def audit_gateway_stop(event_data):
    _log(f"GATEWAY_STOP platforms={event_data.platforms} reason={event_data.reason}")
    return HookResult.allow()

agent = Agent(name="AuditBot", instructions="Be helpful.", hooks=registry)
bot = TelegramBot(token=os.getenv("TELEGRAM_BOT_TOKEN"), agent=agent)

import asyncio
asyncio.run(bot.start())

Per-User Usage Counter

Increment a counter when a session starts, persist it when the session ends.

import os
from collections import defaultdict
from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult
from praisonai.bots import TelegramBot

registry = HookRegistry()
session_counts: dict = defaultdict(int)

@registry.on(HookEvent.SESSION_START)
def count_session(event_data):
    session_counts[event_data.agent_name] += 1
    print(f"[{event_data.agent_name}] total sessions: {session_counts[event_data.agent_name]}")
    return HookResult.allow()

@registry.on(HookEvent.SESSION_END)
def persist_count(event_data):
    print(f"Session {event_data.session_id} ended ({event_data.reason}), "
          f"total for {event_data.agent_name}: {session_counts[event_data.agent_name]}")
    return HookResult.allow()

agent = Agent(name="CounterBot", instructions="Be helpful.", hooks=registry)
bot = TelegramBot(token=os.getenv("TELEGRAM_BOT_TOKEN"), agent=agent)

import asyncio
asyncio.run(bot.start())

Scheduled-Job Observability

Push a metric every time a scheduled job fires.

import os
import urllib.request
import json
from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult
from praisonai.bots import BotOS, TelegramBot

registry = HookRegistry()
METRICS_ENDPOINT = os.getenv("METRICS_ENDPOINT", "http://localhost:9091/metrics")

@registry.on(HookEvent.SCHEDULE_TRIGGER)
def track_schedule(event_data):
    payload = json.dumps({
        "job_name": event_data.job_name,
        "job_id": event_data.job_id,
        "agent": event_data.agent_name,
        "message_preview": event_data.message[:100] if event_data.message else "",
    }).encode()
    try:
        req = urllib.request.Request(
            METRICS_ENDPOINT,
            data=payload,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        urllib.request.urlopen(req, timeout=2)
    except Exception:
        pass
    return HookResult.allow()

agent = Agent(name="ScheduleBot", instructions="Run scheduled tasks.", hooks=registry)
botos = BotOS(
    bots=[TelegramBot(token=os.getenv("TELEGRAM_BOT_TOKEN"), agent=agent)]
)

import asyncio
asyncio.run(botos.start())

---

Configuration / HookResult

Two events that DO gate: MESSAGE_RECEIVED and MESSAGE_SENDING are real policy control points — HookResult.deny("reason") drops the message and the agent is never invoked; HookResult(decision="allow", modified_input={"content": "..."}) rewrites the content before the agent sees it.

Returning HookResult.deny("reason") from gateway or session lifecycle hooks is best-effort for those specific events (GATEWAY_START, GATEWAY_STOP, SESSION_START, SESSION_END): BotOS emits them but does not gate startup or shutdown on the result.

MESSAGE_RECEIVED is now an inbound gate, symmetric with MESSAGE_SENDING on the outbound side. A hook can drop (deny) or redact (rewrite content via modified_input) an inbound message before agent dispatch. This works consistently across sync and async adapters (Telegram, Slack, Discord, WhatsApp, Email, AgentMail) — no async def required. See Hook Events → Message Events.

For policy enforcement on agent internals (e.g. blocking tool calls or LLM requests), use BEFORE_TOOL or BEFORE_LLM hooks.

Exception — MESSAGE_RECEIVED is now a real gate. Since PR #2589, returning HookResult.deny(...) from MESSAGE_RECEIVED drops the inbound message, and HookResult(decision="allow", modified_input={"content": "…"}) rewrites it in place before the agent sees it. See Inbound Message Gate.

---

Best Practices

Keep lifecycle hooks lightweight

Gateway and session hooks may run inside an async event loop. Avoid blocking I/O or heavy computation — use fire-and-forget coroutines or thread pools for slow operations.

Don't raise from a hook

Emission is wrapped in try/except inside the bot runtime, but unhandled exceptions from your hook function are logged at debug level and swallowed. Return HookResult.allow() even on internal errors to avoid silent failures.

Use agent_name to disambiguate multiple agents

When one BotOS runs multiple bots (each with a different agent), the agent_name field on every event identifies which agent the event belongs to. Key your per-agent metrics on event_data.agent_name.

Key per-user state on session_id, not platform user IDs

Platform user IDs differ between Telegram, Discord, Slack, and WhatsApp. Use session_id from SessionStartInput / SessionEndInput as the stable key for per-user state across platforms.

---

Related

Inbound Message Gate

Drop or redact incoming messages before the agent sees them

Hook Events Reference

Complete event reference with all input types and fields

BotOS

Multi-platform bot orchestrator that emits these lifecycle hooks

Hooks

Hook system concepts: registries, decisions, and matchers

Session Management

How per-user sessions are managed across platforms