Gateway & Control Plane

Unified gateway and control plane architecture for PraisonAI integrations

Agents can connect through a unified gateway providing single-entry access to multi-agent coordination, tools, and streaming events.

Quick Start

1. Simple Agent Gateway

   Deploy an agent through the unified gateway:

   from praisonaiagents import Agent
   
   # Gateway enables unified access
   agent = Agent(
       name="Gateway Agent",
       instructions="Connect through unified gateway",
       gateway=True
   )
   
   agent.start("Process through gateway")

2. Multi-Service Gateway

   Configure agents with full gateway capabilities:

   from praisonaiagents import Agent, GatewayConfig
   
   agent = Agent(
       name="Gateway Agent",
       instructions="Multi-service coordination",
       gateway=GatewayConfig(
           unified=True,
           services=["agents", "mcp", "a2a", "a2u"]
       )
   )

How It Works

Agents connect to services through gateway routing:

Hot Reload

Edit gateway.yaml while the gateway is running — file watch (with optional watchdog) and SIGHUP share the same diff-driven reload path, with drain-coordinated channel restarts. Three timing/logging keys (gateway.logging.level, gateway.drain_timeout, gateway.reload_drain_timeout) are hot-applied in place — no channel restart at all — see Hot-Apply (no restart). Rotating gateway.auth_token in the same reload also revokes any live sessions on the old secret — see Gateway Credential Rotation. See Gateway Config Reload.

For proactive/scheduled outbound delivery routing — including friendly alias targets like "family" or "ops" — see Friendly Aliases for Scheduled Delivery.

To declare recurring agent → channel deliveries directly in gateway.yaml (with matching praisonai gateway schedule add | list | remove CLI), see Gateway Schedules.

Voice notes

Voice messages sent to your gateway bot are transcribed automatically on Telegram, Slack, WhatsApp, Discord, and Signal — no code changes required. Configure the stt: block under each channel to force a language or opt out. See Voice Notes (Speech-to-Text).

The bot can also speak back. Configure an opt-in voice: block under any channel to have the agent’s reply delivered as a native voice note — the outbound mirror of stt:. See Voice Reply (Text-to-Speech).

Authentication

Authentication posture changes automatically based on the bind interface.

Rotating the shared secret

Change gateway.auth_token in the running gateway.yaml and hot-reload — every already-connected session that authenticated under the old secret is force-closed with WebSocket close code 4001 and reason credentials_rotated. New connections use the new secret. Defaults on; opt out with gateway.revoke_on_secret_rotation: false. See Gateway Credential Rotation for the full behaviour matrix.

The auto-generated per-install HMAC secret (~/.praisonai/gateway/.gateway_secret, mode 0600) is stored and permission-remediated separately from gateway.auth_token. Both the pairing codes and the inline Approve/Deny callback buttons sign with this same secret, so both survive gateway restarts by default — no operator action required. Set PRAISONAI_CALLBACK_SECRET to a shared value only when running the gateway on multiple hosts/replicas. See Bot Pairing → Secret Management for pairing codes and Interactive Bot Actions for the callback-button flow.

Rejecting weak / placeholder secrets

A gateway bound to an external interface refuses to start if auth_token is a well-known placeholder (change-me, your-token-here, the literal string $(openssl rand -hex 16), etc.). Loopback binds still start but log a warning.

Pre-auth edge protections

Two guards protect internet-exposed gateway deployments by default: a per-IP connection budget (preauth_max_connections_per_ip=32) that caps unauthenticated WebSocket slots before auth runs, and a per-connection flood guard (max_unauthorized_frames=10) that closes connections sending too many unauthorized frames. Loopback clients are always exempt. Both are disabled by setting the option to 0.

• Gateway commands are gated by content provenance — see Control-Trust Provenance.

Trusted proxies

Behind a reverse proxy or tunnel, set gateway.trusted_proxies so per-IP rate-limits and the operator id key on the real client instead of the proxy peer. The resolver walks X-Forwarded-For only across declared trusted hops and fails closed for anything unattributable. Empty (default) keeps the raw-peer behaviour.

Handshake and Version Negotiation

New WebSocket clients should send a hello frame to negotiate protocol version, capabilities, and policy limits in one round trip. The legacy join → joined flow remains supported for existing clients.

Handling connect errors

