Building a Multi-Agent System with CloudSignal: A Practical Guide

In our previous post, we explored why multi-agent systems fail and how pub/sub messaging solves the communication problem. Now let’s build one.

We’ll create a three-agent pipeline:

1. Research Agent: Takes a topic and gathers information
2. Summary Agent: Receives research and creates a concise summary
3. Notifier Agent: Sends the summary to Slack (or any webhook)

Each agent runs independently, communicates via MQTT topics, and handles offline scenarios gracefully.

Prerequisites

• Python 3.10+
• A CloudSignal account (sign up free)
• OpenAI API key (for the AI agents)

Step 1: Set Up CloudSignal

After signing up, grab your MQTT credentials from the CloudSignal dashboard:

• Host: mqtt.cloudsignal.app
• Port: 8883 (TLS)
• Username: Your CloudSignal username
• Password: Your CloudSignal password

Step 2: Install Dependencies

pip install paho-mqtt openai python-dotenv

Create a .env file:

CLOUDSIGNAL_HOST=mqtt.cloudsignal.app
CLOUDSIGNAL_PORT=8883
CLOUDSIGNAL_USERNAME=your-username
CLOUDSIGNAL_PASSWORD=your-password
OPENAI_API_KEY=sk-...
SLACK_WEBHOOK_URL=https://hooks.slack.com/...

Step 3: Create the Base Agent Class

Let’s create a reusable base class that handles MQTT connection and message handling:

# base_agent.py
import os
import json
import ssl
import paho.mqtt.client as mqtt
from dotenv import load_dotenv

load_dotenv()

class BaseAgent:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.client = mqtt.Client(
            client_id=f"agent-{agent_id}",
            protocol=mqtt.MQTTv5
        )
        
        # Set up TLS
        self.client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
        
        # Set credentials
        self.client.username_pw_set(
            os.getenv("CLOUDSIGNAL_USERNAME"),
            os.getenv("CLOUDSIGNAL_PASSWORD")
        )
        
        # Set callbacks
        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message
        self.client.on_disconnect = self._on_disconnect
        
        self.subscriptions = []
    
    def _on_connect(self, client, userdata, flags, rc, properties=None):
        print(f"[{self.agent_id}] Connected with result code {rc}")
        # Resubscribe on reconnect
        for topic in self.subscriptions:
            self.client.subscribe(topic, qos=1)
            print(f"[{self.agent_id}] Subscribed to {topic}")
    
    def _on_message(self, client, userdata, msg):
        try:
            payload = json.loads(msg.payload.decode())
            self.handle_message(msg.topic, payload)
        except Exception as e:
            print(f"[{self.agent_id}] Error handling message: {e}")
    
    def _on_disconnect(self, client, userdata, rc, properties=None):
        print(f"[{self.agent_id}] Disconnected. Will auto-reconnect.")
    
    def subscribe(self, topic: str):
        self.subscriptions.append(topic)
        if self.client.is_connected():
            self.client.subscribe(topic, qos=1)
    
    def publish(self, topic: str, payload: dict):
        self.client.publish(
            topic,
            json.dumps(payload),
            qos=1  # At least once delivery
        )
        print(f"[{self.agent_id}] Published to {topic}")
    
    def handle_message(self, topic: str, payload: dict):
        """Override this in subclasses"""
        pass
    
    def start(self):
        self.client.connect(
            os.getenv("CLOUDSIGNAL_HOST"),
            int(os.getenv("CLOUDSIGNAL_PORT")),
            keepalive=60
        )
        self.client.loop_forever()

Step 4: Build the Research Agent

The research agent listens for research requests and publishes findings:

# research_agent.py
from openai import OpenAI
from base_agent import BaseAgent

class ResearchAgent(BaseAgent):
    def __init__(self):
        super().__init__("research")
        self.openai = OpenAI()
        self.subscribe("tasks/research")
    
    def handle_message(self, topic: str, payload: dict):
        task_id = payload.get("task_id")
        query = payload.get("query")
        
        print(f"[research] Researching: {query}")
        
        # Use OpenAI to "research" (in production, you'd use web search)
        response = self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a research assistant. Provide detailed, factual information."},
                {"role": "user", "content": f"Research the following topic and provide key findings: {query}"}
            ]
        )
        
        findings = response.choices[0].message.content
        
        # Publish results
        self.publish("results/research", {
            "task_id": task_id,
            "query": query,
            "findings": findings
        })

if __name__ == "__main__":
    agent = ResearchAgent()
    agent.start()

Step 5: Build the Summary Agent

