The Sixth Sense

1 Overview

Bedsheet's Supervisor pattern works great when all agents run in a single Python process. But what if your agents need to run on different machines, behind different firewalls, or even on different cloud providers?

The Sixth Sense module adds distributed communication to any Bedsheet agent via PubNub's real-time messaging platform. Agents become network-aware peers that can broadcast alerts, send requests to each other, and coordinate incident responses — all without running HTTP servers or managing service discovery.

2 Setup

Install

pip install bedsheet[sense]

This adds the pubnub package as a dependency.

Get PubNub Keys

1. Sign up at pubnub.com (free tier: 200 MAU, unlimited channels)
2. Create an app in the PubNub dashboard
3. Copy your Subscribe Key and Publish Key

export PUBNUB_SUBSCRIBE_KEY=sub-c-your-key-here
export PUBNUB_PUBLISH_KEY=pub-c-your-key-here

3 Your First Sense Agent

Any Bedsheet Agent becomes network-aware by adding the SenseMixin:

from bedsheet import Agent, ActionGroup, SenseMixin
from bedsheet.llm.anthropic import AnthropicClient
from bedsheet.sense.pubnub_transport import PubNubTransport

# 1. Create a sense-aware agent class
class MyAgent(SenseMixin, Agent):
    pass

# 2. Build the agent as usual
agent = MyAgent(
    name="my-agent",
    instruction="You are a helpful agent.",
    model_client=AnthropicClient(),
)

# 3. Connect to the network
transport = PubNubTransport(
    subscribe_key="sub-c-...",
    publish_key="pub-c-...",
)
await agent.join_network(transport, "my-namespace", ["alerts", "tasks"])

# 4. The agent can now send and receive signals!
# ... do work ...

# 5. Disconnect when done
await agent.leave_network()

4 Signals

A Signal is the unit of inter-agent communication. Every signal has a kind, a sender, and an optional payload:

from bedsheet.sense import Signal

# Broadcast an alert
alert = Signal(
    kind="alert",
    sender="cpu-watcher",
    payload={"severity": "high", "cpu": 95.2},
)
await agent.broadcast("alerts", alert)

Signal Kinds

Channel Naming

Channels follow the convention bedsheet.<namespace>.<purpose>:

bedsheet.cloud-ops.alerts       # Alert broadcasts
bedsheet.cloud-ops.tasks        # Task coordination / claims
bedsheet.cloud-ops.cpu-watcher  # Direct channel to cpu-watcher agent

Custom Signal Handlers

@agent.on_signal("alert")
async def handle_alert(signal: Signal):
    severity = signal.payload.get("severity")
    print(f"Alert from {signal.sender}: {severity}")

5 Request / Response

The request() method sends a task to a remote agent and waits for the response. Under the hood, the receiving agent runs invoke() with the task and sends back the completion:

# Commander asks cpu-watcher to check usage
result = await commander.request(
    "cpu-watcher",
    "What is the current CPU usage?",
    timeout=30.0,
)
print(result)  # "Overall: 45.2%, Per-core: [32.1, 58.3, ...]"

6 Claim Protocol

When multiple commander agents are online, they need to coordinate who handles an incident. The claim protocol provides leaderless conflict resolution:

# When an alert arrives
won = await agent.claim_incident("incident-001", "tasks")
if won:
    # We are responsible for this incident
    await investigate_and_report()
else:
    # Another agent claimed it first
    pass

7 SenseNetwork API

For managing multiple agents in the same process (useful for testing and simpler deployments), use SenseNetwork:

from bedsheet.sense import SenseNetwork
from bedsheet.testing import MockSenseTransport

# For testing (in-memory)
network = SenseNetwork(
    namespace="cloud-ops",
    transport=MockSenseTransport(),
)

# For production (PubNub)
network = SenseNetwork(
    namespace="cloud-ops",
    transport_factory=lambda: PubNubTransport(sub_key, pub_key),
)

# Add agents
await network.add(cpu_agent, channels=["alerts", "tasks"])
await network.add(commander, channels=["alerts", "tasks"])

# Later...
await network.stop()

8 Cloud Monitor Demo

The examples/cloud-monitor/ directory contains a complete demo with 5 agents running as separate processes:

Running the Demo

export PUBNUB_SUBSCRIBE_KEY=sub-c-...
export PUBNUB_PUBLISH_KEY=pub-c-...
export ANTHROPIC_API_KEY=sk-ant-...

pip install bedsheet[sense] psutil

cd examples/cloud-monitor
python run.py

When CPU exceeds 80%, the cpu-watcher broadcasts an alert. The incident-commander claims it, queries the other agents for context, and synthesizes an incident report.

Agent Sentinel Demo

The examples/agent-sentinel/ directory demonstrates AI agent security monitoring, inspired by the OpenClaw crisis of 2026. Three worker agents perform real tasks (DuckDuckGo searches, JSON calendar management, skill installation with SHA-256 verification), while sentinel agents monitor them for rogue behavior:

Each worker has a ~15% chance per cycle of going rogue. Sentinels detect anomalies through real file I/O and alert the commander over PubNub, who investigates and issues quarantine orders.

pip install bedsheet[sense] duckduckgo-search
cd examples/agent-sentinel
python run.py

9 Testing

Use MockSenseTransport for unit tests — it routes signals in-memory without PubNub:

from bedsheet.testing import MockSenseTransport

# Create a shared hub for multiple agents
transport = MockSenseTransport()

# Each agent gets its own peer transport
t1 = transport          # First agent
t2 = transport.create_peer()  # Second agent (shares the hub)

# Agents connected to the same hub can exchange signals
await agent1.join_network(t1, "test", ["alerts"])
await agent2.join_network(t2, "test", ["alerts"])

result = await agent1.request("agent2", "do something")
assert result == "done"

10 Reference

Signal Fields

SenseMixin Methods

New Event Types