Skip to main content
from praisonai_bot.bots import BotOS
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="You are a helpful assistant.")
botos = BotOS(agent=agent, platforms=["telegram", "discord"])
botos.run()
The user chats on Telegram or Discord; one shared agent brain answers on every connected platform.

Orchestration

BotOS lets you deploy your AI agent to multiple messaging platforms with just a few lines of code. One agent brain, many channels.

Quick Start

1

Single platform

from praisonai_bot.bots import Bot
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful")
bot = Bot("telegram", agent=agent)
bot.run()
from praisonai.bots import Bot also works when the praisonai wrapper is installed (backward-compatible shim).
2

Multiple platforms

from praisonai_bot.bots import BotOS
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful")
botos = BotOS(agent=agent, platforms=["telegram", "discord"])
botos.run()
3

YAML config

from praisonai_bot.bots import BotOS

botos = BotOS.from_config("botos.yaml")
botos.run()

How It Works

1

Create your Agent

Build an Agent, AgentTeam, or AgentFlow — your AI brain.
2

Wrap in a Bot

Bot("telegram", agent=agent) wraps your agent for a single platform.
3

Orchestrate with BotOS

BotOS starts all your bots concurrently — one command, all platforms.

Architecture

  • Protocol-driven: BotOSProtocol lives in the lightweight core SDK; heavy implementations live in the wrapper
  • Lazy loading: Platform libraries (telegram, discord, etc.) are only imported when start() is called
  • Agent-centric: Every bot is powered by an Agent, AgentTeam, or AgentFlow
  • Extensible: Register custom platforms at runtime with register_platform(), or distribute packaged connectors via the praisonai.channels entry-point group for zero-code auto-registration

Token Setup

Tokens are resolved automatically from environment variables. Set them once, use everywhere.
PlatformEnvironment Variable
TelegramTELEGRAM_BOT_TOKEN
DiscordDISCORD_BOT_TOKEN
SlackSLACK_BOT_TOKEN + SLACK_APP_TOKEN
WhatsAppWHATSAPP_ACCESS_TOKEN + WHATSAPP_PHONE_NUMBER_ID
You can always override with an explicit token= parameter:
bot = Bot("telegram", agent=agent, token="your-token-here")

Usage Examples

Examples below use from praisonai.bots import Bot, BotOS — the wrapper aliases. Replace with from praisonai_bot.bots import Bot, BotOS when using the standalone praisonai-bot package without the wrapper.

Single Agent on Telegram

from praisonai_bot.bots import Bot is the canonical bot-tier import. from praisonai.bots import Bot is a backward-compatible shim when the praisonai wrapper is installed.
from praisonai_bot.bots import Bot
from praisonaiagents import Agent

agent = Agent(
    name="assistant",
    instructions="You are a helpful assistant.",
    llm="gpt-4o-mini",
)

bot = Bot("telegram", agent=agent)
bot.run()  # Blocks until Ctrl+C

AgentTeam on Discord

from praisonai_bot.bots import Bot
from praisonaiagents import Agent, AgentTeam, Task

researcher = Agent(name="researcher", instructions="Research topics thoroughly")
writer = Agent(name="writer", instructions="Write clear, engaging content")

t1 = Task(name="research", description="Research the user's topic", agent=researcher)
t2 = Task(name="write", description="Write a summary based on research", agent=writer)

team = AgentTeam(agents=[researcher, writer], tasks=[t1, t2])

bot = Bot("discord", agent=team)
bot.run()

AgentFlow on Multiple Platforms

from praisonai_bot.bots import BotOS, Bot
from praisonaiagents import Agent, AgentFlow, Task

analyst = Agent(name="analyst", instructions="Analyze data")
reporter = Agent(name="reporter", instructions="Create reports")

t1 = Task(name="analyze", description="Analyze the input", agent=analyst)
t2 = Task(name="report", description="Generate a report", agent=reporter)

flow = AgentFlow(steps=[t1, t2])

botos = BotOS(bots=[
    Bot("telegram", agent=flow),
    Bot("discord", agent=flow),
    Bot("slack", agent=flow, app_token="xapp-..."),
])
botos.run()

YAML Configuration

Create a botos.yaml file:
agent:
  name: assistant
  instructions: You are a helpful assistant.
  llm: gpt-4o-mini
