Phoenix Channels Python Client

A modern, async Python client library for connecting to Phoenix Channels - the real-time WebSocket layer of the Phoenix Framework.

Installation

For Users

pip install phoenix-channels-python-client

Or from source:

git clone https://github.com/band-ai/phoenix-channels-python-client.git
cd phoenix-channels-python-client
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install .

For Development

Set up development environment:

git clone https://github.com/band-ai/phoenix-channels-python-client.git
cd phoenix-channels-python-client
uv sync --extra dev

Run tests:

uv run pytest

Dependencies

• Python 3.11+
• pydantic>=2.8
• websockets>=14.2

Quick Start

Here's a minimal example to get you started:

import asyncio
from phoenix_channels_python_client import PHXChannelsClient

async def main():
    # Connect to Phoenix WebSocket server
    async with PHXChannelsClient(
        websocket_url="ws://localhost:4000/socket/websocket",
        api_key="your-api-key"
    ) as client:
        
        # Define message handler
        async def handle_message(message):
            print(f"Received: {message}")
        
        # Subscribe to a topic
        await client.subscribe_to_topic("room:lobby", handle_message)
        
        # Ctrl+C or SIGTERM shuts the client down and run_forever() returns.
        # The previous handlers are restored first, so a second signal reaches them.
        await client.run_forever()

if __name__ == "__main__":
    asyncio.run(main())

Owning process signals yourself

If your application owns its process signals (a service, desktop app or test runner), call client.run_forever(install_signal_handlers=False). It then leaves signal handlers alone; stop the client by scheduling client.shutdown(reason) on the client's event loop, and run_forever() returns None:

loop = asyncio.get_running_loop()
async with asyncio.TaskGroup() as tasks:
    loop.add_signal_handler(
        signal.SIGTERM, lambda: tasks.create_task(client.shutdown("SIGTERM"))
    )
    try:
        await client.run_forever(install_signal_handlers=False)
    finally:
        # Once the group exits, a later SIGTERM must not reach the closed group.
        loop.remove_signal_handler(signal.SIGTERM)

From a plain signal.signal handler or another thread, use loop.call_soon_threadsafe(...) instead: a bare asyncio.create_task there doesn't wake a loop blocked on I/O, so the shutdown can stall until the next network activity. See examples/embedded_host_signals.py.

With the default, handlers are installed only when run_forever() runs on the main thread; elsewhere it just waits. A handler you registered with loop.add_signal_handler before calling run_forever() still fires for the same signal; one registered while it runs replaces the client's.

Phoenix System Events

Phoenix reserves phx_join, phx_reply, phx_leave, phx_close and phx_error. Don't use these names for your own events.

• The client sends phx_join and phx_leave and consumes their phx_reply.
• The client handles phx_error and phx_close itself, and neither reaches your handlers: a crashed channel is rejoined, and a channel the server closes loses its topic (see Channel recovery).
• add_event_handler raises ValueError for PHXEvent.error and PHXEvent.close. Use on_topic_lost and is_topic_joined() instead.

Protocol Versions

Phoenix Channels supports two protocol versions. Choose based on your Phoenix server version:

Protocol v2.0 (Default)

from phoenix_channels_python_client import PHXChannelsClient

async with PHXChannelsClient(
    websocket_url="ws://localhost:4000/socket/websocket",
    api_key="your-api-key"
    # protocol_version defaults to v2.0
) as client:
    ...  # your code here

Protocol v1.0

from phoenix_channels_python_client import PHXChannelsClient, PhoenixChannelsProtocolVersion

async with PHXChannelsClient(
    websocket_url="ws://localhost:4000/socket/websocket", 
    api_key="your-api-key",
    protocol_version=PhoenixChannelsProtocolVersion.V1
) as client:
    ...  # your code here

The client adds vsn=2.0.0 (V2) or vsn=1.0.0 (V1) to the socket URL.

Usage Examples

Basic Topic Subscription

import asyncio
from phoenix_channels_python_client import PHXChannelsClient

async def message_handler(message):
    print(f"Topic: {message.topic}")
    print(f"Event: {message.event}")
    print(f"Payload: {message.payload}")

async def main():
    async with PHXChannelsClient("ws://localhost:4000/socket/websocket", "api-key") as client:
        await client.subscribe_to_topic("chat:general", message_handler)
        
        # Use built-in method to keep connection alive
        await client.run_forever()

asyncio.run(main())

Message Handlers vs Event-Specific Handlers

The library provides two complementary ways to handle incoming messages:

