The Protocol That Cleaned Up Our Agent Architecture

The Protocol That Cleaned Up Our Agent Architecture

Building multi-agent systems often feels like herding cats. Each agent has its own way of communicating, its own data format, and its own idea of what a "request" looks like. As our agent architecture grew from a handful of specialized bots to a sprawling ecosystem of 50+ autonomous workers, we hit a wall. The complexity of inter-agent communication was overwhelming our system. Debugging became a nightmare, latency spiked, and adding a new agent required days of integration work.

The solution came not from building a smarter agent, but from adopting a simple, standardized protocol. This article walks through the exact protocol that cleaned up our architecture, the step-by-step installation process, and practical usage examples that you can apply today.

The Problem: Agent Sprawl Without a Common Language

In early 2024, our team was running agents for data extraction, summarization, code review, customer support triage, and content generation. Each agent was built by a different team using different frameworks. Some used OpenAI’s function calling, others used Anthropic’s tool use, and a few used custom REST endpoints.

The result? A tangled web of point-to-point integrations. Agent A would send JSON to Agent B, but Agent B expected XML. Agent C would await a callback from Agent D, but the timeout was set incorrectly. Every new agent required a custom adapter. Our architecture was fragile, opaque, and slow.

We needed a universal language that every agent could speak. We needed a protocol.

The Protocol That Changed Everything

The protocol we adopted is built on three core principles:

1. **Uniform message envelope** – Every message, regardless of content, wraps in a standard header with metadata (sender, receiver, type, timestamp, trace ID). 2. **Typed payloads** – The content body uses a shared schema registry so that every agent knows exactly what fields to expect. 3. **Asynchronous routing** – Agents communicate via a lightweight message broker, not direct HTTP calls. This decouples senders from receivers and enables retries, queues, and load balancing.

This approach is inspired by patterns used in large-scale distributed systems, but adapted for the agentic world. The protocol we implemented is called **Agent Communication Protocol (ACP)**, an open standard that we built on top of existing reliable infrastructure.

Requirements

Before you begin, ensure your environment meets these requirements:

• **Python 3.10+** – The protocol implementation uses modern async features.
• **Redis 7+** – Used as the message broker for routing and queuing.
• **pip** – For installing dependencies.
• **At least one agent** – A simple script or service that you want to connect.
• **Basic familiarity with async Python** – We use `asyncio` and `aiohttp`.

Step-by-Step Installation

1. Install the protocol library

The core library is `acp-py`. Install it with pip:

pip install acp-py

This installs the message envelope, schema registry client, and the Redis-backed router.

2. Install and start Redis

If you don’t have Redis running, install it. On Ubuntu:

sudo apt update
sudo apt install redis-server
sudo systemctl start redis-server

Verify it’s running:

redis-cli ping
# Should return PONG

3. Configure the protocol settings

Create a configuration file `acp_config.yaml` in your project root:

broker:
  type: redis
  host: localhost
  port: 6379
  queue_prefix: "acp:queue:"

schema_registry:
  type: local
  path: "./schemas/"

agent:
  name: "my-agent"
  version: "1.0.0"
  capabilities:
    - text-processing
    - data-extraction

This tells the protocol where to find the broker, where to load schemas from, and what your agent can do.

4. Create a schema file

Define a simple message schema in `./schemas/extract_request.json`:

{
  "type": "object",
  "properties": {
    "text": {"type": "string"},
    "max_tokens": {"type": "integer", "default": 500}
  },
  "required": ["text"]
}

This schema ensures that every `extract_request` message contains a `text` field and optionally a `max_tokens` field.

5. Initialize the protocol in your agent

Create a Python script `agent.py`:

import asyncio
from acp import Agent, Message

async def handle_extract(msg: Message):
    """Handle an extract request."""
    text = msg.payload["text"]
    max_tokens = msg.payload.get("max_tokens", 500)
    result = f"Extracted {len(text.split())} words, limited to {max_tokens} tokens."
    
    # Send a reply using the same message trace
    reply = msg.reply(payload={"result": result, "status": "ok"})
    await agent.send(reply)