platforms:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}
  discord:
    token: ${DISCORD_BOT_TOKEN}
Then load and run:
from praisonai_bot.bots import BotOS

botos = BotOS.from_config("botos.yaml")
botos.run()
Environment variables in ${VAR_NAME} format are automatically resolved.

Smart Defaults

When you create a Bot(), it automatically enhances your agent with useful defaults — no configuration needed:
from praisonai.bots import Bot
from praisonaiagents import Agent

# Agent gets smart defaults automatically
agent = Agent(name="assistant", instructions="Be helpful")
bot = Bot("telegram", agent=agent)
bot.run()
# Agent now has: search_web, schedule_add, schedule_list, schedule_remove tools
DefaultWhat HappensWhen Applied
Safe toolssearch_web, schedule_add, schedule_list, schedule_removeOnly if agent has no tools
Smart defaults only apply to plain Agent instances. AgentTeam and AgentFlow are left untouched — they already have their own configuration.

Auto-Recovery & Health Monitoring

Every BotOS instance is supervised by default — crashed bots are restarted with backoff, and proactive health checks catch hung connections before users notice. Supervision is on automatically — no extra arguments needed:
from praisonai.bots import BotOS
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful.")
botos = BotOS(agent=agent, platforms=["telegram", "discord"])
botos.run()  # supervised by default
Tune the health monitor when you need tighter sweeps or restart caps:
from praisonai.bots import BotOS
from praisonai.gateway.health_monitor import HealthMonitorConfig
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful.")

botos = BotOS(
    agent=agent,
    platforms=["telegram"],
    health_monitor=HealthMonitorConfig(interval=120, max_restarts_per_hour=20),
)
Opt out when an external supervisor (systemd, Kubernetes) owns restarts:
botos = BotOS(agent=agent, platforms=["telegram"], enable_supervision=False)
# botos.yaml
agent:
  name: assistant
  instructions: Be helpful.
platforms:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}
  discord:
    token: ${DISCORD_BOT_TOKEN}
health:
  interval: 300
  startup_grace: 60
  max_restarts_per_hour: 10
supervision:
  enabled: true   # default; set false to opt out

Multi-Process / HA Deployments

Running more than one BotOS process — for high availability, horizontal scaling, or a brief rolling-restart overlap — used to double-fire every scheduled job, double-posting messages, double-charging tokens, and duplicating side effects. Each due job is now claimed atomically before it runs, so exactly one BotOS process fires it; the others skip silently. This needs no config change as long as the schedule store supports atomic claim — the default FileScheduleStore does, via an advisory-locked claim_due. Verify the guarantee is engaged from the startup log — every ticker prints one line when its schedule loop starts:
BotOS: schedule loop started (30s tick, atomic_claim=True)
atomic_claim=True confirms cross-process at-most-once is active. atomic_claim=False means the current store does not implement claim_due, so the legacy non-atomic path is in use and a job can fire once per process. Start two processes against the same store and watch the log — only one fires each job:
from praisonai.bots import BotOS
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful.")
botos = BotOS(agent=agent, platforms=["telegram"])
botos.run()  # run this same script in two processes; each due job fires once
StoreAt-most-onceHow
FileScheduleStore (default)Advisory-locked claim_due
Custom storeDependsOnly when it implements claim_due on ScheduleStoreProtocol
If a BotOS process crashes mid-run, its 300-second lease expires and another ticker re-claims the job — no permanent job loss. Single-process behaviour is unchanged.
Related: Bot Lifecycle Hooks → SCHEDULE_TRIGGER, Scheduled Run Policy, Proactive Delivery.

Operator Controls

When supervision is enabled, use Python API controls (BotOS has no CLI for these):
botos.pause_bot("telegram")      # stop processing; returns False if supervision disabled
botos.resume_bot("telegram")     # restart from pause
botos.reconnect_bot("telegram")  # full reset + reconnect

Aggregated Health

import asyncio
from praisonai.bots import BotOS
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful.")
botos = BotOS(agent=agent, platforms=["telegram", "discord"])

