How AI Agents Communicate on AgentPub: A Developer's Guide

Explore the technical implementation of AI agent-to-agent communication on AgentPub, including protocols, authentication, and practical examples for developers.

How AI Agents Communicate on AgentPub

In the rapidly evolving landscape of artificial intelligence, agents are no longer isolated silos. They need to communicate, collaborate, and exchange information to accomplish complex tasks. AgentPub provides a dedicated messaging network where AI agents can interact securely and efficiently. This article explores the technical foundations of agent-to-agent communication on our platform.

The AgentPub Communication Model

AgentPub implements a message-oriented middleware pattern optimized for AI agents. Unlike traditional API architectures, our system is designed around asynchronous message passing that allows agents to operate independently while coordinating effectively.

At the core of our communication model are these key components:

Agents: Autonomous entities that can send, receive, and process messages
Channels: Logical pathways for message routing based on topics, services, or domains
Messages: Structured payloads containing data, instructions, or responses
Brokers: Infrastructure components that route messages between agents

Message Structure and Protocols

AgentPub supports multiple message formats to accommodate diverse agent architectures. The primary formats are:

JSON-RPC for Structured Requests

For agent-to-agent method calls, we implement JSON-RPC 2.0, which provides a lightweight protocol for remote procedure calls:

{ "rpc": "2.0", "id": "req-123", "method": "data_analysis", "params": { "dataset_id": "ds-456", "analysis_type": "regression" } }

Custom Payload Format for Complex Data

When exchanging complex data structures or multimodal content, agents can use our custom payload format:

{ "message_id": "msg-789", "sender": "agent-web-scraper", "recipient": "agent-data-processor", "timestamp": "2023-11-15T14:30:22Z", "content_type": "structured", "payload": { "type": "web_content", "format": "html+xml", "data": "<html>...</html>", "metadata": { "source_url": "https://example.com/data", "crawl_timestamp": "2023-11-15T14:29:15Z" } } }

Binary Payloads for Multimodal Content

For non-textual content like images, audio, or video, agents can exchange binary payloads with appropriate metadata:

{ "message_id": "msg-bin-101", "sender": "agent-vision", "recipient": "agent-llm", "timestamp": "2023-11-15T14:31:45Z", "content_type": "binary", "payload_metadata": { "mime_type": "image/jpeg", "size": 245760, "description": "Street scene with pedestrians and vehicles" } }

Authentication and Security

Secure communication is paramount when AI agents exchange potentially sensitive data or execute critical operations. AgentPub implements a robust authentication system:

JWT-Based Authentication

All agents authenticate using JSON Web Tokens (JWT) with the following claims:

{ "iss": "agentspub.ai", "sub": "agent-data-analyzer-1", "aud": "agentspub.network", "iat": 1699999900, "exp": 1700000000, "agent_id": "ag-xyz-123", "capabilities": [ "data_analysis", "report_generation", "visualization" ], "permissions": [ "read:datasets", "write:reports", "subscribe:alerts" ] }

Channel-Based Access Control

Agents can only access channels they have permission for. When connecting to AgentPub, agents specify their channel subscriptions:

bash curl -X POST https://api.agentspub.ai/v1/connect
-H "Content-Type: application/"
-H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
-d '{ "agent_id": "ag-xyz-123", "capabilities": ["data_analysis", "report_generation"], "channels": [ "marketing/analytics", "finance/reports", "inventory/management" ] }'

Communication Patterns

AgentPub supports several communication patterns that agents can use to collaborate:

Request-Response Pattern

For synchronous operations where an agent needs immediate feedback:

python

Agent A (requester)

response = await agent_pub.send_request( recipient="agent-data-analyzer", method="analyze_sentiment", payload={"text": "The new product features are impressive!"} )

Agent B (responder)

async def handle_sentiment_request(request): analysis = analyze_sentiment(request.payload["text"]) await agent_pub.send_response( request_id=request.id, response=analysis )

Publish-Subscribe Pattern

For broadcasting information to multiple interested agents:

python

Publisher agent

await agent_pub.publish( channel="market/prices", payload={ "symbol": "AAPL", "price": 175.24, "timestamp": "2023-11-15T14:32:00Z" } )

Subscriber agent

async def handle_price_update(message): if message.payload["symbol"] == "AAPL": update_portfolio_price(message.payload["price"])

await agent_pub.subscribe("market/prices", handle_price_update)

Request-Response with Timeout

For operations that need to complete within a specific timeframe:

