BotOS - PraisonAI

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

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

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()

YAML config

from praisonai_bot.bots import BotOS

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

How It Works

Create your Agent

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

Wrap in a Bot

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

Orchestrate with BotOS

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

Architecture

Design principles

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.

Platform	Environment Variable
Telegram	`TELEGRAM_BOT_TOKEN`
Discord	`DISCORD_BOT_TOKEN`
Slack	`SLACK_BOT_TOKEN` + `SLACK_APP_TOKEN`
WhatsApp	`WHATSAPP_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

Default	What Happens	When Applied
Safe tools	`search_web`, `schedule_add`, `schedule_list`, `schedule_remove`	Only 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)

Bot(..., enable_supervision=...) and BotOS(..., enable_supervision=...) are two separate kwargs on two separate classes — set the one that matches the entry point you’re actually calling. See Auto-reconnect on long-running-bots for the single-Bot path.

# 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 — both bundled stores support atomic claim via an advisory-locked claim_due: the default ConfigYamlScheduleStore and FileScheduleStore. 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

Store	At-most-once	How
`ConfigYamlScheduleStore` (default)	✅	Advisory-locked `claim_due`
`FileScheduleStore`	✅	Advisory-locked `claim_due`
Custom store	Depends	Only 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.

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
)

When the resolver unifies multiple adapters onto one user, BotOS also serialises turns on the resolved session id — a user typing on Telegram while their Slack app fires in the same second cannot scramble the shared transcript. See Concurrent Turns Across Adapters.

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:

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

from praisonai.bots._registry import register_platform

register_platform("line", LineBot)

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()

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.

Adapter contract

Your custom adapter class must implement:

Method	Required	Description
`__init__(**kwargs)`	Yes	Accept `token`, `agent`, and platform-specific params
`async start()`	Yes	Start the bot (connect, listen for messages)
`async stop()`	Yes	Gracefully stop the bot
`async send_message(channel_id, content)`	Optional	Send a message programmatically
`on_message(handler)`	Optional	Register a message handler
`on_command(command)`	Optional	Register a command handler
`is_running` (property)	Optional	Whether 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

Bot class

str

required

Platform name: "telegram", "discord", "slack", "whatsapp", or any registered custom platform.

Agent | AgentTeam | AgentFlow

The AI agent that powers the bot.

str

Explicit token. Falls back to {PLATFORM}_BOT_TOKEN env var.

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.

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

BotOS class

list[Bot]

Pre-built Bot instances to orchestrate.

Agent | AgentTeam | AgentFlow

Shared agent — used with platforms to auto-create Bots.

list[str]

Platform names — auto-creates a Bot per platform using agent.

IdentityResolverProtocol

HealthMonitorConfig

Proactive health sweep settings (interval, startup_grace, stale_after, stuck_after, max_restarts_per_hour).

bool

default:"True"

Run each bot through ChannelSupervisor with auto-recovery. Set False when an external supervisor owns restarts.

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

Platform Registry

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

Set tokens via environment variables

Use TELEGRAM_BOT_TOKEN, DISCORD_BOT_TOKEN, and related env vars so secrets never live in source code.

Keep supervision enabled in production

Default BotOS supervision restarts crashed bots with backoff — disable only when systemd or Kubernetes owns restarts.

Link identities for cross-channel users

Use FileIdentityResolver or StoreBackedIdentityResolver so one person gets one session across Telegram, Discord, and Slack.

Probe tokens before deploy

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

​Orchestration

​Quick Start

​How It Works

​Architecture

​Token Setup

​Usage Examples

​Single Agent on Telegram

​AgentTeam on Discord

​AgentFlow on Multiple Platforms

​YAML Configuration

​Smart Defaults

​Auto-Recovery & Health Monitoring

​Multi-Process / HA Deployments

​Operator Controls

​Aggregated Health

​Unified Per-User Session

​Health Checks

​Extending Platforms

​Managing Bots

​API Reference

​Best Practices

​Related

Messaging Bots

Bot Gateway

Cross-Platform Sessions

Proactive Delivery

Orchestration

Quick Start

How It Works

Architecture

Token Setup

Usage Examples

Single Agent on Telegram

AgentTeam on Discord

AgentFlow on Multiple Platforms

YAML Configuration

Smart Defaults

Auto-Recovery & Health Monitoring

Multi-Process / HA Deployments

Operator Controls

Aggregated Health

Unified Per-User Session

Health Checks

Extending Platforms

Managing Bots

API Reference

Best Practices

Related