Build a Multi-Agent AI App That Shares Context Without Tool Calls (Python Tutorial)

In this tutorial, I'll show you how to build a multi-agent Python app where agents share live context without making tool calls to each other — using SignalMesh as an ambient context layer.

By the end you'll have:

• A running SignalMesh instance (local or HF Space)
• A 3-agent pipeline where agents broadcast and receive context
• A cost comparison showing what you saved

Live demo: https://kyklos.io | GitHub: https://github.com/Ig0tU/SignalMesh

---

Prerequisites

• Python 3.10+
• Basic familiarity with AI agents (LangChain, CrewAI, or raw Python)
• ~15 minutes

---

Step 1: Get SignalMesh Running

Option A — Use the public HF Space (no install):

HF_SPACE = "https://acecalisto3-signalmesh.hf.space"
# All endpoints available at this URL, no auth required

Option B — Self-host with Docker:

git clone https://github.com/Ig0tU/SignalMesh
cd SignalMesh
docker build -t signalmesh .
docker run -p 7860:7860 signalmesh
# Now available at http://localhost:7860

Option C — Import directly:

from signalmesh import signal_registry
# Runs in-process, no network overhead

---

Step 2: Build a 3-Agent Pipeline

We'll build: Researcher → Analyst → Writer, sharing context through the mesh.

The Researcher — broadcasts findings

import json
from signalmesh import signal_registry

def researcher_agent(topic: str) -> dict:
    # Imagine this calls a real API or search tool
    findings = {
        "topic": topic,
        "key_stats": ["99.97% cost reduction", "1.69µs latency"],
        "sources": ["benchmark_suite", "production_data"],
        "sentiment": "positive",
    }

    # Broadcast to the mesh — every agent can now read this
    signal_registry.broadcast(
        name=f"research_{topic.replace(' ', '_')}",
        source_type="context",
        data=findings,
    )

    print(f"Researcher: broadcast findings for '{topic}'")
    return findings

The Analyst — tunes in, adds analysis

def analyst_agent(topic: str) -> dict:
    # Read researcher findings from mesh — no tool call, no API hit
    research = signal_registry.tune_in([topic, "research"])

    if not research:
        return {"error": "No research found in mesh"}

    raw = research[0]["content"]

    # Add analysis layer
    analysis = {
        "summary": f"Based on {len(raw['sources'])} sources",
        "confidence": "high" if raw["sentiment"] == "positive" else "medium",
        "recommendation": "proceed",
        "key_stats": raw["key_stats"],
    }

    # Broadcast analysis back to mesh for the writer
    signal_registry.broadcast("analysis_output", "context", analysis)

    print(f"Analyst: read research, broadcast analysis")
    return analysis

The Writer — tunes in to both

def writer_agent() -> str:
    # Read both research AND analysis from mesh
    research = signal_registry.tune_in(["research"])
    analysis = signal_registry.tune_in(["analysis_output", "recommendation"])

    # Build the article from ambient context — no tool calls
    article = f"""
# {research[0]['content']['topic'].title()}

**Key finding:** {analysis[0]['content']['summary']}
**Confidence:** {analysis[0]['content']['confidence']}

Stats: {', '.join(research[0]['content']['key_stats'])}

Recommendation: {analysis[0]['content']['recommendation'].upper()}
    """.strip()

    print("Writer: built article from mesh context")
    return article

Run the pipeline

def run_pipeline(topic: str):
    print(f"\n=== Running pipeline for: {topic} ===\n")

    researcher_agent(topic)
    analyst_agent(topic)
    article = writer_agent()

    print(f"\n=== Final Article ===\n{article}")

run_pipeline("AI agent cost optimization")

Zero tool calls between agents. Zero redundant fetches. Each agent reads from what the previous one already put in the mesh.

---

Step 3: Verify with the Live API

Check what's in the mesh after your pipeline runs:

curl https://acecalisto3-signalmesh.hf.space/ui/frequencies
# Shows all active frequencies + signal counts

curl -X POST https://acecalisto3-signalmesh.hf.space/api/tune_in \
  -H "Content-Type: application/json" \
  -d '{"keywords":["research","analysis"]}'
# Returns all matching signals

---

Step 4: Measure the Cost Difference

import time

# Time a traditional tool call (simulated)
def mock_tool_call():
    time.sleep(0.0008)  # 800ms simulated network call
    return {"data": "context"}

# Time a mesh tune_in
def mesh_read():
    return signal_registry.tune_in(["research"])

# Benchmark
n = 1000
t0 = time.perf_counter()
for _ in range(n): mock_tool_call()
tool_time = (time.perf_counter() - t0) / n * 1e6

t0 = time.perf_counter()
for _ in range(n): mesh_read()
mesh_time = (time.perf_counter() - t0) / n * 1e6

print(f"Tool call: {tool_time:,.0f}µs avg")
print(f"Mesh read: {mesh_time:.2f}µs avg")
print(f"Speedup: {tool_time/mesh_time:,.0f}×")

---

What You Built

A 3-agent pipeline where:

• Context flows through the mesh, not through tool calls
• Each agent can read any other agent's output without knowing it exists
• Adding a 4th or 5th agent costs $0 in additional context read overhead

---

FAQ

Can agents write to frequencies they don't own?
Yes — any agent can broadcast to any frequency. Use naming conventions (agent_name/output) to avoid collisions.

What if I need ordered message delivery?
SignalMesh is not a message queue — use Kafka or RabbitMQ for ordering guarantees. SignalMesh is for ambient context where "latest state" is what matters.

How do I clear a frequency?
The buffer auto-manages (last 100 signals). For explicit clearing, restart the registry or add a clear_frequency() call to your pipeline teardown.

---

• Full working code + live demo: https://kyklos.io
• HF Space: https://acecalisto3-signalmesh.hf.space
• GitHub: https://github.com/Ig0tU/SignalMesh