A rejected handshake carries a structured code and next_step, so a client can tell a terminal failure (revoked token, unpaired device, unsupported protocol) from a transient one (rate limit) instead of probing-by-failure. The bundled client classifies each rejection with the shared is_recoverable(code) helper: terminal codes stop the reconnect loop and fire on_reconnect_paused(code, next_step); transient codes back off and retry, honouring any server retry_after_seconds as a lower bound.

from praisonai_bot.gateway.client import GatewayClient
from praisonaiagents.gateway import ConnectErrorCode

client = GatewayClient(url="wss://gateway.example.com/ws", agent_id="my-agent")

def on_paused(code: ConnectErrorCode, next_step):
    print(f"Reconnect paused: {code.value} — next: {next_step}")

client.on_reconnect_paused = on_paused
await client.connect()

Configuration Options

Edge protection fields (available on GatewayConfig and gateway.yaml):

Python-first surface (parity with CLI/YAML)

A Python embedder can express the graceful-drain and event-loop-watchdog knobs directly on GatewayConfig — no CLI flag and no YAML key required.

from praisonaiagents.gateway import GatewayConfig

config = GatewayConfig(
    drain_timeout=20,        # bounded drain on shutdown (seconds)
    watchdog=True,           # event-loop watchdog on
    watchdog_timeout=30,     # loop-stall budget before self-exit (seconds)
)

Precedence is unchanged: **CLI flag > explicit YAML (block or scalar, incl. watchdog: false) > GatewayConfig field > default**. An explicit gateway.watchdog: false in YAML is honoured and is not re-enabled by GatewayConfig(watchdog=True). See Graceful Drain and Event-Loop Watchdog for the per-knob detail.

Gateway YAML Configuration

The gateway.yaml schema accepts three optional top-level blocks alongside channels and agents. All three are optional and ignored when not present.

gateway: block

Global gateway settings — reliability preset, drain timeout, health monitor:

gateway:
  reliability: production    # "production" | "default" | "off"
  drain_timeout: 15          # seconds to wait for in-flight turns on shutdown

channels:
  local: {}                  # chat from your own terminal — no token
  telegram:
    platform: telegram
    token: ${TELEGRAM_BOT_TOKEN}
    stt:
      enabled: true          # transcribe inbound voice notes (default on)
      echo_transcripts: true  # echo the recognised text back to the user
  discord:
    platform: discord
    token: ${DISCORD_BOT_TOKEN}
    stt: true                # Discord voice notes are transcribed too (default on)

Inbound voice notes on Telegram, Slack, WhatsApp, Discord, and Signal are transcribed automatically — see Voice Notes.

Local terminal channel

Add local: {} alongside your remote channels to chat with the same live agent from your own terminal — no token, no webhook, one shared session across both surfaces. See Local (Terminal) Channel.

The gateway: block validates every knob below. Extra keys are rejected (extra="forbid"), so a typo like drain_timout fails at load with a field-named error instead of silently running with the default.

api: block

Serve OpenAI-compatible and MCP HTTP endpoints from the same gateway process, sharing the live agents and sessions that chat users reach:

gateway:
  api:
    openai: true    # mounts /v1/chat/completions, /v1/responses (+ background/{id}/cancel), /v1/models
    mcp: true       # mounts /mcp (JSON-RPC)

Both surfaces are opt-in and off by default. See Gateway API Endpoints for the endpoint list, auth, and client examples.

health: block

Nest a health: block under gateway: to configure the worker health monitor. Extra keys are rejected (extra="forbid"):

gateway:
  health:
    enabled: true
    interval: 60
    stale_after: 90
    stuck_after: 600
    max_restarts_per_hour: 5
    fleet_restarts_per_hour: 40
    failing_channel_fraction: 0.5
    breaker_cooldown_s: 120

hooks: block

Register inbound trigger hooks that expose an authenticated POST /hooks/<path> endpoint and run an agent (or wake a session) on each request. Each hook requires a non-empty path::

hooks:
  - path: gmail
    agent: assistant
    action: agent           # optional; "agent" or "wake"

See Gateway Inbound Hooks for signature-verification, event-filter, and deliver-only examples.

scheduler: block

Tune the default RunPolicy the gateway applies to every unattended scheduled run. All fields are optional; omitted fields keep the fail-closed defaults.

