Step 09: Channels - Multi-Platform Support

Extend the agent to support multiple messaging platforms (CLI, Telegram, Discord) through a unified channel abstraction.

What We Will Build

┌────────────────────────────────────────────────────────────────────┐
│                            Server                                   │
│                                                                    │
│  ┌─────────────┐  ┌──────────────┐  ┌──────────────┐             │
│  │EventBus     │  │AgentWorker   │  │DeliveryWorker│             │
│  │             │  │              │  │              │             │
│  └─────────────┘  └──────────────┘  └──────────────┘             │
│         ▲                  ▲                  │                    │
│         │                  │                  │                    │
│         │            ┌─────┴─────┐            │                    │
│         │            │ Agent     │            │                    │
│         │            │ Session   │            │                    │
│         │            └───────────┘            │                    │
│         │                                     │                    │
│  ┌──────┴─────────────────────────────────────┴──────┐            │
│  │              ChannelWorker                        │            │
│  │                                                   │            │
│  │  ┌─────────┐  ┌─────────┐  ┌─────────┐          │            │
│  │  │CLI      │  │Telegram │  │Discord  │          │            │
│  │  │Channel  │  │Channel  │  │Channel  │          │            │
│  │  └─────────┘  └─────────┘  └─────────┘          │            │
│  └──────────────────────────────────────────────────┘            │
│                                                                    │
│  ┌──────────────────────────────────────────────────────┐        │
│  │  RoutingTable                                        │        │
│  │  - Maps sources to agents                            │        │
│  │  - Manages session affinity                          │        │
│  └──────────────────────────────────────────────────────┘        │
└────────────────────────────────────────────────────────────────────┘

Key Components:

EventSource - Abstract base for platform-specific event sources (CLI, Telegram, Discord)
Channel - Abstract base for messaging platforms with run/reply/stop interface
ChannelWorker - Manages multiple channels and publishes InboundEvents
DeliveryWorker - Subscribes to OutboundEvents and delivers via appropriate channel
RoutingTable - Maps sources to sessions and agents
Server - Orchestrates all workers (EventBus, AgentWorker, ChannelWorker, DeliveryWorker)

Key Changes

1. EventSource and Platform Sources (src/mybot/core/events.py)

class EventSource(ABC):
    """Abstract base for all event sources."""

    _registry: ClassVar[dict[str, type["EventSource"]]] = {}
    _namespace: ClassVar[str] = ""

    @property
    def platform_name(self) -> str | None:
        if not self.is_platform:
            return None
        return self._namespace.split("-", 1)[1]

    @classmethod
    def from_string(cls, s: str) -> "EventSource":
        """Parse string to EventSource using namespace registry."""
        namespace = s.split(":")[0]
        source_cls = cls._registry.get(namespace)
        return source_cls.from_string(s)


@dataclass
class CliEventSource(EventSource):
    """Source for CLI-originated events."""
    _namespace = "platform-cli"

    def __str__(self) -> str:
        return "platform-cli:cli-user"

2. Channel Base Class (src/mybot/channel/base.py)

class Channel(ABC, Generic[T]):
    """Abstract base for messaging platforms."""

    @property
    @abstractmethod
    def platform_name(self) -> str:
        pass

    @abstractmethod
    async def run(self, on_message: Callable[[str, T], Awaitable[None]]) -> None:
        """Run the channel. Blocks until stop() is called."""
        pass

    @abstractmethod
    async def reply(self, content: str, source: T) -> None:
        """Reply to incoming message."""
        pass

    @abstractmethod
    async def stop(self) -> None:
        """Stop listening and cleanup resources."""
        pass

3. ChannelWorker (src/mybot/server/channel_worker.py)

class ChannelWorker(Worker):
    """Ingests messages from platforms, publishes INBOUND events."""

    async def run(self) -> None:
        """Start all channels and process incoming messages."""
        channel_tasks = [
            channel.run(self._create_callback(channel.platform_name))
            for channel in self.channels
        ]
        await asyncio.gather(*channel_tasks)

    def _create_callback(self, platform: str):
        async def callback(message: str, source: EventSource) -> None:
            session_id = self.context.routing_table.get_or_create_session_id(source)

            event = InboundEvent(
                session_id=session_id,
                source=source,
                content=message,
            )
            await self.context.eventbus.publish(event)

        return callback

4. DeliveryWorker (src/mybot/server/delivery_worker.py)