Message Handler - Receives ALL messages for a topic:

• Set when subscribing to a topic or separately with set_message_handler()
• Gets the complete message object with topic, event, payload, etc.
• Good for logging, debugging, or handling any message type

Event-Specific Handlers - Receive only messages with specific event names:

• Added with add_event_handler() for particular events like "user_join", "chat_message", etc.
• Only get the payload (data) part of the message
• Perfect for handling specific application events

Execution order when you have both types of handlers:

1. Message handler runs first (if set) - receives the full message
2. Event-specific handler runs second (if matching event) - receives just the payload

All handlers must be async def. You can add, remove, or change either type of handler at any time during your connection.

Event-Specific Handlers

You can register handlers for specific events on a topic:

import asyncio
from phoenix_channels_python_client import PHXChannelsClient

async def handle_user_join(payload):
    print(f"User joined: {payload}")

async def handle_user_leave(payload):
    print(f"User left: {payload}")

async def main():
    async with PHXChannelsClient("ws://localhost:4000/socket/websocket", "api-key") as client:
        # Subscribe to topic first
        await client.subscribe_to_topic("room:lobby")
        
        # Add event-specific handlers for custom events
        client.add_event_handler("room:lobby", "user_join", handle_user_join)
        client.add_event_handler("room:lobby", "user_leave", handle_user_leave)
        
        # Use built-in method to keep connection alive
        await client.run_forever()

asyncio.run(main())

Unsubscribing from Topics

You can unsubscribe from topics to stop receiving messages and clean up resources:

import asyncio
from phoenix_channels_python_client import PHXChannelsClient

async def on_message(msg):
    print(f"Message: {msg.payload}")

async def main():
    async with PHXChannelsClient("ws://localhost:4000/socket/websocket", "api-key") as client:
        # Subscribe to a topic
        await client.subscribe_to_topic("room:lobby", on_message)
        
        # Do some work...
        await asyncio.sleep(5)
        
        # Unsubscribe when no longer needed
        await client.unsubscribe_from_topic("room:lobby")
        print("Unsubscribed from room:lobby")
        
        # Continue with other work or subscribe to different topics
        await client.run_forever()

asyncio.run(main())

Notes on unsubscription:

• Removes all handlers (both message and event-specific) for that topic
• Sends phx_leave and waits for the reply (leave_timeout_s, default 5 s)
• Raises PHXTopicError if the topic isn't subscribed or the leave times out, and PHXConnectionError while disconnected

Configuration

Everything after api_key is keyword-only:

Option	Default	Meaning
protocol_version	V2	Phoenix Channels protocol version
auto_reconnect	True	Reconnect and rejoin topics after a disconnect
reconnect_policy	ReconnectPolicy()	Backoff and close-code handling (see below)
heartbeat_interval_s	30.0	Heartbeat period; None disables heartbeats
join_timeout_s	10.0	Wait for a join reply
leave_timeout_s	5.0	Wait for a leave reply
max_topic_queue_size	1000	Per-topic buffer; the oldest message is dropped when full
callback_drain_timeout_s	2.0	How long the message being handled (its message handler, then its event handler) may finish before a rejoin cancels it; queued messages are dropped
on_reconnect	None	async () -> None, called after topics are rejoined
on_disconnect	None	async (error: Exception | None) -> None; None on a clean close
on_heartbeat_ack	None	Synchronous () -> None; runs on the message path, so keep it fast
on_topic_lost	None	async (topic: str, error: Exception) -> None, called when the client drops a topic you didn't unsubscribe
additional_headers	None	Extra handshake headers sent on every (re)connect

Exceptions raised in callbacks are logged, not raised.

Connection Lifecycle and Errors

• async with waits for the first connection. With auto_reconnect=True it keeps retrying, except after a refused handshake with HTTP 403, which raises PHXConnectionError; with auto_reconnect=False a failed first connection raises PHXConnectionError.
• run_forever() returns None after shutdown(), a signal, or a close that doesn't reconnect. It raises PHXConnectionError on a terminal close, a refused handshake with HTTP 403, after repeated rapid disconnects, or when the client has never entered async with.
• subscribe_to_topic() raises PHXTopicError on a rejected join, a join timeout or a duplicate subscription, and PHXConnectionError while disconnected or if the connection drops before the join completes.
• await client.close_connection(reason) force-closes the current connection; the client then reconnects if auto_reconnect is on. It does nothing if the client isn't connected or the connection is already closing.
• Exceptions live in phoenix_channels_python_client.exceptions.