scheduler:
  deliver_on_failure: true          # send a failure summary to the job's delivery target
  scan_prompt: true                 # scan the assembled prompt before the turn
  denied_toolsets:                  # tool categories removed from unattended runs
    - cronjob
    - shell
  audit_dir: /var/log/praisonai/scheduler   # durable full-output audit trail

Overrides apply on the next tick without a process restart. See Scheduled Run Policy for the full policy semantics.

mcp_servers: block

Declare downstream MCP servers once; every agent that lists the server name in tools: gets the same pooled client.

mcp_servers:
  github:
    command: npx
    args: ["-y", "@modelcontextprotocol/server-github"]
  search:
    url: https://mcp.example.com/sse
    headers:
      Authorization: Bearer ${SEARCH_MCP_TOKEN}

agents:
  support:
    instructions: "Answer questions using GitHub and web search."
    tools: [github, search]

See Gateway MCP Servers for hot-reload, pool shutdown, failure modes, and tool-name namespacing.

BotOS.from_config and tool names

BotOS.from_config("botos.yaml") now routes tool names through the standard ToolResolver chain:

1. Local tools.py in the working directory
2. Wrapper ToolRegistry
3. praisonaiagents built-in tools
4. praisonai-tools package
5. Installed plugins

This means a YAML bot referencing a praisonai-tools symbol or a local tools.py function just works — no ad-hoc importlib lookup required:

# botos.yaml
agent:
  name: assistant
  instructions: You are a helpful assistant.
  tools:
    - web_search          # resolved via praisonai-tools
    - my_custom_tool      # resolved from local tools.py
platforms:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}

from praisonai.bots import BotOS

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

Gateway Agent Defaults

Gateway agents loaded from YAML use chat-optimised defaults that differ from the Python SDK.

agents:
  assistant:
    instructions: "You are a helpful AI assistant."
    model: gpt-4o-mini
    reflection: true   # opt-in: enables self-critique, ~8x slower

Per-Agent Approval Isolation

“Allow always” grants persist across gateway restart and default to being scoped to the approving agent, so one agent’s approval never authorises another. Grants live in a durable SQLite store at ~/.praisonai/state/gateway/approvals.sqlite.

Unanswered approvals fail closed and say so: the Slack / Telegram / Discord card edits to “⌛ Timed out — not run”, and a GatewayApprovalBackend you build yourself with notify_url sends a follow-up "timed_out" POST (the auto-wired default has no notify_url, so it sends none). See Timeout feedback on the card and GatewayApprovalBackend webhook payload.

Gateway also supports isolated exec approval managers for multi-tenant environments where agents must not share approval state.

Default vs Per-Agent Managers

from praisonai.gateway.exec_approval import (
    get_default_exec_approval_manager,
    create_exec_approval_manager
)

# Default manager - shared across all agents
default_manager = get_default_exec_approval_manager(ttl=300.0)

# Per-agent manager - isolated approval state
agent1_manager = create_exec_approval_manager(ttl=120.0)
agent2_manager = create_exec_approval_manager(ttl=180.0)

Multi-Agent Gateway with Isolation

from praisonaiagents import Agent
from praisonai.gateway.exec_approval import create_exec_approval_manager

class IsolatedGateway:
    def __init__(self):
        self.agents = {}

    def add_agent(self, name, instructions, approval_ttl=300):
        # Each agent gets isolated approval manager
        manager = create_exec_approval_manager(ttl=approval_ttl)

        agent = Agent(
            name=name,
            instructions=instructions,
            exec_approval_manager=manager,
            gateway=True
        )

        self.agents[name] = {"agent": agent, "manager": manager}
        return agent

    def get_pending_approvals(self, agent_name):
        if agent_name in self.agents:
            manager = self.agents[agent_name]["manager"]
            return manager.list_pending()
        return []

# Usage
gateway = IsolatedGateway()

# Tenant A agent
tenant_a = gateway.add_agent(
    "TenantA_Assistant", 
    "Help tenant A users",
    approval_ttl=180  # 3 minutes
)

# Tenant B agent  
tenant_b = gateway.add_agent(
    "TenantB_Assistant",
    "Help tenant B users", 
    approval_ttl=600  # 10 minutes
)

# Isolated approval queues
pending_a = gateway.get_pending_approvals("TenantA_Assistant")
pending_b = gateway.get_pending_approvals("TenantB_Assistant")

