Designing Multi-Channel Agent Architectures

The moment you connect an agent to a second communication channel, you face a design fork. Either you write channel-specific logic twice — one handler for Telegram, another for email, a third for your terminal — or you introduce a message abstraction that normalizes everything into a common format. The first path is fast to start. The second is the only one that survives past three channels.

This article walks through the channel abstraction pattern, how to handle capabilities that only exist on some channels, and the async orchestration model that lets channels run independently while sharing agent state. The examples come from humux, where this architecture has been running in production across Telegram, email, CalDAV, and CLI for over a year.

The common message format

Every channel — Telegram, IMAP, a terminal prompt — produces something different. Telegram gives you an Update object with nested entities, reply markup, and media arrays. Email gives you a MIME tree. A CLI gives you a string. The first thing you need is a dataclass that all of these normalize into:

@dataclass
class Message:
    text: str
    sender: str
    channel: str            # "telegram", "email", "cli"
    chat_id: str            # normalized identifier
    timestamp: datetime
    attachments: list[Attachment] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)
    reply_to: str | None = None

The metadata dict is deliberate. It carries channel-specific data — Telegram's message_id for reactions, an email's In-Reply-To header for threading — without polluting the core schema. Everything downstream of this dataclass works with the same interface regardless of origin.

Each channel implements one function: convert its native format into a Message. That's the entire abstraction. No abstract base class with twelve methods. One function per channel.

def normalize(update: Update) -> Message:
    msg = update.message or update.edited_message
    attachments = []
    if msg.voice:
        attachments.append(Attachment(
            kind="voice", file_id=msg.voice.file_id
        ))
    return Message(
        text=msg.text or msg.caption or "",
        sender=str(msg.from_user.id),
        channel="telegram",
        chat_id=str(msg.chat.id),
        timestamp=msg.date,
        attachments=attachments,
        metadata={"message_id": msg.message_id},
    )

Channel-specific capabilities

Not every channel can do the same things. Telegram has inline keyboards, reactions, and voice messages. Email has CC/BCC, attachments, and threading. CLI has neither — but it has the lowest latency and doesn't require network access.

The temptation is to model these as a capability matrix: a registry of features each channel supports, queried at runtime. Don't. That approach scales poorly and forces every new feature through a capabilities check. Instead, handle channel-specific output at the response layer:

@dataclass
class Response:
    text: str
    voice: bytes | None = None
    buttons: list[list[Button]] | None = None
    reaction: str | None = None

async def deliver(response: Response, channel: str, chat_id: str):
    match channel:
        case "telegram":
            if response.voice:
                await send_voice(chat_id, response.voice)
            elif response.buttons:
                await send_with_keyboard(chat_id, response.text,
                                         response.buttons)
            else:
                await send_text(chat_id, response.text)
        case "email":
            await send_email(chat_id, response.text)
        case "cli":
            print(response.text)

The agent core produces a Response with everything it wants to say. Each channel picks what it can handle. Telegram renders the buttons; email ignores them and just sends the text. CLI prints it. The agent doesn't know or care which channel is listening. This is the critical separation: the agent thinks in responses, channels think in delivery.

Async orchestration and shared state

Each channel runs as an independent async loop. Telegram uses long polling (or webhooks). Email polls IMAP on an interval. CLI blocks on stdin. They don't coordinate with each other — they all feed into the same agent through the normalized Message interface.

State sharing happens through the database, not through in-memory structures. In humux, all channels write to the same SQLite history database (WAL mode for concurrent readers). When the agent processes a message from Telegram, it sees the full conversation history including CLI interactions. When you switch to the terminal to test something, the agent has full context from the Telegram thread.

async def run(agent: Agent, channels: list[Channel]):
    async with TaskGroup() as tg:
        for ch in channels:
            tg.create_task(ch.listen(
                callback=lambda msg: handle(agent, msg)
            ))

async def handle(agent: Agent, msg: Message):
    history = await db.get_history(msg.chat_id)
    response = await agent.think(history + [msg])
    await db.save(msg)
    await db.save_response(response, msg.chat_id)
    await deliver(response, msg.channel, msg.chat_id)

The TaskGroup pattern (Python 3.11+) is clean here: each channel is a long-lived task, and if one crashes, the group propagates the exception cleanly. No supervisor trees, no restart policies. The process manager handles restarts at the container level.

Per-agent bots and identity isolation

humux takes this a step further: each agent gets its own Telegram bot token. Instead of one bot routing messages to different agents, each agent is a separate bot with its own username, avatar, and conversation history. This matters in group chats — you can add multiple agents to the same group, and each responds independently based on an addressing protocol (mention the bot's name, it responds; otherwise it stays quiet).

The identity isolation extends to all channels. Each agent can have its own email account, its own calendar bindings, its own set of skills. The channel layer doesn't know about this — it just maps chat_id to the right agent and lets the agent core handle the rest.

CLI as a first-class channel

Most agent frameworks treat CLI as a debug tool. That's a mistake. A terminal channel with the same message format means you can test any agent behavior locally without sending a single Telegram message. In humux, the CLI channel uses SQLite WAL for session persistence — you can Ctrl-C and resume the same conversation later.

It also serves as the fallback. If the Telegram API is down, scheduled jobs still execute and deliver results through the CLI channel's history. When the network recovers, the agent has full context of what happened offline.

Key takeaways

The multi-channel pattern boils down to three decisions: normalize input into a common Message, let the agent produce channel-agnostic Response objects, and share state through the database rather than in-memory structures. Each channel runs as an independent async task. Channel-specific capabilities are handled at the delivery layer, not the agent layer.

The entire abstraction is two dataclasses, one deliver function with a match statement, and one normalize function per channel. No framework, no plugin system, no capability registry. That's the point — the architecture that survives is the one simple enough that you can hold it in your head while debugging at 2am.