class DeliveryWorker(SubscriberWorker):
    """Delivers outbound messages to platforms."""

    async def handle_event(self, event: OutboundEvent) -> None:
        """Handle an outbound message event."""
        session_info = self._get_session_source(event.session_id)
        source = self._get_delivery_source(session_info)

        if source and source.platform_name:
            channel = self._get_channel(source.platform_name)
            if channel:
                await channel.reply(event.content, source)

        self.context.eventbus.ack(event)

5. RoutingTable (src/mybot/core/routing.py)

@dataclass
class RoutingTable:
    """Routes sources to agents using regex bindings."""

    def get_or_create_session_id(self, source: EventSource) -> str:
        """Get existing or create new session_id for source."""
        source_str = str(source)

        # Check for existing session (affinity)
        source_session = self._context.config.sources.get(source_str)
        if source_session:
            return source_session.session_id

        # Create new session
        agent_id = self.resolve(source_str)
        agent_def = self._context.agent_loader.load(agent_id)
        agent = Agent(agent_def, self._context)
        session = agent.new_session(source)

        # Cache session
        self._context.config.set_runtime(
            f"sources.{source_str}", SourceSessionConfig(session_id=session.session_id)
        )

        return session.session_id

5. Server Updates (src/mybot/server/server.py)

class Server:
    """Orchestrates workers with queue-based communication."""

    def _setup_workers(self) -> None:
        # Create WebSocketWorker first and attach to context
        ws_worker = WebSocketWorker(self.context)
        self.context.websocket_worker = ws_worker

        self.workers = [
            self.context.eventbus,
            AgentWorker(self.context),
            DeliveryWorker(self.context),
            ws_worker,  # WebSocketWorker added to workers
        ]

        if self.context.config.channels.enabled:
            self.workers.append(ChannelWorker(self.context))

    async def run(self) -> None:
        self._setup_workers()
        self._start_workers()

        # Start API server if configured
        if self.context.config.server:
            self._api_task = asyncio.create_task(self._run_api())

        await self._monitor_workers()

    async def _run_api(self) -> None:
        """Run the WebSocket API server."""
        app = create_app(self.context)
        config = uvicorn.Config(
            app,
            host=self.context.config.server.host,
            port=self.context.config.server.port,
        )
        server = uvicorn.Server(config)
        await server.serve()

async def run(self) -> None:
    self._setup_workers()
    self._start_workers()
    await self._monitor_workers()


## How to Run

**Start server with WebSocket:**
```bash
cd 10-websocket
uv run my-bot server

The server will start on http://0.0.0.0:8000 (configurable in config.user.yaml):

Web UI: http://localhost:8000/
WebSocket: ws://localhost:8000/ws

Chat via WebSocket client:

const ws = new WebSocket('ws://localhost:8000/ws');

// Send message
ws.send(JSON.stringify({
  source: "user-123",
  content: "Hello, agent!"
}));

// Receive events
ws.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data.type, data.content);
};

Or use the built-in web UI by opening http://localhost:8000/ in your browser.

Architecture Notes

Event Flow:

User sends message via platform (CLI, Telegram, Discord)
Channel receives message and creates EventSource
ChannelWorker creates/picks session via RoutingTable
ChannelWorker publishes InboundEvent to EventBus
AgentWorker processes event and generates response
AgentWorker publishes OutboundEvent to EventBus
DeliveryWorker receives OutboundEvent
DeliveryWorker looks up session's source and sends via appropriate channel

Session Affinity:

Each EventSource (e.g., "platform-telegram:123:456") maps to one session
First message creates session, subsequent messages reuse it
Session ID cached in config.runtime.yaml
Enables persistent conversations across restarts

Multi-Agent Routing:

RoutingTable matches sources to agents via regex patterns
Enables different agents for different platforms/users
Falls back to default_agent if no match

What's Next

Step 10 will add WebSocket UI - real-time web interface for interacting with agents.

class DeliveryWorker(SubscriberWorker):
    """Delivers outbound messages to platforms."""

    async def handle_event(self, event: OutboundEvent) -> None:
        """Handle an outbound message event."""
        session_info = self._get_session_source(event.session_id)
        source = self._get_delivery_source(session_info)

        if source and source.platform_name:
            channel = self._get_channel(source.platform_name)
            if channel:
                await channel.reply(event.content, source)

        self.context.eventbus.ack(event)

12 KiB Raw Blame History