BotOS

Run AI agents across Telegram, Discord, Slack, WhatsApp — and custom platforms — with a single orchestrator.

Diagram

python

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.

Diagram

Orchestration

Diagram

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

python

from praisonai_bot.bots import Bot
from praisonaiagents import Agent

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

Multiple platforms

python

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

python

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

Diagram

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`

Usage Examples

Single Agent on Telegram

python

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

python

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

python

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:

yaml

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:

python

from praisonai_bot.bots import BotOS

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

BotOS YAML vs Gateway YAML

BotOS.from_config() reads the lightweight BotOS YAML above. A full gateway.yaml (the file used by praisonai gateway start --config) is a different format and must be run with the gateway instead.

Diagram

Warning

BotOS.from_config() raises ValueError if the file contains any gateway-only section: channels, agents, routing, schedules, rooms, hooks, access_groups, mcp_servers, identity. Run such a file with the gateway instead:

python

import asyncio
from praisonai_bot import WebSocketGateway

asyncio.run(WebSocketGateway().start_with_config("gateway.yaml"))

Keys `BotOS.from_config()` reads	Behaviour for anything else
`name`, `agent`, `platforms`, `health`, `supervision`, `gateway`, `drain_timeout`, `max_concurrent_runs`, `queue_depth`, `overflow_policy`, `max_concurrent_runs_per_scope`, `max_rss_mb`, `reliability`	Gateway-only sections raise `ValueError`; other unknown keys log a `WARNING` and are ignored
Under `gateway:` — `drain_timeout`, `max_concurrent_runs`, `queue_depth`, `overflow_policy`, `max_concurrent_runs_per_scope`, `max_rss_mb`, `reliability`	Other sub-keys (e.g. `gateway.host`, `gateway.port`) log a `WARNING` and are ignored

Smart Defaults

When you create a Bot(), it automatically enhances your agent with useful defaults — no configuration needed:

python

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

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.

Diagram

Supervision is on automatically — no extra arguments needed:

python

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:

python

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:

python

botos = BotOS(agent=agent, platforms=["telegram"], enable_supervision=False)

yaml

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

Diagram

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:

python

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`

Operator Controls

When supervision is enabled, use Python API controls (BotOS has no CLI for these):

python

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

Diagram

Aggregated Health

python

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:

python

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.

Health Checks

Every bot adapter supports probe() and health() for diagnostics:

python

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)

Extending Platforms

Add your own messaging platform in 3 steps:

Create your adapter

python

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

Register the platform

python

from praisonai.bots._registry import register_platform

register_platform("line", LineBot)

Use it like any built-in platform

python

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:
python
```
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

python

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

platform str

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

BotOS class

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

health_monitor HealthMonitorConfig

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

enable_supervision bool

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.

per_user_per_minute int

Cap how many turns one caller may start per minute at ingress. 0 disables (default). Fractional, boolean, or negative values fail at startup. See Gateway Rate Limit for the full behaviour.

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 (BotOS YAML only; raises ValueError on gateway-only sections — see BotOS YAML vs Gateway YAML)

Platform Registry

python

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', 'local', '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().

BotOS

Orchestration

Quick Start

How It Works

Architecture

Token Setup

Usage Examples

Single Agent on Telegram

AgentTeam on Discord

AgentFlow on Multiple Platforms

YAML Configuration

BotOS YAML vs Gateway YAML

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