Communication Between Agents › Message Formats: Speaking a Common Language

## 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()}")