Communication Between Agents: Message Formats, Protocols & Coordination Patterns

## Using Claude Sonnet 4.5 for agent communication
from anthropic import Anthropic
import json

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

## Agent A: Research Agent
def research_agent(query):
    """
    Researches information and returns findings.
    """
    system_prompt = """You are a research specialist. 
    When asked to research something, provide factual information in a clear format.
    Always structure your response as JSON with 'topic', 'findings', and 'confidence' fields."""
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        system=system_prompt,
        messages=[{"role": "user", "content": query}]
    )
    
    return response.content[0].text

## Agent B: Analysis Agent
def analysis_agent(research_data):
    """
    Analyzes research data and provides recommendations.
    """
    system_prompt = """You are an analysis specialist.
    You receive research findings and provide actionable recommendations.
    Focus on practical insights."""
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        system=system_prompt,
        messages=[{"role": "user", "content": f"Analyze this research: {research_data}"}]
    )
    
    return response.content[0].text

## Simple communication flow
research_result = research_agent("What is the current sentiment around electric vehicles?")
print(f"Research Agent found:\n{research_result}\n")

analysis_result = analysis_agent(research_result)
print(f"Analysis Agent recommends:\n{analysis_result}")

## Agent A sends structured data
message = {
    "from": "research_agent",
    "to": "analysis_agent",
    "timestamp": "2025-11-10T14:30:00Z",
    "message_type": "research_findings",
    "data": {
        "topic": "Bitcoin price",
        "current_price": 45000,
        "currency": "USD",
        "volatility": 0.15,
        "confidence": 0.92
    }
}

## Using Claude Sonnet 4.5 with structured message passing
from anthropic import Anthropic
import json
from datetime import datetime

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

class AgentMessage:
    """
    Standard message format for agent communication.
    """
    def __init__(self, sender, recipient, message_type, data):
        self.sender = sender
        self.recipient = recipient
        self.message_type = message_type
        self.data = data
        self.timestamp = datetime.utcnow().isoformat()
    
    def to_json(self):
        return json.dumps({
            "from": self.sender,
            "to": self.recipient,
            "type": self.message_type,
            "timestamp": self.timestamp,
            "data": self.data
        }, indent=2)
    
    @staticmethod
    def from_json(json_str):
        obj = json.loads(json_str)
        msg = AgentMessage(
            obj["from"],
            obj["to"],
            obj["type"],
            obj["data"]
        )
        msg.timestamp = obj["timestamp"]
        return msg

def extract_json(text):
    """
    Extract JSON from text that might include markdown code blocks.
    """
    import re
    # Build the pattern for code fences dynamically to avoid backtick issues
    fence = chr(96) * 3  # Three backticks
    pattern = fence + r'(?:json)?\s*\n?([\s\S]*?)\n?' + fence
    code_block_match = re.search(pattern, text)
    if code_block_match:
        return code_block_match.group(1).strip()
    # Otherwise, try to find JSON object directly
    json_match = re.search(r'\{[\s\S]*\}', text)
    if json_match:
        return json_match.group(0)
    return text.strip()

def research_agent_v2(query):
    """
    Research agent that returns structured messages.
    """
    system_prompt = """You are a research specialist.
    Return findings as a valid JSON object with these exact fields:
    - topic: what you researched
    - findings: key information discovered (as a string or simple list)
    - sources: where the information came from (as a string or list)
    - confidence: 0-1 score of how confident you are
    
    Return ONLY the JSON object, no other text or markdown."""
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        system=system_prompt,
        messages=[{"role": "user", "content": query}]
    )
    
    # Parse the research findings (extract JSON from potential markdown)
    raw_text = response.content[0].text
    json_text = extract_json(raw_text)
    try:
        findings = json.loads(json_text)
    except json.JSONDecodeError:
        # Fallback: create a simple findings object from the raw text
        findings = {
            "topic": query,
            "findings": raw_text[:500],
            "sources": ["LLM response"],
            "confidence": 0.7
        }
    
    # Wrap in a standard message
    message = AgentMessage(
        sender="research_agent",
        recipient="analysis_agent",
        message_type="research_complete",
        data=findings
    )
    
    return message

