Skip to main content
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

EventWhenInput TypeKey Fields
GATEWAY_STARTBotOS.start()GatewayStartInputplatforms, bot_count
GATEWAY_STOPBotOS.stop()GatewayStopInputplatforms, bot_count, reason
SESSION_STARTFirst message per user (once per session lifetime)SessionStartInputsession_id, agent_name, source, session_name
SESSION_END/new reset, policy auto-reset, stale reap, reset_allSessionEndInputsession_id, agent_name, reason
SCHEDULE_TRIGGERScheduled 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).ScheduleTriggerInputjob_name, job_id, message
MESSAGE_RECEIVEDIncoming message from platform, before agent dispatch — inbound gate: deny drops the message, modified_input["content"] rewrites itMessageReceivedInputplatform, content, sender_id
MESSAGE_SENDINGBefore bot sends a reply — outbound gate: deny cancels sending, modified_input["content"] rewrites itMessageSendingInputplatform, content, channel_id
MESSAGE_SENTAfter bot successfully sends a replyMessageSentInputplatform, 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

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.
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.
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.
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.

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