Bot Gateway

Run multiple bots (Telegram, Discord, Slack) from a single gateway server with multi-agent routing.

from praisonaiagents import Agent

agent = Agent(name="gateway-bot", instructions="Handle messages from Telegram, Slack, and Discord.")
agent.start("Set up the bot gateway for all chat platforms.")

The user messages on Telegram, Slack, or Discord; the gateway routes each channel to the right agent.

Run all your bots from one command. The Gateway Server manages multiple bot connections and routes messages to the right AI agent.

Parity with Bot(): praisonai gateway start now applies the same smart defaults as praisonai bot start, resolving the previous “zero tools in daemon mode” issue. Both entry points produce identical behavior with safe tools auto-injected and auto-approval enabled by default.

Each channel bot gets an isolated agent via Agent.clone_for_channel() — fresh locks, fresh interrupt controller, and no shared handoffs state — so configuring tools or memory on one channel never leaks into another.

Quick Start

1. Install PraisonAI

   pip install praisonai

2. Create gateway.yaml

   Create a gateway.yaml file in your project (Gateway YAML files are read as UTF-8 — non-ASCII characters work on all platforms including Windows):

   gateway:
     host: "127.0.0.1"
     port: 8765
   
   agents:
     personal:
       instructions: "You are a helpful personal assistant"
       model: gpt-4o-mini
       tools:
         - internet_search
         - get_current_time
       reflection: true
     support:
       instructions: "You are a customer support agent"
       model: gpt-4o
       role: "Customer Support Specialist"
       goal: "Resolve customer issues quickly and accurately"
       backstory: "Expert in troubleshooting and customer care"
       tools:
         - internet_search
       tool_choice: auto
       allow_delegation: false
   
   channels:
     telegram:
       token: ${TELEGRAM_BOT_TOKEN}
       allowed_users: []                        # or a list of allowed user IDs
       unknown_user_policy: deny                # deny (default, secure) | pair | allow
       owner_user_id: ${TELEGRAM_OWNER_ID}      # required if unknown_user_policy: pair
       streaming:
         mode: draft       # off | draft | progress — see /features/bot-streaming-replies
       routes:
         dm: personal
         group: support
         default: personal
     discord:
       token: ${DISCORD_BOT_TOKEN}
       allowed_users: []
       unknown_user_policy: deny
       owner_user_id: ${DISCORD_OWNER_ID}
       routes:
         default: personal

3. Set Environment Variables

   export TELEGRAM_BOT_TOKEN=your_telegram_token
   export DISCORD_BOT_TOKEN=your_discord_token
   export OPENAI_API_KEY="${OPENAI_API_KEY:?Set OPENAI_API_KEY in your shell}"

4. Start the Gateway

   praisonai gateway start --config gateway.yaml

   All bots start together. Messages are routed to the correct agent automatically.

Supported Channels

Configuration Reference

Gateway Section

Agents Section

Each agent is defined by a unique ID and its configuration:

agents:
  my_agent_id:
    instructions: "Your system prompt here"
    model: gpt-4o-mini
    memory: true
    tools:
      - internet_search
      - get_current_time
    reflection: true
    role: "Research Analyst"
    goal: "Find accurate information"
    backstory: "Expert researcher"
    tool_choice: auto
    allow_delegation: false

Gateway memory identity

Every gateway agent gets a stable memory identity derived from its configured agent id, so facts learned through one chat survive restarts, hot-reloads, and channel clones.

The same id is carried onto every per-channel clone, so Telegram, Slack, and other channels served by one agent share its memory.

What gateway-<agent_id> looks like

Memory directories only allow A-Z, a-z, 0-9, _ and -, so the gateway replaces any other character with _. When a replacement happens, an 8-character hash of the original id is appended so ids stay distinct: support_bot → gateway-support_bot, but support.bot → gateway-support_bot-<hash> — they live in different memory directories.

from praisonaiagents import Agent

# Plain Agent (outside the gateway) — still gets a random id, by design.
agent = Agent(name="assistant", instructions="Help the user.", memory=True)
# → memory user_id = agent-<random hex>

# gateway.yaml — gateway agents get a stable id for free.
agents:
  support-bot:
    instructions: "Help the customer."
    model: gpt-4o-mini
    memory: true
    # memory user_id = gateway-support-bot (stable across restarts)

  billing-bot:
    instructions: "Handle invoices."
    model: gpt-4o-mini
    memory:
      user_id: billing-team        # explicit override wins

Channels Section

Each channel maps to a bot platform:

channels:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}    # Environment variable
    unknown_user_policy: deny        # deny (default) | pair | allow
    owner_user_id: ${TELEGRAM_OWNER_ID}  # required if unknown_user_policy: pair
    routes:
      dm: personal                   # DMs → personal agent
      group: support                 # Groups → support agent
      default: personal              # Fallback agent