def analysis_agent_v2(message):
    """
    Analysis agent that expects structured messages.
    """
    # Verify message type
    if message.message_type != "research_complete":
        return AgentMessage(
            sender="analysis_agent",
            recipient=message.sender,
            message_type="error",
            data={"error": f"Expected 'research_complete', got '{message.message_type}'"}
        )
    
    system_prompt = """You are an analysis specialist.
    You receive research findings as JSON and provide recommendations.
    Return your analysis as a valid JSON object with:
    - recommendation: your main recommendation (as a string)
    - reasoning: why you recommend this (as a string)
    - confidence: 0-1 score of confidence
    
    Return ONLY the JSON object, no other text or markdown."""
    
    # Extract research data
    research_data = json.dumps(message.data)
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        system=system_prompt,
        messages=[{"role": "user", "content": f"Analyze: {research_data}"}]
    )
    
    # Parse the analysis (extract JSON from potential markdown)
    raw_text = response.content[0].text
    json_text = extract_json(raw_text)
    try:
        analysis = json.loads(json_text)
    except json.JSONDecodeError:
        # Fallback: create a simple analysis object from the raw text
        analysis = {
            "recommendation": raw_text[:300],
            "reasoning": "Based on the research findings",
            "confidence": 0.7
        }
    
    # Return structured response
    return AgentMessage(
        sender="analysis_agent",
        recipient="research_agent",
        message_type="analysis_complete",
        data=analysis
    )

## Example usage
print("=== Structured Agent Communication ===\n")

## Research agent does its work
research_msg = research_agent_v2("Research the benefits of renewable energy")
print(f"Research Agent sent:\n{research_msg.to_json()}\n")

## Analysis agent receives and processes
analysis_msg = analysis_agent_v2(research_msg)
print(f"Analysis Agent sent:\n{analysis_msg.to_json()}")

## Using Claude Sonnet 4.5 for pub-sub coordination
class MessageBroker:
    """
    Simple publish-subscribe message broker for agent communication.
    """
    def __init__(self):
        self.subscriptions = {}  # topic -> list of agents
        self.message_history = []
    
    def subscribe(self, topic, agent_name):
        """
        Agent subscribes to a topic.
        """
        if topic not in self.subscriptions:
            self.subscriptions[topic] = []
        if agent_name not in self.subscriptions[topic]:
            self.subscriptions[topic].append(agent_name)
            print(f"{agent_name} subscribed to '{topic}'")
    
    def publish(self, topic, message):
        """
        Publish a message to all subscribers of a topic.
        """
        self.message_history.append({
            "topic": topic,
            "message": message,
            "timestamp": datetime.utcnow().isoformat()
        })
        
        if topic in self.subscriptions:
            subscribers = self.subscriptions[topic]
            print(f"\nPublishing to '{topic}' ({len(subscribers)} subscribers)")
            return subscribers
        else:
            print(f"\nPublishing to '{topic}' (no subscribers)")
            return []
    
    def get_history(self, topic=None):
        """
        Get message history, optionally filtered by topic.
        """
        if topic:
            return [msg for msg in self.message_history if msg["topic"] == topic]
        return self.message_history

## Example: Multiple agents coordinating through pub-sub
broker = MessageBroker()

## Agents subscribe to topics they care about
broker.subscribe("market_data", "trading_agent")
broker.subscribe("market_data", "analysis_agent")
broker.subscribe("market_data", "reporting_agent")
broker.subscribe("user_actions", "logging_agent")
broker.subscribe("user_actions", "analytics_agent")

## When market data arrives, all interested agents are notified
market_subscribers = broker.publish("market_data", {
    "symbol": "AAPL",
    "price": 178.50,
    "change": 2.3
})
print(f"Notified: {', '.join(market_subscribers)}")

