Multi-Channel Chat Agents: One Agent for Web, WhatsApp, Slack, and SMS

The Multi-Channel Problem

Your customers are on WhatsApp, Slack, your website, and SMS. Building a separate chat agent for each channel means quadrupling your development and maintenance effort. The solution is a single agent brain with a channel abstraction layer that normalizes incoming messages and adapts outgoing responses to each channel's capabilities.

The architecture has three layers: channel adapters that handle platform-specific APIs, a normalization layer that converts all messages into a common format, and a rendering layer that converts the agent's structured output back into channel-specific formats.

The Common Message Format

Define a canonical message format that every channel converts to and from:

from pydantic import BaseModel
from enum import Enum
from datetime import datetime

class Channel(str, Enum):
    WEB = "web"
    WHATSAPP = "whatsapp"
    SLACK = "slack"
    SMS = "sms"

class Attachment(BaseModel):
    type: str  # "image", "file", "audio"
    url: str
    filename: str | None = None
    mime_type: str | None = None

class InboundMessage(BaseModel):
    channel: Channel
    channel_user_id: str
    conversation_id: str
    text: str
    attachments: list[Attachment] = []
    metadata: dict = {}
    timestamp: datetime

class OutboundMessage(BaseModel):
    text: str
    rich_elements: list[dict] = []  # Cards, buttons, etc.
    attachments: list[Attachment] = []
    quick_replies: list[dict] = []
    metadata: dict = {}

Channel Adapters

Each channel adapter translates between the platform's native format and your common format. Here is the WhatsApp adapter using the Cloud API:

from fastapi import APIRouter, Request
import httpx

router = APIRouter(prefix="/webhooks")

class WhatsAppAdapter:
    def __init__(self, phone_number_id: str, access_token: str):
        self.phone_number_id = phone_number_id
        self.access_token = access_token
        self.api_base = f"https://graph.facebook.com/v18.0/{phone_number_id}"

    def normalize(self, webhook_data: dict) -> InboundMessage | None:
        changes = webhook_data.get("entry", [{}])[0].get("changes", [{}])[0]
        value = changes.get("value", {})
        messages = value.get("messages", [])

        if not messages:
            return None

        msg = messages[0]
        return InboundMessage(
            channel=Channel.WHATSAPP,
            channel_user_id=msg["from"],
            conversation_id=msg["from"],  # WhatsApp uses phone as conversation ID
            text=msg.get("text", {}).get("body", ""),
            attachments=self._extract_attachments(msg),
            timestamp=datetime.utcnow(),
        )

    async def send(self, to: str, message: OutboundMessage):
        headers = {"Authorization": f"Bearer {self.access_token}"}

        # WhatsApp supports interactive messages with buttons
        if message.quick_replies:
            payload = {
                "messaging_product": "whatsapp",
                "to": to,
                "type": "interactive",
                "interactive": {
                    "type": "button",
                    "body": {"text": message.text},
                    "action": {
                        "buttons": [
                            {"type": "reply", "reply": {"id": r["value"], "title": r["label"][:20]}}
                            for r in message.quick_replies[:3]  # WhatsApp limit: 3 buttons
                        ],
                    },
                },
            }
        else:
            payload = {
                "messaging_product": "whatsapp",
                "to": to,
                "type": "text",
                "text": {"body": message.text},
            }

        async with httpx.AsyncClient() as client:
            await client.post(f"{self.api_base}/messages", json=payload, headers=headers)

    def _extract_attachments(self, msg: dict) -> list[Attachment]:
        attachments = []
        for media_type in ("image", "document", "audio", "video"):
            if media_type in msg:
                attachments.append(Attachment(
                    type=media_type,
                    url=msg[media_type].get("id", ""),
                    mime_type=msg[media_type].get("mime_type"),
                ))
        return attachments

And the Slack adapter:

