Out-of-Process Relay Transport

Run a thin connector that owns the platform socket and relay events to a headless gateway over an authenticated transport

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:

RelayTransport Protocol

Implement these methods to create a custom transport:

When to Use a Relay

graph TB
    Q1{Is the gateway\nin a private network\nor serverless?}
    Q1 -->|Yes| Q2{Do you need\none gateway to front\nmultiple platforms?}
    Q1 -->|No| INPROC[✅ In-process adapter\n(default)]
    Q2 -->|Yes| RELAY[✅ Relay transport]
    Q2 -->|No| Q3{Will the gateway\nscale to zero?}
    Q3 -->|Yes| RELAY
    Q3 -->|No| INPROC

    classDef decision fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef inproc fill:#10B981,stroke:#7C90A0,color:#fff
    classDef relay fill:#F59E0B,stroke:#7C90A0,color:#fff

    class Q1,Q2,Q3 decision
    class INPROC inproc
    class RELAY 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

Related