async def main():
    global agent
    agent = Agent.from_config("acp_config.yaml")
    agent.register_handler("extract_request", handle_extract)
    await agent.start()
    print("Agent listening on queue...")
    await asyncio.Event().wait()  # Run forever

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python agent.py

Your agent is now listening for messages on the Redis queue.

Usage Examples

Example 1: Sending a request from another agent

Create a second agent `requester.py` that sends a message to the first agent:

import asyncio
from acp import Agent, Message

async def main():
    requester = Agent.from_config("acp_config.yaml", name="requester-agent")
    await requester.start()
    
    # Build a valid message using the schema
    msg = Message(
        target="my-agent",
        type="extract_request",
        payload={"text": "Hello world, this is a test extraction.", "max_tokens": 100}
    )
    
    # Send and wait for reply
    reply = await requester.send_and_wait(msg, timeout=10.0)
    print(f"Got reply: {reply.payload}")

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python requester.py

You’ll see the reply printed: `Got reply: {'result': 'Extracted 7 words, limited to 100 tokens.', 'status': 'ok'}`

Example 2: Broadcasting a message to multiple agents

Sometimes you want all agents of a certain capability to receive a message. Use the `broadcast` method:

from acp import Agent, Message

async def broadcast_example():
    broadcaster = Agent.from_config("acp_config.yaml", name="broadcaster")
    await broadcaster.start()
    
    # Send to all agents with capability "text-processing"
    msg = Message(
        target="*",  # broadcast wildcard
        type="status_request",
        payload={"query": "health"}
    )
    # Broadcast returns a dictionary of agent_name -> reply
    replies = await broadcaster.broadcast(msg, capability_filter="text-processing", timeout=5.0)
    for agent_name, reply in replies.items():
        print(f"{agent_name}: {reply.payload}")

Example 3: Adding a schema validation error handler

The protocol automatically validates payloads against schemas. If validation fails, the sender receives an error message. You can customize this:

from acp import SchemaValidationError

async def handle_validation_error(msg: Message, error: SchemaValidationError):
    error_reply = msg.reply(
        payload={"error": f"Validation failed: {error}", "status": "error"}
    )
    await agent.send(error_reply)

agent.register_validation_error_handler(handle_validation_error)

This turns schema violations into structured error responses instead of silent failures.

How This Protocol Cleaned Up Our Architecture

Before the protocol, our architecture had:

• 15 different message formats
• 30+ custom adapters
• 40% of agent code dedicated to parsing and formatting
• Average latency of 2.3 seconds per cross-agent call
• Debugging required tracing through 6 different log formats

After adopting ACP:

• All agents speak one format
• Zero adapters between agents
• Agent code is 60% smaller (no parsing logic)
• Average latency dropped to 0.4 seconds (async broker with queuing)
• Every message has a trace ID, making debugging trivial

The protocol also enabled features we didn’t plan for:

• **Message replay** – We can replay failed messages for debugging or retraining.
• **Schema versioning** – Old agents still work with new schemas via backward-compatible fields.
• **Observability** – Every message carries metadata that feeds into our monitoring dashboards.

Lessons Learned

1. **Start with a schema registry early.** Even if you have only two agents, define shared schemas. It prevents drift. 2. **Use async communication.** Synchronous calls between agents create tight coupling. The Redis-backed broker gave us resilience. 3. **Don’t over-engineer the protocol.** Our first version tried to support every edge case. The simplified version (envelope + typed payloads + async routing) covered 95% of needs. 4. **Test with schema validation turned on.** It catches bugs before they become runtime failures.

Conclusion

The protocol that cleaned up our agent architecture wasn’t a revolutionary AI breakthrough. It was a return to software engineering fundamentals: standardize communication, decouple components, and validate data at the boundary. By adopting the Agent Communication Protocol with Redis as the broker, we transformed a chaotic multi-agent system into a clean, maintainable, and fast architecture.

If your agent ecosystem is growing faster than your ability to manage it, consider adopting a similar protocol. The installation takes 15 minutes, but the payoff in reduced complexity and increased velocity is enormous. Our agents now spend their time doing what they’re good at—processing data—instead of wrestling with each other’s message formats.

The protocol didn’t just clean up our architecture. It made our agents work together like a well-orchestrated team. And that, in the end, is what multi-agent systems should be about.