## When a user action occurs, different agents are notified
action_subscribers = broker.publish("user_actions", {
    "action": "purchase",
    "item": "premium_subscription"
})
print(f"Notified: {', '.join(action_subscribers)}")

class ConversationThread:
    """
    Manages a conversation thread between agents.
    """
    def __init__(self, thread_id, participants):
        self.thread_id = thread_id
        self.participants = participants
        self.messages = []
    
    def add_message(self, sender, content):
        """
        Add a message to the thread.
        """
        if sender not in self.participants:
            raise ValueError(f"{sender} is not a participant in this thread")
        
        message = {
            "thread_id": self.thread_id,
            "sender": sender,
            "content": content,
            "timestamp": datetime.utcnow().isoformat(),
            "sequence": len(self.messages)
        }
        self.messages.append(message)
        return message
    
    def get_history(self):
        """
        Get the full conversation history.
        """
        return self.messages
    
    def get_context_for_agent(self, agent_name):
        """
        Get conversation context formatted for a specific agent.
        """
        context = f"Conversation thread {self.thread_id}\n"
        context += f"Participants: {', '.join(self.participants)}\n\n"
        for msg in self.messages:
            context += f"{msg['sender']}: {msg['content']}\n"
        return context

## Example: Two agents having a conversation
thread = ConversationThread(
    thread_id="meeting_planning_001",
    participants=["coordinator_agent", "calendar_agent"]
)

thread.add_message("coordinator_agent", "I need to schedule a team meeting for next week.")
thread.add_message("calendar_agent", "I can see Tuesday 2pm or Thursday 3pm are available. Which works better?")
thread.add_message("coordinator_agent", "Tuesday 2pm works. Please book it.")
thread.add_message("calendar_agent", "Meeting scheduled for Tuesday at 2pm. I've sent invites to all team members.")

print("=== Conversation Thread ===")
print(thread.get_context_for_agent("coordinator_agent"))

## Using Claude Sonnet 4.5 for task delegation
class TaskDelegationProtocol:
    """
    Protocol for delegating tasks between agents.
    """
    def __init__(self):
        self.active_tasks = {}
    
    def request_task(self, task_id, from_agent, to_agent, task_description):
        """
        Coordinator requests a task from a worker.
        """
        task = {
            "task_id": task_id,
            "coordinator": from_agent,
            "worker": to_agent,
            "description": task_description,
            "status": "requested",
            "created_at": datetime.utcnow().isoformat(),
            "updates": []
        }
        self.active_tasks[task_id] = task
        
        return AgentMessage(
            sender=from_agent,
            recipient=to_agent,
            message_type="task_request",
            data=task
        )
    
    def acknowledge_task(self, task_id, worker_agent):
        """
        Worker acknowledges receiving the task.
        """
        if task_id not in self.active_tasks:
            raise ValueError(f"Unknown task: {task_id}")
        
        task = self.active_tasks[task_id]
        task["status"] = "in_progress"
        task["updates"].append({
            "timestamp": datetime.utcnow().isoformat(),
            "message": "Task acknowledged and started"
        })
        
        return AgentMessage(
            sender=worker_agent,
            recipient=task["coordinator"],
            message_type="task_acknowledged",
            data={"task_id": task_id}
        )
    
    def update_progress(self, task_id, worker_agent, progress_message):
        """
        Worker sends a progress update.
        """
        if task_id not in self.active_tasks:
            raise ValueError(f"Unknown task: {task_id}")
        
        task = self.active_tasks[task_id]
        task["updates"].append({
            "timestamp": datetime.utcnow().isoformat(),
            "message": progress_message
        })
        
        return AgentMessage(
            sender=worker_agent,
            recipient=task["coordinator"],
            message_type="task_progress",
            data={
                "task_id": task_id,
                "progress": progress_message
            }
        )
    
    def complete_task(self, task_id, worker_agent, results):
        """
        Worker completes the task and sends results.
        """
        if task_id not in self.active_tasks:
            raise ValueError(f"Unknown task: {task_id}")
        
        task = self.active_tasks[task_id]
        task["status"] = "completed"
        task["results"] = results
        task["completed_at"] = datetime.utcnow().isoformat()
        
        return AgentMessage(
            sender=worker_agent,
            recipient=task["coordinator"],
            message_type="task_complete",
            data={
                "task_id": task_id,
                "results": results
            }
        )
    
    def get_task_status(self, task_id):
        """
        Get the current status of a task.
        """
        return self.active_tasks.get(task_id)

