Proactive Delivery

Send messages from your bot to any channel — by origin, platform, channel ID, or friendly alias.

Push outbound messages from an agent-backed bot — reply to the requesting chat, a home channel, or a named alias.

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

agent = Agent(name="ops", instructions="Alert humans about incidents.")
botos = BotOS(bots=[Bot("telegram", agent=agent)])

botos.configure_channels({
    "telegram": {"home_channel": "123456", "aliases": {"ops-alerts": "123456"}},
})

async def notify():
    src = SessionSource(platform="telegram", channel_id="123456")
    await botos.deliver("origin", "Build finished!", origin=src)

asyncio.run(notify())

The user triggers an alert; BotOS delivers the agent’s message to the target channel.

Quick Start

1. Simple Usage

   Reply to where the request came from:

   import asyncio
   from praisonai.bots import BotOS, Bot, SessionSource
   from praisonaiagents import Agent
   
   agent = Agent(name="ops", instructions="Send status updates.")
   botos = BotOS(bots=[Bot("telegram", agent=agent)])
   
   async def notify():
       src = SessionSource(platform="telegram", channel_id="123456")
       await botos.deliver("origin", "Build finished!", origin=src)
   
   asyncio.run(notify())

2. With Configuration

   Set home channels and deliver by alias:

   botos.configure_channels({
       "telegram": {"home_channel": "123456", "aliases": {"ops-alerts": "123456"}},
   })
   
   await botos.deliver("telegram", "Nightly digest ready")
   await botos.deliver("ops-alerts", "Disk usage at 90%")

How It Works

Resolution order: origin → platform:id → bare platform → alias. Platform names win over aliases.

Home and observed channels persist to ~/.praisonai/state/channel_directory.json.

Delivery from Scheduled Runs

For a scheduled agent that isn’t hosted inside BotOS, use AgentScheduler(deliver=…) — same DeliveryRouter, no gateway required.

from praisonaiagents import Agent
from praisonai.scheduler import AgentScheduler

agent = Agent(name="Brief", instructions="Summarise the morning news")
AgentScheduler(agent, task="Morning brief", deliver="telegram:123456").start("hourly")

Configuration Options

YAML

channels:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}
    home_channel: "123456"
    aliases:
      ops-alerts: "123456"

Alias overlay

Pre-name channels in ~/.praisonai/state/channel_aliases.json:

{
  "engineering": { "platform": "discord", "channel_id": "555" },
  "ops": "slack:C111"
}

Call botos.delivery_router.refresh_directory() periodically so adapters enumerate channels the bot has not yet heard from.

Rate Limiting

Scheduled and background sends share the reply-path rate limiter automatically — no config change required.

Scheduled sends also fire at most once per due window across multiple BotOS processes — see BotOS → Multi-Process / HA Deployments.

When the gateway is the process performing the proactive/scheduled send (via WebSocketGateway._deliver_scheduled_result — e.g. any AgentScheduler job whose deliver= resolves inside a running gateway), the send is enqueued into a durable SQLite outbox at ~/.praisonai/state/gateway_outbox.sqlite before it reaches the router. The UNIQUE idempotency key survives a process crash-and-restart, so the same result is never posted twice. See Durable Delivery → Proactive Path for the full contract and comparison against direct BotOS.deliver(...).

# Scheduled daily digest — proactive send to a named channel
await botos.deliver(
    "ops-alerts",
    "Nightly digest ready",
)

# Targeted proactive send with origin context
await botos.deliver(
    "ops-alerts",
    "Build finished",
    origin=source,
)

Best Practices

Related