Channels

Narrative contract (sockets, frames, HITL): Connections.

Publish helpers and wire protocol (Monitor, Messenger, Catalog):

WebSocket publish helpers for Agency Channels.

Wire protocol and abstractions (Monitor, Messenger, Catalog) live in agency.channel.protocol. Consumers live under agency.channel.consumer. This module exposes publish_event, group helpers, and Publisher.

agency.channel.publish_event(event_type, groups, **kwargs)

Publish an event to the task’s channel group and to the general tasks-all group.

Args: - event_type: The type of event (task_started, task_success, etc.) - groups: The list of groups to publish the event to - **kwargs: The event data to be passed to the consumer handler

agency.channel.channel_log(message, data=None, obj=None, level='info')

Publish a log event to the logs channel group.

Notes: - type=”log” is required for Django Channels dispatch to LogConsumer.log(…). - The browser-level “severity” should be sent as level (info|warning|error|debug).

agency.channel.channel_run(event_type, run, **kwargs)

Publish a thread event to the thread channel groups.

Parameters:

• event_type (str) – The type of event (tool_call, tool_result, model_request, etc.)

• run (Run) – Run instance to get thread_id

• **kwargs – Event data to be passed to the consumer handler (will be serialized)

agency.channel.channel_user(event_name, run, **kwargs)

Send a message to the user’s channel group.

Parameters:

• event_name (str) – The event name (e.g., “Run.before_agent”, “Run.after_agent”)

• run (Run) – Run instance to get user from thread

• **kwargs – Event data to be passed (will be serialized)

class agency.channel.Publisher(alias='default')

Bases: object

Centralized class for sending messages via Django Channels.

Django Channels provides a channel layer abstraction for sending messages between different parts of your application, typically from backend code to WebSocket consumers.

How it works:

1. Channel Layer: A messaging system (usually Redis or in-memory) that routes messages between different parts of your application.

2. Groups: Named collections of WebSocket connections. When you send to a group, all connected WebSocket clients in that group receive the message.

3. Message Format: Messages are dictionaries that must include a ‘type’ key. The ‘type’ determines which handler method is called in the consumer.

4. Consumers: WebSocket handlers that receive messages. They define methods like async def event_type(self, event) to handle different message types.

Note: All data is automatically serialized to ensure JSON compatibility.

__init__(alias='default')

Initialize the Channel with a specific channel layer.

Parameters:

alias (str) – The channel layer alias from Django settings (default: ‘default’)

Raises:

ValueError – If the channel layer is not found

send(channel, message)

Send a message directly to a specific channel.

This sends to a single channel name (not a group). Use this when you know the exact channel name of a WebSocket connection.

Parameters:

• channel (str) – The specific channel name to send to

• message (dict) – The message dict (must include ‘type’ key)

group_send(group, message)

Send a message to a channel group.

All WebSocket connections in the group will receive this message.

Parameters:

• group (str) – The group name (e.g., “logs”, “thread-123”, “user-456”)

• message (dict) – The message dict (must include ‘type’ key)

send_to_groups(groups, message)

Send a message to multiple channel groups.

Parameters:

• groups (list[str]) – List of group names to send to

• message (dict) – The message dict (must include ‘type’ key)

consumer_handle(event_type, groups, payload=None)

Publish an event to channel groups with automatic timestamp and serialization.

This is a convenience method that: - Adds a timestamp to the event - Serializes all kwargs to ensure JSON compatibility - Sends to multiple groups

Parameters:

• event_type (str) – The event type (determines consumer handler method)

• groups (list[str]) – List of group names to send to

• **kwargs – Event data (will be serialized)

Agency real-time protocol.

This module defines how Agency communicates execution state and messages to clients over WebSockets.

Philosophy

• The database is the source of truth.

• Frontends rebuild state from the API and apply real-time deltas via WS.

• WebSocket traffic is a projection of persisted state, not a replacement for it.

• Events and user messages are intentionally separated.

Notes

• Monitor emits structured runtime and lifecycle events (telemetry).

• Messenger streams end-user conversation messages.

• Catalog publishes catalog / schema-style updates for UI surfaces.

• Groups maps domain objects (Run, User, …) to Channels group names.

• Signals (agency.signals.wire) bridge persistence to real-time updates.

• Prefer Monitor/Messenger over legacy channel_run helpers (deprecated).

• Payloads should be lightweight, idempotent, and replay-safe.

• This module defines protocol and transport concerns, not business logic.

class agency.channel.protocol.Groups(obj=None)

Bases: list[str]

Resolve domain objects into Django Channels group names.

Groups define how events and messages are routed over WebSockets. They map Runs, Threads and Users to stable, broadcast-friendly channel group identifiers.

