# Does Shadowbroker Use a Message Queue or Event Bus in Its Backend Architecture?

> Discover how Shadowbroker manages agent communication with its internal event bus, leveraging asyncio.Queue and deques instead of external message brokers like Kafka.

- Repository: [Shadowbroker/Shadowbroker](https://github.com/BigBodyCobain/Shadowbroker)
- Tags: architecture
- Published: 2026-05-07

---

**Shadowbroker does not employ an external message broker such as RabbitMQ or Kafka; instead, it implements an internal event bus using lightweight, in-process queues including `asyncio.Queue`, `collections.deque`, and thread-protected Python lists to manage real-time agent communication and mesh networking.**

Shadowbroker is an open-source operational intelligence platform designed for high-agency agent coordination and decentralized mesh operations. According to the source code in the `BigBodyCobain/Shadowbroker` repository, the backend architecture deliberately avoids the operational overhead of standalone message queues by utilizing native Python concurrency primitives. These in-process queues function as an **event bus** within the application, handling Server-Sent Events (SSE), WebSocket notifications, and mesh packet batching without requiring separate infrastructure.

## SSE Queue Implementation in [`ai_intel.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/ai_intel.py)

The primary mechanism for pushing real-time updates to HTTP clients relies on per-client `asyncio.Queue` instances. In [`backend/routers/ai_intel.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/routers/ai_intel.py) at line 2863, the system maintains a global list of active queues:

```python

# backend/routers/ai_intel.py – line 2863

_sse_queues: list[asyncio.Queue] = []

```

When an agent connects to the `/api/ai/channel/sse` endpoint, the system instantiates a new `asyncio.Queue(maxsize=512)` and registers it in `_sse_queues` (line 2904). The `broadcast_to_sse_clients` function (lines 2868–2875) iterates over these queues to distribute events:

```python
async def broadcast_to_sse_clients(event_type: str, data: dict[str, Any]):
    async with _sse_queues_lock:
        queues = list(_sse_queues)
    for q in queues:
        try:
            q.put_nowait({"event": event_type, "data": data})
        except asyncio.QueueFull:
            pass   # slow client – drop the message

```

If a client’s queue reaches capacity, the system applies **backpressure handling** by dropping the message rather than blocking the broadcaster, ensuring that slow consumers do not degrade system performance.

## WebSocket Channel Management

For bidirectional real-time communication, Shadowbroker uses a similar pattern with WebSocket connections. Located in the same file at lines 3031–3033, the implementation stores active connections in a protected list:

```python

# backend/routers/ai_intel.py – lines 3031‑3033

_ws_clients: list[WebSocket] = []
_ws_clients_lock = asyncio.Lock()

```

When the server generates a new task or alert, it iterates over `_ws_clients` and transmits the payload directly via `await ws.send_json(payload)`. This design eliminates the need for an external pub/sub broker while maintaining low-latency push delivery to connected agents.

## Mesh RNS Batch Queues

The mesh networking layer handles high-latency, covert communication channels using thread-protected lists rather than async queues. In [`backend/services/mesh/mesh_rns.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/mesh/mesh_rns.py) at lines 148–149, the `MeshRNS` class maintains two distinct batch buffers:

```python

# backend/services/mesh/mesh_rns.py – line 148‑149

self._batch_queue: list[dict] = []
self._gate_batch_queue: list[tuple[str, dict]] = []

```

These lists aggregate outbound packets for **cover-traffic timing** and bandwidth optimization. A background timer (`_batch_timer`) periodically flushes accumulated items (line 1415–1417), ensuring that transmissions adhere to predefined timing intervals to obfuscate traffic patterns:

```python
queued = list(self._batch_queue)
self._batch_queue.clear()

# … send `queued` to the network …

```

This batching strategy is thread-safe through the use of `_batch_lock`, allowing concurrent access from multiple mesh threads without requiring an external message queue.

## Agent Actions Queue

For frontend interaction, the system uses a bounded `collections.deque` to buffer UI actions generated by AI agents. Defined at line 38 in [`backend/routers/ai_intel.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/routers/ai_intel.py), this queue stores actions with a maximum length of 20:

```python
from collections import deque

# backend/routers/ai_intel.py – line 38

_agent_actions: deque[dict] = deque(maxlen=20)

```

The frontend polls the `/api/ai/agent_actions` endpoint to consume these actions, creating a simple request-response event mechanism that decouples AI-generated commands from the UI rendering cycle without persistent websocket overhead for short-lived interactions.

## Practical Implementation Examples

### Broadcasting an SSE Event

To push a task update to all connected agents from anywhere in the backend:

```python
from backend.routers.ai_intel import broadcast_to_sse_clients

await broadcast_to_sse_clients(
    event_type="task",
    data={
        "task_type": "fetch_satellite",
        "payload": {"lat": 40.7, "lon": -74.0}
    }
)

```

### Queueing a Mesh Packet

To add data to the mesh batch queue for covert transmission:

```python
from backend.services.mesh.mesh_rns import MeshRNS

mesh = MeshRNS()
packet = {"type": "dm", "dest": "gate123", "payload": {"msg": "Hello"}}
mesh._batch_queue.append(packet)  # thread-safe via _batch_lock

```

### Pushing a UI Action

To record an action for frontend consumption:

```python
from backend.routers.ai_intel import push_agent_action

push_agent_action({
    "type": "display_image",
    "url": "https://example.com/sat.png",
    "caption": "Recent satellite view"
})

```

## Summary

- **No external brokers**: Shadowbroker avoids RabbitMQ, Kafka, or Redis by using in-process Python queues.
- **Asyncio primitives**: Real-time channels rely on `asyncio.Queue` with explicit size limits (max 512) to handle backpressure.
- **Thread-safe batching**: The mesh layer uses locked Python lists (`_batch_queue`) to aggregate packets for timing-based transmission.
- **Bounded buffering**: The agent actions system uses `collections.deque` with `maxlen=20` to prevent memory growth from untended UI events.
- **Single-process architecture**: All queue operations occur within the application process, simplifying deployment and reducing infrastructure dependencies.

## Frequently Asked Questions

### Does Shadowbroker use RabbitMQ or Kafka for its message queue?

No, Shadowbroker does not integrate external message brokers. The codebase implements an internal **event bus** using `asyncio.Queue` for asynchronous streams and thread-protected lists for mesh batching, keeping all message handling within the Python application process.

### How does Shadowbroker handle slow SSE clients?

The system implements **drop-on-overflow** backpressure. When `broadcast_to_sse_clients` encounters a full queue (`asyncio.QueueFull`), it silently drops the message for that specific client (line 2874). This prevents slow HTTP consumers from blocking the broadcaster and affecting other connected agents.

### What is the maximum capacity of the agent actions queue?

The agent actions queue is initialized with `maxlen=20` using `collections.deque` at line 38 in [`backend/routers/ai_intel.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/routers/ai_intel.py). When this limit is reached, new actions automatically evict the oldest entries, ensuring the memory footprint remains bounded regardless of AI generation frequency.

### Are the mesh batch queues thread-safe?

Yes, the `_batch_queue` and `_gate_batch_queue` in [`backend/services/mesh/mesh_rns.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/mesh/mesh_rns.py) are protected by `_batch_lock`. While they are standard Python lists rather than formal queue structures, all append and flush operations acquire this lock, making them safe for concurrent access from multiple mesh networking threads.