Reconnects

Close code or handshake	Behaviour (defaults)
1000, 1001	Reconnect (Phoenix ≥ 1.8.8 sends 1001 for "reconnect expected"); stop with reconnect_on_normal_close=False
1008	Stop with PHXConnectionError (policy_violation_is_terminal=True)
Handshake refused with HTTP 403	Stop with PHXConnectionError; other statuses (401, 5xx, …) retry with the backoff below
1012	Reconnect after at least a random 1–5 s
1013	Reconnect after at least a random 30–60 s
Anything else, or no close frame	Reconnect with jittered backoff: 0.5 s × 2ⁿ, capped at 30 s, reset after 60 s of uptime

Disconnects within 5 s of connecting count as rapid and get longer minimum delays; ten within 60 s stop the client with PHXConnectionError. Tune all of this with ReconnectPolicy.

A ReconnectPolicy is validated when it is built: an out-of-range or unknown field raises pydantic.ValidationError, which is a ValueError. To derive one policy from another, validate the merged fields:

base = ReconnectPolicy()
policy = ReconnectPolicy.model_validate(base.model_dump() | {"max_delay_s": 10.0})

Channel recovery

A channel can fail while the socket stays up. When its process crashes, the server sends phx_error, and the client rejoins that topic on the same socket, waiting the ReconnectPolicy backoff (base_delay_s, factor, max_delay_s, with equal jitter) before each attempt. A rejoin that times out is retried the same way, including one after a socket reconnect. A drop of the socket hands the topic to the socket's own rejoin.

The client drops the topic and calls on_topic_lost(topic, error) once when the server rejects a rejoin, after a crash or a reconnect, closes the channel with phx_close, or its message processor fails. It never calls it for unsubscribe_from_topic(), shutdown() or a failed first subscribe_to_topic(), which raises instead. Subscribing to the topic again from the callback works. shutdown() cancels the callback, including when the callback itself called shutdown(), so code after that call does not run.

get_current_subscriptions() lists the subscribed topics. is_topic_joined(topic) tells whether a topic's channel is joined right now: it's false while the socket is down or a crashed channel waits to rejoin. Unsubscribing a topic whose channel is being recovered completes without waiting for a leave reply.

Heartbeats

The client sends a heartbeat on the phoenix topic every heartbeat_interval_s. A missed reply only logs a warning; to act on it, combine on_heartbeat_ack with close_connection().

Logging

import logging
from phoenix_channels_python_client import setup_logging

setup_logging(logging.DEBUG)  # loggers live under "phoenix_channels_python_client"

Examples

• examples/minimal_phx_events_client.py: subscribe to a topic and log messages until Ctrl+C or SIGTERM.
• examples/embedded_host_signals.py: run the client in a host that owns its process signals.

---

Security Considerations

API Key in URL: Without an authentication header, this client passes the API key as a URL query parameter when connecting to the WebSocket (e.g., wss://server.com/socket?api_key=xxx). Even over WSS (encrypted in transit), the API key may be logged by:

• Server access logs
• Reverse proxies and load balancers
• Network monitoring tools

Recommendations:

• Ensure your infrastructure does not log full URLs in production
• Use short-lived, rotatable tokens rather than long-lived API keys
• Consider this limitation when evaluating this client for sensitive environments

Pass additional_headers={"x-api-key": key} or {"x-auth-token": jwt} to use header-only authentication on every connection and reconnect.

---

Need help? Open an issue on GitHub or check the Phoenix Channels documentation for more information about the Phoenix Channels protocol.

Authentication and diagnostic safety

A nonempty x-api-key or x-auth-token handshake header selects header-only authentication on every connection and reconnect. Embedded API-key parameters are removed; empty or conflicting authentication headers and simultaneous token-query authentication are rejected before connecting. A JWT header overrides an embedded token and takes precedence over an API-key header, matching the platform. Without an authentication header, legacy query authentication is preserved and remains visible to servers and proxies. No fallback is attempted.

client.logger forwards sanitized connection records without raw arguments, exception objects, or socket metadata. client.redact_diagnostic(text) applies the credential policy to transport errors a caller needs to retain. Caller-visible errors may still contain personal data and should not be logged outside this policy.

Connection diagnostics omit dynamic strings and transport payloads, including URLs, headers, frame bodies, user/topic identities, and untrusted server error details. Logs retain static lifecycle descriptions, native connection states, and exception types. Application data and caller-visible errors are not changed; applications remain responsible for logs outside this connection-owned boundary.