Channels

Trigger agents from cron schedules, HTTP requests, WebSockets, file changes, and more.

Channels define when and how messages enter and leave Zgent. A channel implements the Receiver protocol (to produce messages), the Sender protocol (to consume messages), or both.

Built-in channels

Channel	Direction	Use case
`MemoryChannel`	Receiver + Sender	Testing, in-process communication
`HttpChannel`	Receiver + Sender	REST API endpoints, webhooks
`WebSocketChannel`	Receiver + Sender	Real-time bidirectional communication
`CronChannel`	Receiver	Scheduled triggers (cron expressions)
`FileWatchChannel`	Receiver	Trigger on file system changes

MemoryChannel

An in-process message queue, useful for testing and programmatic interaction:

from zgent.channel.memory import MemoryChannel
from zgent.message.types import UserMessage

channel = MemoryChannel(name="test")
await channel.connect()

# Send a message into the channel
await channel.send(UserMessage(session_id="s1", text="Hello"))

# Receive from the channel
msg = await channel.receive(timeout=5.0)

CronChannel

Triggers agent execution on a schedule:

from zgent import CronChannel

# Every hour
cron = CronChannel(expression="0 * * * *", session_id="hourly-job")

HttpChannel

Exposes an HTTP endpoint for receiving messages:

from zgent import HttpChannel

http = HttpChannel(host="0.0.0.0", port=8080, path="/webhook")

WebSocketChannel

Bidirectional real-time communication:

from zgent import WebSocketChannel

ws = WebSocketChannel(url="ws://localhost:9000/agent")

FileWatchChannel

Triggers when files change in a directory:

from zgent import FileWatchChannel

watcher = FileWatchChannel(path="/data/inbox", patterns=["*.csv"])

Custom channels

Implement the Receiver and/or Sender protocols to create a custom channel:

from zgent.channel.base import Receiver, Sender
from zgent.message import Message

class TelegramChannel(Receiver, Sender):
    async def connect(self) -> None:
        # Initialize bot connection
        ...

    async def disconnect(self) -> None:
        # Clean up
        ...

    async def receive(self, timeout: float | None = None) -> Message | None:
        # Poll for Telegram updates
        ...

    async def send(self, message: Message) -> None:
        # Send message to Telegram chat
        ...

See the MiniClaw example for working Telegram, Discord, and Slack channel implementations.