## Example usage
protocol = TaskDelegationProtocol()

print("=== Task Delegation Protocol ===\n")

## Step 1: Coordinator requests a task
request_msg = protocol.request_task(
    task_id="research_001",
    from_agent="coordinator",
    to_agent="research_agent",
    task_description="Research the top 3 cloud providers and compare pricing"
)
print(f"1. Task Requested:\n{request_msg.to_json()}\n")

## Step 2: Worker acknowledges
ack_msg = protocol.acknowledge_task("research_001", "research_agent")
print(f"2. Task Acknowledged:\n{ack_msg.to_json()}\n")

## Step 3: Worker sends progress update
progress_msg = protocol.update_progress(
    "research_001",
    "research_agent",
    "Completed research on AWS, starting Azure"
)
print(f"3. Progress Update:\n{progress_msg.to_json()}\n")

## Step 4: Worker completes task
complete_msg = protocol.complete_task(
    "research_001",
    "research_agent",
    {
        "providers": ["AWS", "Azure", "Google Cloud"],
        "comparison": "Detailed pricing comparison...",
        "recommendation": "AWS offers best value for your use case"
    }
)
print(f"4. Task Complete:\n{complete_msg.to_json()}\n")

## Check final status
status = protocol.get_task_status("research_001")
print(f"Final Status: {status['status']}")
print(f"Updates: {len(status['updates'])} progress updates")

class ConsensusProtocol:
    """
    Simple consensus protocol for agent decision-making.
    """
    def __init__(self, participants, threshold=0.5):
        self.participants = participants
        self.threshold = threshold  # Fraction of votes needed to pass
        self.proposals = {}
    
    def propose(self, proposal_id, proposer, description):
        """
        An agent proposes something for the group to decide.
        """
        if proposer not in self.participants:
            raise ValueError(f"{proposer} is not a participant")
        
        self.proposals[proposal_id] = {
            "id": proposal_id,
            "proposer": proposer,
            "description": description,
            "votes": {},
            "status": "voting",
            "created_at": datetime.utcnow().isoformat()
        }
        
        print(f"\n{proposer} proposed: {description}")
        print(f"Proposal ID: {proposal_id}")
        print(f"Waiting for votes from: {', '.join(self.participants)}")
    
    def vote(self, proposal_id, voter, vote):
        """
        An agent votes on a proposal (True for agree, False for disagree).
        """
        if proposal_id not in self.proposals:
            raise ValueError(f"Unknown proposal: {proposal_id}")
        
        if voter not in self.participants:
            raise ValueError(f"{voter} is not a participant")
        
        proposal = self.proposals[proposal_id]
        proposal["votes"][voter] = vote
        
        print(f"{voter} voted: {'agree' if vote else 'disagree'}")
        
        # Check if all votes are in
        if len(proposal["votes"]) == len(self.participants):
            self._finalize_proposal(proposal_id)
    
    def _finalize_proposal(self, proposal_id):
        """
        Tally votes and finalize the proposal.
        """
        proposal = self.proposals[proposal_id]
        votes = proposal["votes"]
        
        agree_count = sum(1 for v in votes.values() if v)
        total_votes = len(votes)
        agreement_ratio = agree_count / total_votes
        
        if agreement_ratio >= self.threshold:
            proposal["status"] = "accepted"
            result = "ACCEPTED"
        else:
            proposal["status"] = "rejected"
            result = "REJECTED"
        
        proposal["finalized_at"] = datetime.utcnow().isoformat()
        proposal["agreement_ratio"] = agreement_ratio
        
        print(f"\nProposal {proposal_id}: {result}")
        print(f"Votes: {agree_count}/{total_votes} agreed ({agreement_ratio:.0%})")
        print(f"Threshold: {self.threshold:.0%}")
    
    def get_result(self, proposal_id):
        """
        Get the result of a proposal.
        """
        return self.proposals.get(proposal_id)

