---
title: "Capstone: Building a Multi-Channel Chat Agent Platform (Web, Slack, WhatsApp)"
description: "Build a unified AI agent backend that serves conversations across web chat, Slack, and WhatsApp using a channel abstraction layer, shared agent logic, and centralized conversation storage."
canonical: https://callsphere.ai/blog/capstone-multi-channel-chat-agent-platform
category: "Learn Agentic AI"
tags: ["Capstone Project", "Multi-Channel", "Slack", "WhatsApp", "Chat Agent", "Full-Stack AI"]
author: "CallSphere Team"
published: 2026-03-17T00:00:00.000Z
updated: 2026-05-06T01:02:44.794Z
---

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

> Build a unified AI agent backend that serves conversations across web chat, Slack, and WhatsApp using a channel abstraction layer, shared agent logic, and centralized conversation storage.

## 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.

```mermaid
flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus
classify"]
    PLAN["Plan and tool
selection"]
    AGENT["Agent loop
LLM plus tools"]
    GUARD{"Guardrails
and policy"}
    EXEC["Execute and
verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus
next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff
```

```python
# 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.

```python
# 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

```python
# 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

```python
# 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`.

```python
# 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.

```python
# 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.

```python
# 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

---

Source: https://callsphere.ai/blog/capstone-multi-channel-chat-agent-platform