Allow-list endpoint

GET/POST/DELETE /api/approval/allow-list manages allow-always grants. POST and DELETE accept an optional agent_id; GET returns a grants view alongside the legacy allow_list.

{
  "allow_list": ["read_file"],
  "grants": [
    {"agent_id": "shell-bot", "tool_name": "shell_exec", "arg_signature": null, "created_at": 1720000000.0, "approver": "gateway:human"}
  ]
}

Omitting agent_id on POST/DELETE targets a legacy any-agent grant; supplying it scopes to that agent. See Gateway Scoped Approvals for full curl examples.

Gateway lifecycle predicates

Core SDK exports three pure predicates for gateway lifecycle decisions. Each is side-effect free and testable in isolation; the wrapper owns the actual teardown.

from praisonaiagents.gateway import (
    ScaleToZeroPolicy,
    DrainTimeoutPolicy,
    DrainMarkerPolicy,
    current_epoch,
    is_weak_secret,
    assert_gateway_secret_strong,
    WeakGatewaySecretError,
    KNOWN_WEAK_SECRETS,
    ThreadResult,
    ThreadStatus,
)

Python API — repair and provisioning

The same detect → repair → re-validate lifecycle behind praisonai gateway doctor --fix and the interactive praisonai onboard wizard is available as a typed Python API — call it from a management script or notebook instead of shelling out to the CLI.

from praisonai_bot import (
    repair_gateway_config,
    provision_gateway_config,
    GatewayRepairResult,
)

Repair from a management script

from praisonai_bot import repair_gateway_config

result = repair_gateway_config("~/.praisonai/bot.yaml", fix=True)

print("token minted:", result.auth_token_minted)
print("version migrated:", result.config_version_migrated)
for line in result.changes:
    print("changed:", line)

# JSON-safe copy for structured logging
import json
print(json.dumps(result.to_dict()))

Provision on first run

from praisonai_bot import provision_gateway_config

path = provision_gateway_config(
    platform="telegram",
    token="${TELEGRAM_BOT_TOKEN}",   # persisted to ~/.praisonai/.env
    agents=["assistant"],
)
print(path)   # ~/.praisonai/bot.yaml

provision_gateway_config is fail-closed: an empty token or unsupported platform raises ValueError and writes nothing.

Kanban Dispatcher

Kanban dispatcher (v4.6.x): Worker file descriptors are now released as soon as the subprocess starts (fixes a slow leak under sustained dispatch). release_claim failures are logged with full stack traces instead of being silently swallowed — orphaned claimed tasks now leave a clear log trail for triage. KeyboardInterrupt / SystemExit / asyncio cancellation propagate cleanly through dispatcher shutdown.

Kanban dispatcher (off-loop git): Every git command the dispatcher runs — worktree add, merge --no-ff, commit, status, worktree remove — now runs in a worker thread, off the gateway event loop. A slow git step no longer freezes chats, WebSocket pings or health probes.

• Serialised per gateway process. A single in-process lock, shared by every dispatcher, lets only one blocking git section run at a time, so two dispatchers in the same process never run git merge concurrently. Running dispatchers in separate processes against one repo? Serialise at the repo level yourself (file lock, dedicated worker host).
• Shutdown stops the whole worker group. On POSIX, workers start in their own session. Shutdown sends SIGTERM to each worker’s process group, waits up to 5 s for all of them together, then SIGKILLs the group — so shells, dev servers and test runners a worker spawned don’t outlive the gateway. Windows is unchanged.
• Cancel during worktree prep is clean. If the dispatcher stops while git worktree add is running, it waits for git to finish, removes the unused worktree and kanban/<task_id> branch, closes the run as failed and releases the claim so a retry starts fresh.
• Every git command the dispatcher runs is now bounded by PRAISONAI_KANBAN_GIT_TIMEOUT (default 300 s). A hung git operation returns returncode=124 with a diagnosis in stderr instead of stalling the event loop. See Gateway Subprocess Guard.

Common Patterns

Single Gateway Deployment

from praisonaiagents import Agent

# Simple unified gateway
agent = Agent(
    name="Gateway Agent",
    gateway=True,  # Enables unified gateway
)

# CLI deployment
# praisonai serve unified --port 8765

Multi-Agent Gateway

from praisonaiagents import Agent, Task, PraisonAIAgents