class SlackAdapter:
    def __init__(self, bot_token: str):
        self.bot_token = bot_token

    def normalize(self, event: dict) -> InboundMessage | None:
        if event.get("type") != "message" or "bot_id" in event:
            return None

        return InboundMessage(
            channel=Channel.SLACK,
            channel_user_id=event["user"],
            conversation_id=event["channel"],
            text=event.get("text", ""),
            timestamp=datetime.utcnow(),
        )

    async def send(self, channel: str, message: OutboundMessage):
        blocks = [{"type": "section", "text": {"type": "mrkdwn", "text": message.text}}]

        # Slack supports rich blocks natively
        for element in message.rich_elements:
            if element.get("type") == "card":
                blocks.append({
                    "type": "section",
                    "text": {"type": "mrkdwn", "text": f"*{element['title']}*\n{element['body']}"},
                    "accessory": {
                        "type": "image",
                        "image_url": element.get("image_url", ""),
                        "alt_text": element["title"],
                    } if element.get("image_url") else None,
                })

        if message.quick_replies:
            blocks.append({
                "type": "actions",
                "elements": [
                    {"type": "button", "text": {"type": "plain_text", "text": r["label"]},
                     "action_id": r["value"], "value": r["value"]}
                    for r in message.quick_replies[:5]  # Slack limit per actions block
                ],
            })

        async with httpx.AsyncClient() as client:
            await client.post(
                "https://slack.com/api/chat.postMessage",
                json={"channel": channel, "blocks": blocks},
                headers={"Authorization": f"Bearer {self.bot_token}"},
            )

The Router

A central router dispatches normalized messages to the agent and routes responses back through the correct adapter:

class ChannelRouter:
    def __init__(self, agent_processor):
        self.adapters: dict[Channel, object] = {}
        self.agent = agent_processor

    def register(self, channel: Channel, adapter):
        self.adapters[channel] = adapter

    async def handle_inbound(self, channel: Channel, raw_data: dict):
        adapter = self.adapters[channel]
        message = adapter.normalize(raw_data)
        if not message:
            return

        # Process through the single agent brain
        response: OutboundMessage = await self.agent.process(message)

        # Send back through the same channel
        await adapter.send(message.channel_user_id, response)

router = ChannelRouter(agent_processor=my_agent)
router.register(Channel.WHATSAPP, whatsapp_adapter)
router.register(Channel.SLACK, slack_adapter)

Channel-Aware Response Adaptation

The agent's core logic is channel-agnostic, but some responses need adaptation. A TypeScript utility handles this:

function adaptForChannel(message: OutboundMessage, channel: Channel): OutboundMessage {
  const adapted = { ...message };

  switch (channel) {
    case "sms":
      // SMS: text only, no rich elements, max 160 chars per segment
      adapted.rich_elements = [];
      adapted.quick_replies = [];
      if (adapted.text.length > 300) {
        adapted.text = adapted.text.substring(0, 297) + "...";
      }
      break;

    case "whatsapp":
      // WhatsApp: max 3 buttons, 20 char button labels
      adapted.quick_replies = adapted.quick_replies.slice(0, 3).map((r) => ({
        ...r,
        label: r.label.substring(0, 20),
      }));
      break;

    case "slack":
      // Slack: convert markdown links, support up to 5 buttons
      adapted.text = adapted.text.replace(
        /[(.+?)]((.+?))/g, "<$2|$1>"
      );
      adapted.quick_replies = adapted.quick_replies.slice(0, 5);
      break;
  }

  return adapted;
}

FAQ

How do I handle channels that support images versus those that do not?

Include images as optional attachments in your OutboundMessage. Each channel adapter decides how to handle them. Web renders inline images. WhatsApp sends them as media messages. Slack uses image blocks. SMS sends a shortened URL to the image. The agent does not need to know which channel it is talking to — the adapter handles the translation.

How do I maintain conversation state across channels for the same user?

Map channel-specific user IDs to a unified user profile. When a known user messages on WhatsApp, look up their phone number in your user table to find their unified ID. Load conversation history from all channels for that unified ID. This lets a user start a conversation on the web widget and continue it on WhatsApp without losing context.

What about channel-specific features like Slack slash commands or WhatsApp templates?

Handle these at the adapter level before normalization. Slack slash commands get parsed by the Slack adapter and converted into regular InboundMessages with metadata indicating the command. WhatsApp template messages are handled by the send method with a separate code path. The agent brain sees a standard message either way.

---

#MultiChannel #WhatsApp #Slack #SMS #ChannelAbstraction #AgenticAI #LearnAI #AIEngineering