DEV Community

Mavos.by.Kyklos
Mavos.by.Kyklos

Posted on

How to Reduce AI Agent API Costs by 99% with Ambient Context (SignalMesh)

#ai #python #tutorial #webdev

TL;DR: Every agent in your fleet is making redundant API calls to read context that hasn't changed. SignalMesh replaces those calls with a broadcast-once, tune-in-many pattern — cutting context read costs by 99.97% and latency from 800ms to 1.69µs.

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

Why Multi-Agent AI Costs More Than It Should

If you've built a pipeline with LangChain, CrewAI, AutoGen, or raw Python agents, you've hit this pattern:

  1. Agent A needs to know the current system state → tool call → 800ms → tokens burned
  2. Agent B needs the same system state → another tool call → 800ms → more tokens
  3. Agent C, D, E — same thing

This isn't a framework problem. It's an architectural problem. Read-only context fetches are being treated as write operations — each agent independently verifying what it could just receive.

Here's what that costs at scale:

Setup Context fetches/session Latency cost Annual token cost
5 agents, 3 reads each 15 12,000ms $1,387
5 agents, SignalMesh 1 broadcast ~2ms $0.46

$1,387 → $0.46. Same agents. Same information.

What is SignalMesh?

SignalMesh is an open source ambient context protocol — a lightweight in-memory mesh where data sources broadcast signals onto named frequencies, and agents tune in to receive matching context without making tool calls.

Think of it like a radio tower for your agent fleet. One tower broadcasts. Every receiver picks it up instantly. Nobody has to call the tower individually.

It's MIT licensed, runs on Python 3.10+, and deploys in minutes via Docker or HuggingFace Spaces.

How to Use SignalMesh in Your Agent Pipeline

Step 1: Broadcast context from your data source 

from signalmesh import signal_registry

# Any source — API, database, RSS feed, another agent
signal_registry.broadcast(
    name="market_data",
    source_type="api",
    data={"asset": "BTC", "price": 42000, "volume": 1.2e9},
    metadata={"source": "coinbase", "timestamp": 1718000000}
)
Enter fullscreen mode Exit fullscreen mode

Step 2: Tune in from any agent 

# Agent A
context = signal_registry.tune_in(["market_data", "price"])

# Agent B — different keyword, same result
context = signal_registry.tune_in(["BTC", "market"])

# Agent C — partial match, still works
context = signal_registry.tune_in(["asset", "volume"])
Enter fullscreen mode Exit fullscreen mode

The mesh resolves keyword variants — partial names, token overlaps, edge-case spellings — so your agents find the right context even when naming isn't perfectly consistent across your codebase.

Step 3: Inject into system prompt 

system_prompt = f"""
You are a trading analyst.
Current market context: {json.dumps(context)}
"""
Enter fullscreen mode Exit fullscreen mode

No tool call. No round trip. No tokens spent fetching data.

LangChain Integration 

from langchain.tools import tool
from signalmesh import signal_registry

@tool
def get_market_context(query: str) -> str:
    """Get current market data from the ambient mesh."""
    results = signal_registry.tune_in([query])
    return json.dumps(results) if results else "No context found"

# In your agent
agent = create_react_agent(llm, tools=[get_market_context], ...)
Enter fullscreen mode Exit fullscreen mode

CrewAI Integration 

from crewai import Agent
from signalmesh import signal_registry

class MeshAwareAgent(Agent):
    def get_context(self, keywords: list) -> list:
        return signal_registry.tune_in(keywords)

analyst = MeshAwareAgent(
    role="Market Analyst",
    goal="Analyze current market conditions",
    backstory="Expert analyst with real-time mesh access"
)
Enter fullscreen mode Exit fullscreen mode

Benchmark Results

We benchmarked tune_in() across payload sizes and concurrency levels:

Scenario Latency vs. 800ms tool call
Single agent, small payload 1.69 µs 473,000× faster
Single agent, 1MB payload ~10 µs 80,000× faster
10 concurrent agents ~138 µs median 5,800× faster
50 concurrent agents ~635 µs median 1,260× faster
100 concurrent agents ~1.25 ms median 640× faster

Key finding: payload size doesn't significantly affect latency because Python stores dict references, not copies.

Live API — Try It Right Now

The public SignalMesh mesh is running at https://acecalisto3-signalmesh.hf.space. No auth required, CORS open.

# See all active frequencies
curl https://acecalisto3-signalmesh.hf.space/ui/frequencies

# Broadcast a signal
curl -X POST https://acecalisto3-signalmesh.hf.space/api/broadcast \
  -H "Content-Type: application/json" \
  -d '{"name":"test_freq","source_type":"context","data":{"hello":"world"}}'

# Tune in
curl -X POST https://acecalisto3-signalmesh.hf.space/api/tune_in \
  -H "Content-Type: application/json" \
  -d '{"keywords":["test"]}'
Enter fullscreen mode Exit fullscreen mode

Deployment Options

Open Source Managed Cloud Enterprise
Price Free (MIT) $299/mo Custom
Nodes Unlimited (self-host) 500 Unlimited
SLA — 99.9% 99.99%
Private namespaces ✓ ✓ ✓
Support Community Email + Slack Dedicated engineer

Custom integration with your existing stack (LangGraph, AutoGen, CrewAI) available — flat-rate project, delivery in days. Contact: abra.autopreneur@gmail.com

Frequently Asked Questions

Does SignalMesh replace a vector database?
No — it's complementary. Vector DBs are for semantic search over large document corpora. SignalMesh is for low-latency ambient context that changes frequently (system state, live feeds, agent outputs). Use both.

What happens if an agent tunes in with a keyword that doesn't match any frequency?
The mesh scores the keyword against all live frequencies and bridges to the nearest match if confidence is above threshold. It logs the gap and remembers the mapping for future calls. Silent failures are surfaced, not hidden.

Can multiple agents broadcast to the same frequency?
Yes. Each frequency maintains a buffer of the last 100 signals from any source. Useful for aggregating outputs from parallel agents.

Is it thread-safe for concurrent agents?
Yes. The registry uses Python's GIL-protected dict for reads. At 100 concurrent agents, median per-agent latency is ~1.25ms — still far faster than any network call.

How do I self-host?

git clone https://github.com/Ig0tU/SignalMesh
docker build -t signalmesh .
docker run -p 7860:7860 signalmesh
Enter fullscreen mode Exit fullscreen mode

Resources

Top comments (0)