## Example: Agents reaching consensus
agents = ["agent_a", "agent_b", "agent_c", "agent_d"]
consensus = ConsensusProtocol(agents, threshold=0.75)

## Propose a decision
consensus.propose(
    "upgrade_001",
    "agent_a",
    "Upgrade to the new model version (higher cost but better performance)"
)

## Agents vote
consensus.vote("upgrade_001", "agent_a", True)   # Proposer agrees
consensus.vote("upgrade_001", "agent_b", True)   # Agent B agrees
consensus.vote("upgrade_001", "agent_c", True)   # Agent C agrees
consensus.vote("upgrade_001", "agent_d", False)  # Agent D disagrees

## Result is automatically finalized when all votes are in
result = consensus.get_result("upgrade_001")
print(f"\nFinal decision: {result['status']}")

import time

def send_with_timeout(sender_func, timeout_seconds=5):
    """
    Send a message and wait for response with timeout.
    """
    start_time = time.time()
    
    try:
        # In a real system, this would be async
        response = sender_func()
        elapsed = time.time() - start_time
        
        if elapsed > timeout_seconds:
            return {
                "status": "timeout",
                "message": f"No response after {timeout_seconds} seconds"
            }
        
        return {
            "status": "success",
            "response": response
        }
    except Exception as e:
        return {
            "status": "error",
            "message": str(e)
        }

def send_with_retry(sender_func, max_retries=3):
    """
    Send a message with automatic retries on failure.
    """
    for attempt in range(max_retries):
        try:
            response = sender_func()
            return {
                "status": "success",
                "response": response,
                "attempts": attempt + 1
            }
        except Exception as e:
            if attempt == max_retries - 1:
                # Last attempt failed
                return {
                    "status": "failed",
                    "error": str(e),
                    "attempts": max_retries
                }
            # Wait before retrying (exponential backoff)
            time.sleep(2 ** attempt)

class ReliableMessaging:
    """
    Ensures messages are received by requiring acknowledgments.
    """
    def __init__(self):
        self.pending_acks = {}
    
    def send_reliable(self, message_id, sender, recipient, content):
        """
        Send a message that requires acknowledgment.
        """
        self.pending_acks[message_id] = {
            "sender": sender,
            "recipient": recipient,
            "content": content,
            "sent_at": datetime.utcnow().isoformat(),
            "acknowledged": False
        }
        
        return AgentMessage(
            sender=sender,
            recipient=recipient,
            message_type="reliable_message",
            data={
                "message_id": message_id,
                "content": content,
                "requires_ack": True
            }
        )
    
    def acknowledge(self, message_id, recipient):
        """
        Recipient acknowledges receiving the message.
        """
        if message_id not in self.pending_acks:
            raise ValueError(f"Unknown message: {message_id}")
        
        self.pending_acks[message_id]["acknowledged"] = True
        self.pending_acks[message_id]["acked_at"] = datetime.utcnow().isoformat()
        
        msg = self.pending_acks[message_id]
        return AgentMessage(
            sender=recipient,
            recipient=msg["sender"],
            message_type="acknowledgment",
            data={"message_id": message_id}
        )
    
    def check_pending(self):
        """
        Check for messages that haven't been acknowledged.
        """
        return {
            msg_id: msg 
            for msg_id, msg in self.pending_acks.items() 
            if not msg["acknowledged"]
        }