The summary agent receives research and creates concise summaries:

# summary_agent.py
from openai import OpenAI
from base_agent import BaseAgent

class SummaryAgent(BaseAgent):
    def __init__(self):
        super().__init__("summary")
        self.openai = OpenAI()
        self.subscribe("results/research")
    
    def handle_message(self, topic: str, payload: dict):
        task_id = payload.get("task_id")
        query = payload.get("query")
        findings = payload.get("findings")
        
        print(f"[summary] Summarizing research for: {query}")
        
        response = self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "Create concise, actionable summaries. Use bullet points."},
                {"role": "user", "content": f"Summarize these research findings in 3-5 bullet points:\n\n{findings}"}
            ]
        )
        
        summary = response.choices[0].message.content
        
        # Publish summary
        self.publish("results/summary", {
            "task_id": task_id,
            "query": query,
            "summary": summary
        })

if __name__ == "__main__":
    agent = SummaryAgent()
    agent.start()

Step 6: Build the Notifier Agent

The notifier sends completed summaries to Slack:

# notifier_agent.py
import os
import requests
from base_agent import BaseAgent

class NotifierAgent(BaseAgent):
    def __init__(self):
        super().__init__("notifier")
        self.webhook_url = os.getenv("SLACK_WEBHOOK_URL")
        self.subscribe("results/summary")
    
    def handle_message(self, topic: str, payload: dict):
        query = payload.get("query")
        summary = payload.get("summary")
        
        print(f"[notifier] Sending notification for: {query}")
        
        # Send to Slack
        requests.post(self.webhook_url, json={
            "text": f"*Research Complete: {query}*\n\n{summary}"
        })
        
        print(f"[notifier] Notification sent!")

if __name__ == "__main__":
    agent = NotifierAgent()
    agent.start()

Step 7: Create the Trigger Script

Finally, a simple script to kick off research tasks:

# trigger.py
import uuid
from base_agent import BaseAgent

class TriggerAgent(BaseAgent):
    def __init__(self):
        super().__init__("trigger")
    
    def request_research(self, query: str):
        task_id = str(uuid.uuid4())[:8]
        self.publish("tasks/research", {
            "task_id": task_id,
            "query": query
        })
        print(f"[trigger] Requested research: {query} (task: {task_id})")

if __name__ == "__main__":
    import sys
    agent = TriggerAgent()
    
    # Connect briefly to publish
    agent.client.connect(
        os.getenv("CLOUDSIGNAL_HOST"),
        int(os.getenv("CLOUDSIGNAL_PORT"))
    )
    agent.client.loop_start()
    
    query = " ".join(sys.argv[1:]) or "Benefits of MQTT for AI agents"
    agent.request_research(query)
    
    import time
    time.sleep(1)  # Wait for publish
    agent.client.disconnect()

Running the System

Open four terminal windows:

# Terminal 1 - Research Agent
python research_agent.py

# Terminal 2 - Summary Agent  
python summary_agent.py

# Terminal 3 - Notifier Agent
python notifier_agent.py

# Terminal 4 - Trigger a task
python trigger.py "Latest developments in AI agent frameworks"

You’ll see:

1. The trigger publishes to tasks/research
2. Research agent picks it up, processes, publishes to results/research
3. Summary agent picks that up, summarizes, publishes to results/summary
4. Notifier agent sends the final summary to Slack

What Makes This Robust

Offline Handling

Stop the summary agent mid-task. The research results queue in CloudSignal. Restart the summary agent; it picks up right where it left off.

Easy Scaling

Need two research agents for faster processing? Just run another instance. MQTT load-balances automatically with shared subscriptions.

Loose Coupling

Add a new agent that also listens to results/summary? No changes needed to existing agents. Just subscribe to the topic.

Observable

Every message flows through CloudSignal. You can monitor, debug, and replay messages from the dashboard.

Next Steps

This is a basic example. In production, you’d add:

• Error handling topics: Publish failures to errors/{agent-id} for monitoring
• Task status tracking: Use retained messages on status/{task-id}
• Request-response patterns: For synchronous-style communication when needed
• Authentication per agent: Each agent gets its own credentials with topic-level ACLs

Conclusion

Building multi-agent systems doesn’t have to be complex. With pub/sub messaging:

• Agents are decoupled and independently deployable
• Messages are guaranteed to deliver
• Offline scenarios are handled automatically
• Scaling is straightforward

The infrastructure disappears. You focus on what your agents actually do.

---

Ready to build? Sign up for CloudSignal and get 50,000 messages/month free. No credit card required.