# Event Bus Architecture in Claude-Code-Telegram: Decoupled Message Routing Explained

> Discover the event bus architecture in Claude-Code-Telegram. Learn how this typed, asynchronous system decouples producers and consumers with publish-subscribe for efficient message routing.

- Repository: [Richard A/claude-code-telegram](https://github.com/richardatct/claude-code-telegram)
- Tags: architecture
- Published: 2026-02-20

---

**The Claude-Code-Telegram repository implements an asynchronous, typed event bus that decouples event producers from consumers through a publish-subscribe pattern with concurrent handler execution and error isolation.**

The event bus architecture in `RichardAtCT/claude-code-telegram` serves as the backbone for routing messages between Telegram updates, webhook endpoints, and agent execution logic. By leveraging Python's `asyncio` primitives and strict type hierarchies, the system enables clean separation of concerns while maintaining high throughput and fault tolerance.

## Core Components of the Event Bus Architecture

### Event Base Class and Typed Subclasses

All events in the system inherit from a base `Event` class defined in [`src/events/bus.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/bus.py). This base class provides every event with a unique UUID, UTC timestamp, and source identifier, ensuring traceability across the distributed pipeline.

```python

# From src/events/bus.py

@dataclass
class Event:
    id: UUID = field(default_factory=uuid4)
    timestamp: datetime = field(default_factory=datetime.utcnow)
    source: str = "unknown"

```

Concrete event types in [`src/events/types.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/types.py) extend this base to create strongly-typed payloads for specific domains:

- **`UserMessageEvent`** – Carries Telegram message data including `user_id`, `chat_id`, and `text`
- **`WebhookEvent`** – Encapsulates external webhook payloads with `provider` and `event_type_name` fields
- **`ScheduledEvent`** – Represents cron-triggered or timer-based automation
- **`AgentResponseEvent`** – Contains output from Claude API calls ready for delivery back to users

### The EventBus Central Hub

The `EventBus` class in [`src/events/bus.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/bus.py) acts as the central message broker. It maintains two internal registries: `_handlers` (a mapping of event types to handler lists) and `_global_handlers` (handlers receiving all events regardless of type).

```python

# Core registries in EventBus.__init__

self._handlers: Dict[Type[Event], List[EventHandler]] = {}
self._global_handlers: List[EventHandler] = []
self._queue: asyncio.Queue[Event] = asyncio.Queue()

```

## How the Event Bus Facilitates Decoupled Routing

### Publish-Subscribe Pattern Implementation

The event bus architecture enables decoupling through explicit subscription APIs. Handlers register for specific event types using `subscribe()`, or capture all events using `subscribe_all()`. This eliminates direct dependencies between message producers and consumers.

In [`src/events/bus.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/bus.py), the subscription logic appends handlers to the appropriate registry:

```python
def subscribe(self, event_type: Type[Event], handler: EventHandler) -> None:
    """Register a handler for a specific event type."""
    if event_type not in self._handlers:
        self._handlers[event_type] = []
    self._handlers[event_type].append(handler)
    logger.info(f"Handler subscribed to {event_type.__name__}")

```

Producers remain completely unaware of downstream handlers. They simply instantiate an event and call `publish()`, which enqueues the event into the internal `asyncio.Queue`:

```python
async def publish(self, event: Event) -> None:
    """Enqueue an event for processing."""
    await self._queue.put(event)
    logger.debug(f"Event {event.id} published")

```

### Concurrent Handler Execution

The event bus architecture maximizes throughput by executing all matching handlers concurrently. When the background `_process_events` loop retrieves an event from the queue, it invokes `_dispatch()`, which collects both type-specific and global handlers, then executes them using `asyncio.gather()`.

From [`src/events/bus.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/bus.py), the dispatch logic demonstrates this parallelism:

```python
async def _dispatch(self, event: Event) -> None:
    """Route event to all registered handlers concurrently."""
    handlers = []
    
    # Collect type-specific handlers (including subclass matches)

    for event_type, handler_list in self._handlers.items():
        if isinstance(event, event_type):
            handlers.extend(handler_list)
    
    # Add global handlers

    handlers.extend(self._global_handlers)
    
    # Execute all handlers concurrently

    if handlers:
        await asyncio.gather(*[self._safe_call(h, event) for h in handlers])

```

This design ensures that a slow handler—such as one performing a long-running Claude API call—cannot block other critical handlers like audit logging or metrics collection.

### Error Isolation and Resilience

Decoupled systems require fault tolerance. The event bus architecture isolates handler failures through the `_safe_call` wrapper, which catches exceptions, logs them via `structlog`, and prevents propagation to other handlers or the main event loop.

The error isolation implementation in [`src/events/bus.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/bus.py):

```python
async def _safe_call(self, handler: EventHandler, event: Event) -> None:
    """Execute handler with error isolation."""
    try:
        await handler(event)
    except Exception as e:
        logger.error(
            "Handler failed for event",
            event_id=str(event.id),
            handler=handler.__name__,
            error=str(e)
        )
        # Exception is not re-raised, preserving event loop stability

```

This resilience pattern ensures that a buggy handler cannot crash the entire messaging pipeline, maintaining system availability even under partial failure conditions.

## Practical Implementation Examples

### Subscribing Handlers to Event Types

The following example demonstrates registering specific handlers for Telegram messages and global handlers for monitoring:

```python
from src.events.bus import EventBus
from src.events.types import UserMessageEvent, WebhookEvent

bus = EventBus()

# Specific handler for user messages

async def process_message(event: UserMessageEvent) -> None:
    await claude_client.send_message(event.text, event.chat_id)

# Specific handler for webhooks

async def process_webhook(event: WebhookEvent) -> None:
    if event.provider == "github":
        await handle_github_event(event.payload)

# Global audit handler

async def audit_all_events(event) -> None:
    await database.log_event(event.id, event.source, event.timestamp)

# Register subscriptions

bus.subscribe(UserMessageEvent, process_message)
bus.subscribe(WebhookEvent, process_webhook)
bus.subscribe_all(audit_all_events)

# Start the bus

await bus.start()

```

### Publishing Events from Producers

Producers remain decoupled from handlers, publishing events without knowledge of downstream processing:

```python
from src.events.types import UserMessageEvent
from src.events.bus import bus  # Singleton instance

from pathlib import Path

async def on_telegram_update(update):
    """Called by python-telegram-bot when a message arrives."""
    msg = update.effective_message
    
    # Create typed event

    event = UserMessageEvent(
        user_id=msg.from_user.id,
        chat_id=msg.chat.id,
        text=msg.text or "",
        working_directory=Path.cwd()
    )
    
    # Publish without knowing which handlers will process it

    await bus.publish(event)

```

### Adding New Event Types Without Modifying the Bus

The architecture supports extension through inheritance without touching core routing logic:

```python
from src.events.bus import Event
from dataclasses import dataclass

# Define new event type

@dataclass
class ScheduledBackupEvent(Event):
    backup_type: str
    target_path: str

# Handler automatically eligible for dispatch

async def backup_handler(event: ScheduledBackupEvent):
    await perform_backup(event.target_path)

# Subscribe without modifying EventBus source code

bus.subscribe(ScheduledBackupEvent, backup_handler)

```

## Summary

The event bus architecture in `RichardAtCT/claude-code-telegram` provides a robust foundation for decoupled message routing through these key mechanisms:

- **Typed Event Hierarchy**: All events inherit from a base `Event` class in [`src/events/bus.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/bus.py), enabling type-safe routing and extensibility through subclasses defined in [`src/events/types.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/types.py).
- **Publish-Subscribe Pattern**: Producers call `publish()` without knowing consumers, while handlers register via `subscribe()` or `subscribe_all()`, eliminating direct dependencies.
- **Concurrent Dispatch**: The `_dispatch()` method executes all matching handlers using `asyncio.gather()`, ensuring high throughput and preventing slow operations from blocking the pipeline.
- **Fault Isolation**: The `_safe_call()` wrapper catches and logs exceptions without propagating them, keeping the event loop stable even when individual handlers fail.

This architecture allows the Telegram bot, webhook endpoints, and background schedulers to operate independently while maintaining clean, testable boundaries between components.

## Frequently Asked Questions

### How does the event bus handle different event types without hardcoding logic?

The event bus uses Python's `isinstance()` checks within the `_dispatch()` method to match events against registered handler types. When a handler subscribes to a specific event class, the bus stores that mapping in `_handlers`. During dispatch, it iterates through all registered types and checks if the incoming event is an instance of that type, including subclass matches. This allows new event types to be added in [`src/events/types.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/types.py) without modifying the core routing logic in [`src/events/bus.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/bus.py).

### What happens if one event handler crashes or raises an exception?

Individual handler failures are isolated through the `_safe_call()` method, which wraps each handler execution in a try-except block. If a handler raises an exception, the error is logged with the event ID and handler name via `structlog`, but the exception is not re-raised. This ensures that `asyncio.gather()` continues executing remaining handlers, and the background `_process_events()` loop remains stable to process subsequent events from the queue.

### Can multiple handlers process the same event simultaneously?

Yes, the event bus architecture supports concurrent execution of all matching handlers. When an event is dispatched, the `_dispatch()` method collects both type-specific and global handlers into a single list, then executes them concurrently using `asyncio.gather()`. This design ensures that independent operations—such as sending a Claude API request, writing to an audit log, and updating metrics—can happen in parallel rather than sequentially, maximizing throughput and preventing slow handlers from blocking faster ones.

### How do producers remain decoupled from consumers in this architecture?

Producers interact with the event bus exclusively through the `publish()` method, which accepts an `Event` instance and places it onto an internal `asyncio.Queue`. The producer has no knowledge of which handlers are registered, how many exist, or what business logic they implement. Conversely, handlers register interest in specific event types via `subscribe()` or `subscribe_all()`, receiving only the events they care about without knowing their origin. This publish-subscribe pattern, implemented in [`src/events/bus.py`](https://github.com/RichardAtCT/claude-code-telegram/blob/main/src/events/bus.py), creates a clean boundary that allows components to evolve independently.