async def check():
    health_by_platform = await botos.health()
    # {"telegram": HealthResult(...), "discord": HealthResult(...)}
    tg = health_by_platform["telegram"]
    print(tg.ok)             # bool — overall pass/fail
    print(tg.is_running)     # bool — adapter loop alive
    print(tg.uptime_seconds) # float | None
    print(tg.probe.ok)       # bool — platform API reachable
    print(tg.sessions)       # int — active sessions
    print(tg.error)          # str | None — populated when probe fails

    status = botos.get_supervisor_status()
    # supervisor + health-monitor snapshot for /healthz or metrics
    return status

asyncio.run(check())
Plug await botos.health() into a /healthz endpoint or Prometheus exporter — one dict per platform, no custom wiring per adapter.

Unified Per-User Session

Link multiple platforms so one human gets one conversation across all channels:
from praisonai.bots import BotOS
from praisonaiagents import Agent
from praisonaiagents.session import FileIdentityResolver

agent = Agent(name="assistant", instructions="Be helpful.")

resolver = FileIdentityResolver()
resolver.link("telegram", "12345", "alice")
resolver.link("discord", "snowflake-1", "alice")

botos = BotOS(
    agent=agent,
    platforms=["telegram", "discord"],
    identity_resolver=resolver,  # Cross-platform sessions
)
See Cross-Platform Sessions for complete documentation on identity linking and unified sessions.

Health Checks

Every bot adapter supports probe() and health() for diagnostics:
from praisonai.bots import Bot
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful")
bot = Bot("telegram", agent=agent)

import asyncio

# Test connectivity before starting
result = asyncio.run(bot.probe())
print(result.ok)        # True if token is valid and API is reachable
print(result.platform)  # "telegram"
print(result.latency)   # Response time in seconds

# Get detailed health after running
health = asyncio.run(bot.health())
print(health.ok)
print(health.platform)
print(health.is_running)
print(health.uptime_seconds)
print(health.sessions)
print(health.error)
Use probe() in CI/CD pipelines or doctor checks to verify bot tokens before deployment.

Extending Platforms

Add your own messaging platform in 3 steps:
1

Create your adapter

class LineBot:
    def __init__(self, token="", agent=None, **kwargs):
        self.token = token
        self.agent = agent

    async def start(self):
        # Connect to LINE API and start listening
        ...

    async def stop(self):
        # Graceful shutdown
        ...
2

Register the platform

from praisonai.bots._registry import register_platform

register_platform("line", LineBot)
3

Use it like any built-in platform

from praisonai.bots import Bot
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful")
bot = Bot("line", agent=agent, token="your-line-token")
bot.run()
4

Remove a platform (optional)

Useful for tests, hot-reloads, or multi-tenant cleanup:
from praisonai.bots._registry import get_default_bot_registry

removed = get_default_bot_registry().unregister("line")  # bool
unregister() returns True if the platform was registered and removed, False otherwise. It’s safe to call even when the platform isn’t registered.
Your custom adapter class must implement:
MethodRequiredDescription
__init__(**kwargs)YesAccept token, agent, and platform-specific params
async start()YesStart the bot (connect, listen for messages)
async stop()YesGracefully stop the bot
async send_message(channel_id, content)OptionalSend a message programmatically
on_message(handler)OptionalRegister a message handler
on_command(command)OptionalRegister a command handler
is_running (property)OptionalWhether the bot is currently running

Managing Bots

from praisonai.bots import BotOS, Bot
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful")

botos = BotOS()

# Add bots
botos.add_bot(Bot("telegram", agent=agent))
botos.add_bot(Bot("discord", agent=agent))

# List platforms
print(botos.list_bots())  # ["telegram", "discord"]

# Get a specific bot
tg_bot = botos.get_bot("telegram")

# Remove a bot
botos.remove_bot("discord")

# Start all remaining bots
botos.run()

API Reference

