Praison AI

Bot Lifecycle Hooks

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

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.

python
from praisonaiagents import Agent

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

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

Diagram

🚀 GATEWAY_START

📥 SESSION_START

💬 MESSAGE_RECEIVED

🤖 BEFORE_AGENT

✅ AFTER_AGENT

📤 MESSAGE_SENT

📴 SESSION_END

🛑 GATEWAY_STOP

⏰ SCHEDULE_TRIGGER

Quick Start

  1. Register a SESSION_START hook
    python
    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
    python
    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

Diagram
AgentHookRunnerBotSessionManagerBotPlatformUserAgentHookRunnerBotSessionManagerBotPlatformUserSend messageIncoming messagechat(message)SESSION_START (once per session lifetime)MESSAGE_RECEIVEDagent.chat()BEFORE_AGENTAFTER_AGENTMESSAGE_SENT/new commandreset()SESSION_END (reason=clear)

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_ADDA schedule is persisted by the store (add) — from the schedule_add tool, the praisonai schedule CLI, or the config.yaml reconcilerScheduleAddInputjob_name, job_id, schedule, message, agent_id, principal, enabled
SCHEDULE_REMOVEA schedule is deleted by the store (remove/remove_by_name) or auto-cleaned after a one-shot delete_after_run firesScheduleRemoveInputjob_name, job_id, schedule, principal
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
MESSAGE_UNDELIVEREDA reply failed permanently (target confirmed dead or retries exhausted) — opt-in via gateway.notify_on_undeliveredMessageUndeliveredInputplatform, content, channel_id, error, notice_delivered

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.

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

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

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

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

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