Multi-Agent Rooms

Declare a roster of agents under a top-level rooms: block and bind it to a channel with a room:<name> route target. The roster takes bounded, @-addressed turns on each new human message, and the transcript is persisted via the configured session store.

rooms:
  standup:
    agents: [researcher, critic, writer]
    max_rounds: 3
    max_messages: 20

channels:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}
    routes:
      group: room:standup

Channel Security

Each channel enforces the same access-control pipeline as standalone bots. Every key below is read per-channel from gateway.yaml — including unknown_user_policy and owner_user_id, which the gateway now wires through at start and on hot-reload (PR #2856).

Reusable sender sets. If several channels share the same user list, define it once under top-level access_groups: and reference it as group:<name> from allowed_users, allowlist or blocklist. See Access Groups.

# Same effect from Python — for consumers building custom adapters
from praisonaiagents.bots import BotConfig

config = BotConfig(
    token="...",
    allowed_users=["telegram:123456"],
    unknown_user_policy="pair",     # deny | pair | allow
    owner_user_id="telegram:987654",
)

# is_explicitly_allowed skips the pairing flow for whitelisted users
config.is_explicitly_allowed("telegram:123456")  # True
config.is_explicitly_allowed("telegram:99999")   # False → policy decides

The same configuration comes straight from gateway.yaml:

Opt a channel into bot-to-bot traffic with the runaway-loop guard active:

channels:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}
    allow_bots: true
    bot_loop_protection:
      max_events_per_window: 20
      window_seconds: 60
      cooldown_seconds: 60

observe: passive group context

observe records unmentioned group messages as passive context so the bot can answer with full conversational awareness when it’s finally mentioned.

With mention_only (the default), messages that don’t @mention the bot are dropped and leave no session record. With observe, those messages are appended to the session transcript as ordinary user turns — the bot still does not answer them, but the next time it is addressed (e.g. @bot summarise what we just decided) the preceding conversation is in its history.

Pair observe with session_scope: per_chat so passive messages route to the shared per-chat key that a later addressed turn reads from. Under the default per_user scope, passive turns are retained only in that sender’s own history — they do not leak into any single user’s per_user history when per_chat is used.

channels:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}
    group_policy: observe
    session_scope: per_chat

When to use it: a busy group where users want to @bot and have it see prior discussion; the bot stays quiet until addressed.

When not to use it: noise-sensitive or high-PII channels — every unmentioned message is stored, increasing the volume of retained conversation text.

See Session Scopes and Intentional Silence.

Shell execution

Set allow_shell: true on any channel to turn its bot into a shell-capable agent, with approval routed per platform. Policy is per-channel — one channel can auto-approve while another requires owner sign-off. On channels with routing rules, routed agents inherit execute_command too — the shell setup is re-applied to each routed agent, cloned and cached per (channel, agent). See Bot Shell Execution.

channels:
  telegram:
    token: "${TELEGRAM_BOT_TOKEN}"
    # allow_shell: true
    # auto_approve_shell: false
    # approval_channel: "123456789"

channels:
  telegram_support:
    token: ${TELEGRAM_BOT_TOKEN}
    allowed_users:                # User-ID allowlist (empty list → unknown_user_policy decides)
      - "123456789"
      - "987654321"
    allowed_channels:             # Group/channel allowlist (empty = anywhere)
      - "-1001234567890"
    unknown_user_policy: deny     # deny (default, secure) | pair | allow
    owner_user_id: ${TELEGRAM_OWNER_ID}  # required if unknown_user_policy: pair
    group_policy: "mention_only"  # mention_only | command_only | respond_all | observe
    auto_approve_tools: true      # Auto-approve safe tools (default true)
    routes:
      dm: personal
      group: support
      default: personal

Startup warnings

An empty allowed_users triggers a policy-aware warning at gateway start and on hot-reload. Grep these exact strings in your logs:

Channel 'telegram' has no allowed_users AND unknown_user_policy='deny' — inbound DMs from unknown users are SILENTLY DROPPED. Set unknown_user_policy: pair (with owner_user_id) or allow to change this.

Channel 'telegram' has no allowed_users AND unknown_user_policy='pair' — unknown users go through the owner-approval pairing flow. Ensure owner_user_id is set.

Channel 'telegram' has no allowed_users AND unknown_user_policy='allow' — bot accepts messages from everyone. This is intentional only in trusted networks.

If you see the deny warning and expected messages to flow through, your config is secure-by-default and DMs are intentionally dropped. Add IDs to allowed_users, or set unknown_user_policy: pair (with owner_user_id) or allow.

Interactive presentations (optional)