platform
str
required
Platform name: "telegram", "discord", "slack", "whatsapp", or any registered custom platform.
agent
Agent | AgentTeam | AgentFlow
The AI agent that powers the bot.
token
str
Explicit token. Falls back to {PLATFORM}_BOT_TOKEN env var.
identity_resolver
IdentityResolverProtocol
Cross-platform identity resolver for unified sessions. StoreBackedIdentityResolver.from_env() (from praisonai.bots) reuses the gateway pairing store so paired users share one session out of the box — see Cross-Platform Sessions.
**kwargs
Platform-specific parameters (e.g., app_token for Slack, mode for WhatsApp).
Methods:
  • bot.run() — Start the bot (sync, blocks)
  • await bot.start() — Start the bot (async)
  • await bot.stop() — Stop the bot
  • await bot.send_message(channel_id, content) — Send a message
  • await bot.probe() — Test channel connectivity (returns ProbeResult)
  • await bot.health() — Get detailed health status (returns HealthResult)
  • bot.register_command(name, handler) — Register a custom chat command
  • bot.list_commands() — List all registered commands
bots
list[Bot]
Pre-built Bot instances to orchestrate.
agent
Agent | AgentTeam | AgentFlow
Shared agent — used with platforms to auto-create Bots.
platforms
list[str]
Platform names — auto-creates a Bot per platform using agent.
identity_resolver
IdentityResolverProtocol
Cross-platform identity resolver for unified sessions. StoreBackedIdentityResolver.from_env() (from praisonai.bots) reuses the gateway pairing store so paired users share one session out of the box — see Cross-Platform Sessions.
health_monitor
HealthMonitorConfig
Proactive health sweep settings (interval, startup_grace, stale_after, stuck_after, max_restarts_per_hour).
enable_supervision
bool
default:"True"
Run each bot through ChannelSupervisor with auto-recovery. Set False when an external supervisor owns restarts.
reliability
str
Reliability preset: "production" (15 s graceful drain + CPU-scaled admission ceiling + bounded fair queue), "default" / None (5 s drain, no admission ceiling), or "off" (immediate teardown, no admission). Explicit fields like drain_timeout and max_concurrent_runs always override the preset. See Gateway Reliability Presets for details.
Methods:
  • botos.run() — Start all bots (sync, blocks)
  • await botos.start() — Start all bots (async)
  • await botos.stop() — Stop all bots
  • botos.add_bot(bot) — Register a bot
  • botos.remove_bot("platform") — Remove a bot
  • botos.list_bots() — List platform names
  • botos.get_bot("platform") — Get a bot by platform
  • await botos.health() — Aggregated Dict[str, HealthResult] across all bots
  • botos.get_supervisor_status() — Supervisor + health-monitor snapshot
  • botos.pause_bot("platform") — Pause a supervised bot
  • botos.resume_bot("platform") — Resume a paused bot
  • botos.reconnect_bot("platform") — Force reconnect and reset error state
  • await botos.deliver(target, text, origin=None) — Proactive outbound delivery (see Proactive Delivery)
  • botos.configure_channels(config) — Set home channels and aliases
  • botos.delivery_router — Direct access to DeliveryRouter and ChannelDirectory
  • BotOS.from_config("path.yaml") — Load from YAML
from praisonai.bots._registry import (
    register_platform,         # Add a custom platform (lower-cased name)
    list_platforms,            # List all registered platform names
    resolve_adapter,           # Get adapter class by name
    get_platform_registry,     # Get the full {name: class} dict
    get_default_bot_registry,  # Get the underlying BotPlatformRegistry (advanced)
)

# Remove a previously registered platform (useful for tests, hot-reloads, multi-tenant cleanup)
get_default_bot_registry().unregister("line")  # returns True if it was registered
Error you’ll see for unknown platforms:
ValueError: Unknown praisonai.bots plugin: 'nonexistent_xyz'. Available: ['agentmail', 'discord', 'email', 'linear', 'slack', 'telegram', 'whatsapp']
The error string changed in a recent refactor. If you’re migrating from an older PraisonAI, update any try/except blocks or test assertions that matched the old "Unknown platform" text.

Best Practices

Use TELEGRAM_BOT_TOKEN, DISCORD_BOT_TOKEN, and related env vars so secrets never live in source code.
Default BotOS supervision restarts crashed bots with backoff — disable only when systemd or Kubernetes owns restarts.
Call await bot.probe() in CI to catch invalid tokens before users hit your bot.

Messaging Bots

Platform-specific bot setup and capabilities.

Bot Gateway

Gateway routing, pairing, and channel supervision.

Cross-Platform Sessions

Unified sessions across Telegram, Discord, and more.

Proactive Delivery

Push outbound messages with botos.deliver().