Capstone: Building a Multi-Channel Chat Agent Platform (Web, Slack, WhatsApp)

The Multi-Channel Challenge

Most organizations interact with customers across multiple channels simultaneously. A user might start a conversation on your website, follow up via WhatsApp, and your team manages internal queries through Slack. Building a separate AI agent for each channel creates maintenance nightmares, inconsistent responses, and fragmented conversation histories.

This capstone builds a unified platform where a single agent backend serves all channels. The key architectural insight is the channel adapter pattern: each channel has a thin adapter that translates channel-specific message formats into a canonical internal format, passes it to the shared agent, and translates the response back.

Canonical Message Format

Define a universal message format that all channel adapters produce and consume.

# core/models.py
from pydantic import BaseModel
from typing import Optional
from enum import Enum

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

class InboundMessage(BaseModel):
    channel: Channel
    channel_user_id: str       # channel-specific user identifier
    channel_thread_id: str     # channel-specific thread/conversation ID
    text: str
    attachments: list[str] = []  # URLs to any attached files

class OutboundMessage(BaseModel):
    text: str
    channel: Channel
    channel_thread_id: str
    metadata: dict = {}  # channel-specific formatting hints

Channel Adapter Interface

Each adapter implements two methods: parse_inbound to convert a channel-specific webhook payload into an InboundMessage, and send_outbound to deliver an OutboundMessage back through the channel.

# adapters/base.py
from abc import ABC, abstractmethod

class ChannelAdapter(ABC):
    @abstractmethod
    async def parse_inbound(self, raw_payload: dict) -> InboundMessage:
        """Convert channel-specific payload to canonical format."""
        ...

    @abstractmethod
    async def send_outbound(self, message: OutboundMessage) -> None:
        """Send response back through the channel."""
        ...

Slack Adapter

# adapters/slack_adapter.py
from slack_sdk.web.async_client import AsyncWebClient

class SlackAdapter(ChannelAdapter):
    def __init__(self):
        self.client = AsyncWebClient(token=os.environ["SLACK_BOT_TOKEN"])

    async def parse_inbound(self, raw_payload: dict) -> InboundMessage:
        event = raw_payload["event"]
        return InboundMessage(
            channel=Channel.SLACK,
            channel_user_id=event["user"],
            channel_thread_id=event.get("thread_ts", event["ts"]),
            text=event["text"],
        )

    async def send_outbound(self, message: OutboundMessage) -> None:
        await self.client.chat_postMessage(
            channel=os.environ["SLACK_CHANNEL_ID"],
            text=message.text,
            thread_ts=message.channel_thread_id,
        )

WhatsApp Adapter via Twilio

# adapters/whatsapp_adapter.py
from twilio.rest import Client

class WhatsAppAdapter(ChannelAdapter):
    def __init__(self):
        self.client = Client(
            os.environ["TWILIO_ACCOUNT_SID"],
            os.environ["TWILIO_AUTH_TOKEN"],
        )

    async def parse_inbound(self, raw_payload: dict) -> InboundMessage:
        return InboundMessage(
            channel=Channel.WHATSAPP,
            channel_user_id=raw_payload["From"],
            channel_thread_id=raw_payload["From"],  # WhatsApp uses phone as thread
            text=raw_payload["Body"],
        )

    async def send_outbound(self, message: OutboundMessage) -> None:
        self.client.messages.create(
            body=message.text,
            from_=f"whatsapp:{os.environ['TWILIO_WHATSAPP_NUMBER']}",
            to=message.channel_thread_id,
        )

Unified Agent Pipeline

The core pipeline receives a canonical InboundMessage, loads conversation history from the database, runs the agent, stores the response, and returns an OutboundMessage.

# core/pipeline.py
from agents import Agent, Runner

support_agent = Agent(
    name="Support Agent",
    instructions="You are a helpful support assistant. Be concise.",
    tools=[search_kb, create_ticket, check_order],
)

async def process_message(msg: InboundMessage, db) -> OutboundMessage:
    # Load or create conversation
    conv = await get_or_create_conversation(
        db, msg.channel, msg.channel_user_id, msg.channel_thread_id
    )

    # Build message history
    history = await get_message_history(db, conv.id, limit=20)

    # Store inbound message
    await store_message(db, conv.id, "user", msg.text)

    # Run agent
    result = await Runner.run(support_agent, msg.text, context={"history": history})

    # Store agent response
    await store_message(db, conv.id, "assistant", result.final_output)

    return OutboundMessage(
        text=result.final_output,
        channel=msg.channel,
        channel_thread_id=msg.channel_thread_id,
    )

Webhook Routes

Each channel has a dedicated webhook endpoint. All endpoints converge on the same process_message pipeline.

# routes/webhooks.py
from fastapi import APIRouter, Request

router = APIRouter()
adapters = {
    Channel.SLACK: SlackAdapter(),
    Channel.WHATSAPP: WhatsAppAdapter(),
    Channel.WEB: WebAdapter(),
}

@router.post("/webhooks/slack")
async def slack_webhook(request: Request, db=Depends(get_db)):
    payload = await request.json()
    if payload.get("type") == "url_verification":
        return {"challenge": payload["challenge"]}
    adapter = adapters[Channel.SLACK]
    inbound = await adapter.parse_inbound(payload)
    outbound = await process_message(inbound, db)
    await adapter.send_outbound(outbound)
    return {"ok": True}

@router.post("/webhooks/whatsapp")
async def whatsapp_webhook(request: Request, db=Depends(get_db)):
    form = await request.form()
    adapter = adapters[Channel.WHATSAPP]
    inbound = await adapter.parse_inbound(dict(form))
    outbound = await process_message(inbound, db)
    await adapter.send_outbound(outbound)
    return {"ok": True}

Testing Multi-Channel Behavior

Test each adapter independently by mocking the channel SDK and verifying the canonical format conversion. Test the pipeline with synthetic InboundMessage objects to verify agent behavior is identical regardless of channel.

# tests/test_slack_adapter.py
import pytest
from adapters.slack_adapter import SlackAdapter

@pytest.mark.asyncio
async def test_parse_slack_message():
    adapter = SlackAdapter()
    payload = {"event": {"user": "U123", "text": "hello", "ts": "111.222"}}
    msg = await adapter.parse_inbound(payload)
    assert msg.channel == Channel.SLACK
    assert msg.text == "hello"
    assert msg.channel_thread_id == "111.222"

FAQ

How do I handle rich formatting differences between channels?

Store formatting hints in the OutboundMessage.metadata field. The Slack adapter can convert markdown to Slack blocks, WhatsApp can use WhatsApp-specific formatting, and web can render full HTML. The agent always outputs plain text or markdown, and the adapter transforms it.

How do I track a single user across multiple channels?

Implement a user resolution layer that maps channel-specific user IDs to a unified user record. When a user verifies their email via the web widget and also uses WhatsApp, link both channel IDs to the same user record. This allows conversation history to persist across channels.

How do I handle channel-specific rate limits?

Implement per-adapter rate limiters. Slack has a 1 message per second limit per channel, WhatsApp has a 24-hour messaging window, and web has no external limits. Each adapter should queue messages and respect the channel rate limits independently.

---

#CapstoneProject #MultiChannel #Slack #WhatsApp #ChatAgent #FullStackAI #AgenticAI #LearnAI #AIEngineering