GatewayMessage carries an optional presentation: MessagePresentation field. When set, channel adapters that implement SupportsPresentation render it as a native widget (Telegram inline keyboard, Slack Block Kit, Discord components). Channels that do not implement the protocol fall back to the message’s plain-text content. See Interactive Bot Messages for the full presentation model.

Session Learning (default-on)

Every gateway/bot session enables the coordinated learn=True posture by default, so agents accrue persona and preferences from plain chat across sessions.

Opt a channel out with learn: false (or session_learning: false) in the channel YAML — the gateway forwards it through config.metadata:

channels:
  telegram:
    token: "${TELEGRAM_TOKEN}"
    learn: false

See Bot Session Learning for the full behaviour, coordinated-posture table, and opt-out surfaces.

Hot-Reload

Editing gateway.yaml while the gateway runs triggers a diff-driven reload — only affected agents or channels restart. The WebSocket server stays up.

See Gateway Hot-Reload for the full restart-scope table.

CLI Commands

praisonai gateway start --config gateway.yaml

praisonai gateway start --config gateway.yaml --host 0.0.0.0 --port 9000

praisonai gateway status

praisonai gateway stop

Python Usage

from praisonai.gateway import WebSocketGateway

# Load gateway config, agents, and channels from YAML
gateway = WebSocketGateway.from_config_file("gateway.yaml")

import asyncio
asyncio.run(gateway.start())

import asyncio
from praisonai.gateway import WebSocketGateway
from praisonaiagents import Agent
from praisonaiagents import GatewayConfig

# Create agents
personal = Agent(name="personal", instructions="You are a helpful assistant")
support = Agent(name="support", instructions="You are a support agent")

# Create gateway
config = GatewayConfig(host="127.0.0.1", port=8765)
gateway = WebSocketGateway(config=config)

# Register agents (use overwrite=False to prevent accidental replacement)
gateway.register_agent(personal, agent_id="personal")
gateway.register_agent(support, agent_id="support", overwrite=False)

# Start gateway
asyncio.run(gateway.start())

# After starting the gateway, inspect connected channel bots
gateway.list_channel_bots()      # ['telegram', 'discord']
gateway.get_channel_bot("telegram")  # Returns bot instance
gateway.has_channel_bot("slack")     # False

# Inspect registered agents
gateway.list_agents()            # ['personal', 'support']
gateway.get_agent("personal")    # Returns Agent instance

Custom agent entry-point contract

When a custom agent is registered with the gateway, its arun / achat / chat method receives a per-turn interrupt in one of two ways depending on its signature.

The gateway prefers an async arun / achat and falls back to a sync chat.

class MyAgent:
    def chat(self, content, cancel_token=None):
        # cancel_token is this turn's interrupt controller
        return f"echo: {content}"

gateway.register_agent(MyAgent(), agent_id="mine")

class MyAgent:
    async def achat(self, content, **kwargs):
        cancel_token = kwargs.get("cancel_token")  # this turn's interrupt controller
        return f"echo: {content}"

gateway.register_agent(MyAgent(), agent_id="mine")

class MyAgent:
    interrupt_controller = None  # must exist — the gateway only writes the turn's controller to an existing attribute

    def chat(self, content):
        return f"echo: {content}"

gateway.register_agent(MyAgent(), agent_id="mine")

curl http://127.0.0.1:8765/health

{
  "status": "healthy",
  "uptime": 120.5,
  "agents": 2,
  "sessions": 3,
  "clients": 1,
  "channels": {
    "telegram": {
      "platform": "telegram",
      "running": true
    },
    "discord": {
      "platform": "discord", 
      "running": false
    }
  },
  "push": {
    "enabled": true,
    "push_channels": 2,
    "online_clients": 5,
    "redis_connected": true
  }
}

praisonai bot telegram --token $TELEGRAM_BOT_TOKEN

BotOS Options

When creating a BotOS instance in Python, the following options control gateway-level behaviour:

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

agent = Agent(name="assistant", instructions="Help users")

botos = BotOS(
    bots=[Bot("telegram", agent=agent)],
    idle_policy=ScaleToZeroPolicy(
        idle_timeout_minutes=5,
        wake_url="https://my-bot.fly.dev/_wake",
    ),
)
botos.run()

Observability

Code-Skew Guard

After running git pull or pip install -U, the gateway’s /model command detects that the installed code version has changed and blocks the model switch with a “restart required” message. This prevents the model from changing when the running code is out of sync with the installed libraries.

To opt out (useful in development):

session_manager.code_skew_guard = False

Helpers for custom tooling:

from praisonaiagents.gateway import detect_code_skew, read_code_fingerprint

fingerprint = read_code_fingerprint()
if detect_code_skew(fingerprint):
    print("Code has changed since gateway started — restart recommended")

Best Practices

Related