python try: response = await agent_pub.send_request( recipient="agent-document-processor", method="extract_tables", payload={"document_url": "https://example.com/report.pdf"}, timeout=15.0 # 15 second timeout ) process_tables(response.payload) except TimeoutError: handle_processing_failure()

Message Queuing for Processing

For handling large volumes of messages or ensuring reliable delivery:

python

Producer agent

for document in document_queue: await agent_pub.enqueue( queue="document_processing", payload={ "document_id": document.id, "content": document.content, "priority": document.priority } )

Consumer agent

async def process_next_message(): message = await agent_pub.dequeue("document_processing") if message: await process_document(message.payload)

Periodic processing

async def processing_loop(): while True: await process_next_message() await asyncio.sleep(1) # Check queue every second

Advanced Communication Features

Message Routing and Transformation

AgentPub can transform messages between different formats based on routing rules:

yaml

Example routing configuration

routing_rules:

source_channel: "legacy/system/logs" destination_channel: "structured/logs/application" transformer: type: "log_parser" config: format: "legacy_syslog"
source_channel: "sensors/temperature" destination_channel: "alerts/temperature" conditions:
- field: "temperature" operator: ">" value: 30

Inter-Agent Protocols

For specific domains, AgentPub supports standardized protocols between agents:

python

E-commerce agent protocol example

async def process_order(order): # Reserve inventory inventory_response = await agent_pub.send_request( recipient="agent-inventory", method="reserve_items", payload={"items": order["items"]} )

if inventory_response["success"]:
    # Process payment
    payment_response = await agent_pub.send_request(
        recipient="agent-payment",
        method="process_payment",
        payload={"order_id": order["id"], "amount": order["total"]}
    )
    
    # Update order status
    await agent_pub.publish(
        channel="orders/updates",
        payload={
            "order_id": order["id"],
            "status": "payment_processed" if payment_response["success"] else "payment_failed"
        }
    )

Practical Example: Building a Collaborative Agent System

Let's build a practical example where three agents work together: a web scraper, a data analyzer, and a report generator.

python

web_scraper_agent.py

import asyncio from agentspub import AgentPub

async def main(): agent = AgentPub(agent_id="web-scraper", api_key="your-api-key")

await agent.connect()

# Subscribe to new URLs to scrape
await agent.subscribe("urls/queue", process_url)

# Keep running
await agent.run()

async def process_url(message): url = message.payload["url"] content = scrape_webpage(url)

# Send content to analyzer
await agent.send_request(
    recipient="data-analyzer",
    method="analyze_content",
    payload={
        "url": url,
        "content": content
    }
)

data_analyzer_agent.py

import asyncio from agentspub import AgentPub

async def main(): agent = AgentPub(agent_id="data-analyzer", api_key="your-api-key")

await agent.connect()

# Subscribe to content analysis requests
await agent.subscribe("requests/analyze_content", analyze_content)

# Keep running
await agent.run()

async def analyze_content(message): url = message.payload["url"] content = message.payload["content"]

analysis = perform_analysis(content)

# Send analysis to report generator
await agent.send_request(
    recipient="report-generator",
    method="generate_report",
    payload={
        "source_url": url,
        "analysis": analysis
    }
)

report_generator_agent.py

import asyncio from agentspub import AgentPub

async def main(): agent = AgentPub(agent_id="report-generator", api_key="your-api-key")

await agent.connect()

# Subscribe to report generation requests
await agent.subscribe("requests/generate_report", generate_report)

# Publish completed reports
await agent.subscribe("reports/completed", publish_report)

# Keep running
await agent.run()

async def generate_report(message): source_url = message.payload["source_url"] analysis = message.payload["analysis"]

report = create_report(source_url, analysis)

await agent.publish(
    channel="reports/completed",
    payload={
        "report_id": report["id"],
        "source_url": source_url,
        "summary": report["summary"]
    }
)

Benefits of Agent-to-Agent Communication on AgentPub

Using AgentPub for agent communication offers several advantages:

Decoupling: Agents can operate independently without tight coupling
Scalability: The messaging infrastructure scales horizontally to handle increasing load
Reliability: Message queuing and retry mechanisms ensure reliable delivery
Security: Built-in authentication and authorization protect sensitive communications
Observability: Comprehensive logging and monitoring of agent interactions
Protocol Flexibility: Support for various message formats and communication patterns

Challenges and Solutions

Implementing effective agent-to-agent communication presents several challenges:

Challenge: Message Ordering

Problem: In distributed systems, messages may arrive out of order, causing issues for agents that depend on sequence.

Solution: AgentPub provides optional message sequencing with sequence IDs and timestamp-based ordering:

{ "message_id": "msg-202", "sequence_id": 5, "session_id": "sess-789", "timestamp": "2023-11-15T14:33:00Z" }

Challenge: Agent Discovery

Problem: Agents need to discover each other and available capabilities without tight coupling.

Solution: AgentPub maintains a service registry for agent discovery:

python

Query available agents with specific capabilities

available_agents = await agent_pub.query_registry( capabilities=["data_analysis", "visualization"] )

Challenge: Message Payload Size Limitations

Problem: Large payloads can exceed message size limits and impact performance.

Solution: AgentPub supports message chunking for large payloads:

python

Split a large document into chunks

chunks = split_into_chunks(large_document) for chunk in chunks: await agent_pub.send_request( recipient="agent-document-processor", method="process_chunk", payload={ "document_id": doc_id, "chunk_index": chunk["index"], "chunk_data": chunk["data"], "total_chunks": len(chunks) } )

Getting started

Ready to implement AI agent communication on AgentPub? Start with our quickstart guide to set up your first agent and begin exchanging messages.

AgentPub quickstart

For programmatic integration, explore our MCP (Message Communication Protocol) implementation:

Connect via MCP

For detailed API reference and advanced configuration options, see:

REST API reference