Notes: - Group names must be safe for Channels (no colons). - Groups are derived, never persisted. - A single event may be routed to multiple groups (e.g. thread + user).

__init__(obj=None)

Initialize Groups.

Parameters:

obj (Run | Any | list[str] | None) –

• Run instance: derives groups for thread and user

• User instance: derives groups for that user

• list[str]: uses the provided group names as-is

• None: creates empty list

class agency.channel.protocol.Messenger(user)

Bases: Publisher

Stream end-user conversation messages over WebSockets.

Messenger delivers BaseMessage instances (human, assistant, tool, system) to clients in real time. This is the user-facing message stream.

Notes: - Messages are streamed for UI consumption. - Messages reflect execution output but are not the source of truth.

ICON_MAP = {'ai': '😑', 'assistant': '😑', 'human': '👤', 'system': '💬', 'tool': '🔧'}

__init__(user)

Initialize Messenger with a user.

The publisher sends messages to the user’s channel group.

Parameters:

user (Any) – User instance to send messages to

stream(message, **kwargs)

Publish a LangChain message to the user’s channel.

Automatically determines the icon based on message type and includes the serialized message data. The message will be sent to the user’s channel group and can be handled by UserConsumer.

Parameters:

message (BaseMessage | dict) – LangChain BaseMessage instance or dict with message data (must have ‘type’ field if dict)

Example

publisher = Messenger(user) publisher.stream(AIMessage(content=”Hello!”)) # Or with a dict: publisher.stream({“type”: “ai”, “content”: “Hello!”})

class agency.channel.protocol.Monitor(instance)

Bases: Publisher

Emit runtime and lifecycle events for a Run.

Monitor produces structured events describing what happens during execution (state transitions, model and tool activity, interruptions). These events are intended for telemetry, debugging and admin UIs.

Notes: - Events describe what happened, not what to do. - Some events originate from persistence (signals), others from runtime. - Delivery is best-effort and may be lossy.

class Event

Bases: object

Canonical agency event names used over WebSockets.

These constants form the wire protocol: JS clients should switch on these exact strings when handling “monitor_event” messages in the WebSocket consumer.

CHECKPOINT_CREATED = 'checkpoint_created'

RUN_STATUS = 'run_status'

RUN_START = 'run_start'

MODEL_REQUEST = 'model_request'

MODEL_RESPONSE = 'model_response'

TOOL_CALL = 'tool_call'

TOOL_RESULT = 'tool_result'

TOOL_ERROR = 'tool_error'

TOOL_INTERRUPT = 'tool_interrupt'

TOOL_TIMEOUT = 'tool_timeout'

TOOL_CANCEL = 'tool_cancel'

TOOL_ABORT = 'tool_abort'

RUN_INTERRUPT = 'run_interrupt'

INTERRUPTION_RESOLVED = 'interruption_resolved'

RUN_END = 'run_end'

RUN_ERROR = 'run_error'

CATALOG_CHANGED = 'catalog_changed'

__init__(instance)

Initialize Monitor publisher with an instance.

The publisher uses the Groups class to automatically determine the groups to publish to based on the instance type (e.g., Run).

Parameters:

instance (Any) – The instance to publish events for (typically a Run instance)

emit(event, data=None)

Emit an agency event to the appropriate WebSocket groups.

This is the main API used by backend components (e.g. RunMiddleware) to send structured events about a Run / Thread to the frontend.

It automatically: - derives target groups via Groups(instance) (user/thread-level groups) - includes the bound instance in the payload - merges any additional data dict from the caller

Parameters:

• event (str) – The event type (e.g., Monitor.Event.RUN_START)

• data (Any) – Optional additional data to include. If None, only the instance is included in the event data.

Example

events = Monitor(run) events.emit(Monitor.Event.RUN_START) # Or with additional data: events.emit(Monitor.Event.TOOL_CALL, {“tool_name”: “search”})

class agency.channel.protocol.Catalog(alias='default')

Bases: Publisher

Publish catalog CRUD deltas to the users group as bus envelopes.

Used by explorers (Agents, …): REST mutates DB; signals emit <resource>.<id>.catalog.<action> so open UIs stay in sync.

GROUPS = ['users']

emit(*, resource, action, id, record=None, actor_id=None, scope=None)

Emit a catalog change.

Parameters:

• resource (str) – API stem (agent, tool, …).

• action (str) – created | updated | deleted.

• id (str) – Primary key as string.

• record (dict | None) – Optional card payload (omit/empty on delete).

• actor_id (int | None) – Optional user id that caused the change.

• scope (dict | None) – Optional filter hints for clients (workspace_id, …).