Real-Time Push Notifications

Channel-based pub/sub over WebSocket with HTTP polling fallback

Subscribe to gateway channels and receive real-time messages — WebSocket first, polling when blocked.

from praisonaiagents import Agent
from praisonaiagents.push import PushClient

agent = Agent(
    name="alerts-agent",
    instructions="Summarise incoming alert events in one sentence",
)

client = PushClient("ws://localhost:8765/ws", auth_token="my-token")
await client.connect()

@client.on("channel_message")
async def on_msg(msg):
    print(agent.start(f"Summarise: {msg.data}"))

await client.subscribe("alerts")
await client.wait_closed()

The user subscribes to a channel; push events arrive over WebSocket and the agent summarises each message.

How It Works

Quick Start

1. Simple Usage

   Connect and subscribe:

   from praisonaiagents.push import PushClient
   
   client = PushClient("ws://localhost:8765/ws", auth_token="my-token")
   await client.connect()
   
   @client.on("channel_message")
   async def on_message(msg):
       print(f"{msg.channel}: {msg.data}")
   
   await client.subscribe("alerts")
   await client.wait_closed()

2. With Configuration

   Enable push on the gateway:

   from praisonaiagents import GatewayConfig
   from praisonaiagents.gateway import PushConfig
   
   config = GatewayConfig(push=PushConfig(enabled=True))

How It Works

Import paths:

from praisonai.push import PushClient                    # wrapper (recommended)
from praisonaiagents.push import PushClient              # lazy re-export
from praisonaiagents.push import ChannelMessage          # core dataclass

After fallback: producer contract

Polling is subscribe-only. After a WebSocket → polling fallback, publish(), create_channel(), and get_presence() raise NotImplementedError — the polling contract has no matching server routes. Gate producer code on PushClient.supports_publish before calling any of them.

await client.connect()
if client.supports_publish:
    await client.publish("alerts", {"kind": "info", "text": "hi"})
else:
    # WS blocked → subscribe-only mode. Publish elsewhere.
    logger.warning("Push client in polling fallback; skipping publish.")

supports_publish is True only when a transport is connected and is not the polling fallback. It is False before connect(), after disconnect(), and after a WS → polling fallback.

The client also emits a WARNING log the moment it falls back, so an operator watching the log knows publish is unavailable without instrumenting every call site:

PushClient fell back to polling transport. Unavailable operations under polling:
publish, create_channel, get_presence (they will raise NotImplementedError).
Gate producer code on the supports_publish property.

HA & cross-instance delivery

With PushConfig(redis=...), multiple gateway instances fan out channel messages through Redis pub/sub so a subscriber on any instance receives every publish.

Self-healing on Redis outage

The fan-out adapter (RedisPubSubAdapter) survives a dropped Redis connection without operator intervention:

• Bounded exponential backoff — reconnects starting at 1s, doubling on each failed attempt, capped at 30s. It never gives up unless the adapter is disconnected.
• Automatic re-subscription — every channel is re-subscribed on the fresh pub/sub handle after reconnect, so operators do not re-subscribe.
• Degraded surfacing — records itself as route:redis-pubsub in the degraded registry with a redacted reason and a retry_hint, so gateway status / gateway doctor / health()["degraded_owners"] show the outage.
• Dropped-write counting — publish, set_presence, remove_presence, store_message, and delete_message calls made while disconnected are counted (previously silent no-ops) and exposed as dropped_writes.
• Clears on recovery — the degraded record is removed and delivery resumes once Redis is reachable again.

What operators see

During an outage, gateway status reports the degraded transport under push and lists the owner in degraded_owners:

{
  "push": {
    "redis_connected": false,
    "redis_degraded": true,
    "redis_dropped_writes": 3,
    "online_clients": 12
  },
  "degraded_owners": [
    {
      "owner_kind": "route",
      "owner_id": "redis-pubsub",
      "state": "stale",
      "reason": "redis pub/sub disconnected: Connection refused",
      "retry_hint": "check Redis connectivity; see `praisonai gateway doctor`"
    }
  ]
}

After recovery, the three redis_* fields return to healthy and the route:redis-pubsub entry is gone:

{
  "push": {
    "redis_connected": true,
    "redis_degraded": false,
    "redis_dropped_writes": 3,
    "online_clients": 12
  },
  "degraded_owners": []
}

The reason only ever contains the exception message — the Redis connection URL and password are never surfaced. The log line to watch for on recovery is Redis push adapter reconnected (server_id=<uuid>).

Configuration Options

PushConfig

DeliveryConfig

PollingConfig

Best Practices

Related