Skip to main content
A relay transport lets a lightweight connector process own the Telegram/Discord/WhatsApp socket while a separate headless gateway handles all agent logic — enabling NAT-friendly hosting, one gateway fronting many connectors, and lossless scale-to-zero. 
from praisonaiagents import Agent

agent = Agent(
    name="Support",
    instructions="Answer messages routed from the connector gateway.",
)

agent.start("Summarise this inbound chat thread")
The user messages on a chat platform; the connector relays events to a private gateway that runs the agent.

​
Quick Start

1

In-Process (Default — No Change Needed)

Without a relay transport, the bot adapter runs in the same process as the gateway. This is the default and requires no extra config:
from praisonaiagents import Agent
from praisonai.bots import TelegramBot

agent = Agent(name="Chat Agent", instructions="Help users.")
bot = TelegramBot(token="YOUR_TOKEN", agent=agent)
await bot.start()
2

Out-of-Process via Relay Transport

Pass a transport= to the Bot constructor to use a remote connector. The transport= parameter is optional and backward-compatible:
from praisonaiagents import Agent
from praisonai.bots import Bot

agent = Agent(name="assistant", instructions="Help users")
bot = Bot(
    "telegram",
    agent=agent,
    transport=WebSocketRelayTransport(url="wss://gw.internal/relay"),
)
bot.run()  # capabilities negotiated at handshake; events relayed in
3

Implement a custom RelayTransport

from praisonaiagents.gateway import RelayTransport, CapabilityDescriptor

class MyRelayTransport:
    async def connect(self) -> CapabilityDescriptor:
        # Connect to relay; return channel capabilities
        return CapabilityDescriptor(
            max_message_length=4096,
            length_unit="chars",
            supports_edit=True,
            supports_draft_streaming=True,
            markdown_dialect="telegram",
        )

    def set_inbound_handler(self, handler):
        self._handler = handler

    async def send_outbound(self, message) -> None:
        # Send reply back through the relay
        await self._relay_connection.send(message)

    async def go_dormant(self) -> None:
        # Signal scale-to-zero; buffer inbound until resumed
        await self._relay_connection.send({"type": "dormant"})

    async def disconnect(self) -> None:
        await self._relay_connection.close()
4

Enable scale-to-zero

bot = Bot("telegram", agent=agent, transport=MyRelayTransport())

# Signal dormant state (connector buffers, gateway can shut down)
await bot.go_dormant()

# On wake-up, connector replays buffered events
bot.run()

from praisonai.bots import WebSocketRelayTransport

agent = Agent(name="Headless Agent", instructions="Help users.")

transport = WebSocketRelayTransport(
    url="wss://gw.internal/relay",
    token="YOUR_RELAY_SECRET",
)

bot = Bot("telegram", agent=agent, transport=transport)
await bot.start()
5

CLI Gateway with Relay

praisonai gateway relay \
  --platform telegram \
  --to wss://gw.internal/relay \
  --token YOUR_RELAY_SECRET

​
How It Works

At handshake time the connector attests a CapabilityDescriptor — the gateway uses this to adapt streaming, markdown, and message-length behaviour to the actual platform being fronted, even though it never touches the platform socket directly.

​
CapabilityDescriptor Fields

The connector attests these capabilities at handshake time:
FieldTypeDefaultDescription
max_message_lengthintrequiredMaximum outbound message length the platform accepts
length_unitstr"chars"How length is measured: "chars" (Unicode code points) or "utf16" (UTF-16 code units)
supports_editboolFalsePlatform supports editing a sent message (enables draft-streaming)
supports_draft_streamingboolFalseConnector can stream partial drafts incrementally
markdown_dialectstr"none"Markdown flavour: "none", "markdown", "markdownv2", "html"

​
RelayTransport Protocol

Implement these methods to create a custom transport:
MethodDescription
connect() -> CapabilityDescriptorEstablish relay and complete handshake; returns connector capabilities
set_inbound_handler(handler)Register coroutine called for each inbound event
send_outbound(target, message) -> DeliveryResultRelay an outbound message to a target via the connector
go_dormant()Pause inbound dispatch without dropping the connection (scale-to-zero)
disconnect()Tear down the relay connection

​
When to Use a Relay

​
Scale-to-Zero with go_dormant()

When the gateway scales to zero, call go_dormant() to pause inbound dispatch: 
transport = WebSocketRelayTransport(url="wss://gw.internal/relay")
caps = await transport.connect()

await transport.go_dormant()
The connector keeps the platform socket open and buffers inbound events while the gateway is dormant. When the gateway wakes, it drains the backlog losslessly — no messages are dropped during the idle period.

​
Best Practices

The relay transport authenticates with a token passed at connect() time. Use a long, random secret and rotate it with your other infrastructure credentials.
disconnect() tears down the connection and the connector stops buffering. go_dormant() keeps the connector alive and buffering, so events that arrive while the gateway is sleeping are drained on wake.
Always use the CapabilityDescriptor returned by connect() to configure your delivery layer. Never hard-code platform limits — the connector is the authority on what the actual platform supports.
Omit transport= entirely to keep the existing in-process behaviour. No existing code needs to change; just add the parameter when you need an out-of-process connector.

Gateway Overview

How the gateway connects agents to channels

Gateway Scale-to-Zero

Idle dormancy and wake-up behaviour

Durable Outbound Delivery

Retry and DLQ for all channels

Multi-Channel Bots

One gateway, many platforms