agents = [
    Agent(name="Researcher", gateway=True),
    Agent(name="Writer", gateway=True),
]

# Multi-agent coordination through gateway
tasks = [Task(description="Research topic", agent=agents[0])]
crew = PraisonAIAgents(agents=agents, tasks=tasks)

Development Gateway

from praisonaiagents import Agent

agent = Agent(
    name="Dev Agent",
    gateway=True,
    debug=True  # Development features
)

# CLI: praisonai serve unified --reload

Best Practices

Exit Codes & Supervisor Integration

praisonai serve gateway uses sysexits-based exit codes so process supervisors can distinguish a transient failure (restart) from a fatal misconfiguration (stop and fix).

What Triggers Fatal (78)

systemd Unit

[Unit]
Description=PraisonAI Gateway
After=network.target

[Service]
ExecStart=/usr/local/bin/praisonai serve gateway
Restart=on-failure
RestartSec=5

# Stop the crash-loop on fatal config — fix the config and start manually
RestartPreventExitStatus=78

[Install]
WantedBy=multi-user.target

Kubernetes Restart Policy

apiVersion: v1
kind: Pod
spec:
  containers:
    - name: praisonai-gateway
      image: praisonai:latest
      command: ["praisonai", "serve", "gateway"]
  restartPolicy: OnFailure

For Kubernetes, use a wrapper script that maps exit code 78 to success (exit 0) to prevent the pod from restarting on fatal config:

#!/bin/sh
praisonai serve gateway "$@"
code=$?
if [ $code -eq 78 ]; then
  echo "Fatal config error — not restarting (code 78)" >&2
  exit 0
fi
exit $code

In-Place Code Updates and /model

When a running gateway’s code changes on disk (via git pull, pip install -U, or an auto-update on a durable volume), the first-use lazy import triggered by a /model switch can hit a code mismatch and crash with a cryptic ImportError. The code-skew guard refuses with a clear message instead.

How It Works

At startup, the gateway captures a fingerprint of the installed code (git SHA + newest .py mtime). Before each /model switch it compares this to the current on-disk fingerprint:

User-Facing Message

When skew is detected, the user sees:

⚠️ Gateway code changed on disk since it started (<boot-short> → <disk-short>). Restart the gateway to apply updates before switching models.

Where <boot-short> and <disk-short> are 7-character shortened fingerprints (e.g. git SHAs or mtime: suffixes).

Fail-Open

The guard is best-effort and fail-open: if the fingerprint cannot be determined (no git, unreadable directory), the guard returns None and the /model switch proceeds normally. Normal operation is never blocked by the guard itself.

Opt-Out

Disable the guard for a session manager:

session_manager.code_skew_guard = False

Threads / Topics

Open a new thread under a chat so a delegated subtask has its own scoped conversation instead of flooding the shared channel.

The running gateway registers an OutboundMessengerProtocol into the per-turn context. Resolve it with get_outbound_messenger(), open a thread, then route the subtask into it — falling back to the parent target when the platform can’t thread.

from praisonaiagents.session.context import get_outbound_messenger

async def handoff():
    messenger = get_outbound_messenger()
    res = await messenger.create_thread("telegram:12345", name="research")
    target = f"telegram:12345:{res.thread_id}" if res.ok else "telegram:12345"
    await messenger.send(target, "starting research...")

How It Works

ThreadResult

create_thread(target, name) -> ThreadResult never raises — a channel that cannot thread returns a typed unsupported outcome and the caller falls back to the parent target. Exported from praisonaiagents.gateway as ThreadResult and ThreadStatus.

ThreadStatus is a closed set of four values:

• ok — a new thread/topic was opened; thread_id carries its id.
• unsupported — the channel has no supports_threads capability.
• failed — the transport rejected the attempt (topics mode off, or missing permission).
• no_route — the target could not be resolved to a reachable channel.

Fallback Pattern

Compose the routed target from the result — never wrap the call in try/except, it always resolves to a typed outcome:

res = await messenger.create_thread(parent, name="subtask")
target = f"{parent}:{res.thread_id}" if res.ok else parent
await messenger.send(target, subtask_output)

Capabilities: supports_threads

Threading is gated on the channel’s PlatformCapabilities.supports_threads, which defaults to False — so unaware adapters degrade to unsupported, safe by default. Adapters that can open native threads must set supports_threads=True.

Related