Communication Patterns

import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from env

def run_agent(role: str, system_prompt: str, user_input: str) -> str:
    # Each agent is just a Claude call with a specialized system prompt
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": user_input}]
    )
    return response.content[0].text

# Agent A: Research agent finds information
research_result = run_agent(
    role="researcher",
    system_prompt="You are a research agent. Return findings as JSON with keys: topic, findings (list), confidence (0-1).",
    user_input="Research the top 3 competitors in AI education pricing."
)

# Agent B: Analysis agent receives research output DIRECTLY
analysis_result = run_agent(
    role="analyst",
    system_prompt="You are an analysis agent. You receive research data and produce strategic recommendations.",
    user_input=f"Analyze this research and recommend a pricing strategy:\n{research_result}"
)

print(analysis_result)
# "Based on the competitor analysis, I recommend a freemium model...
#  Tier 1: Free (52 courses). Tier 2: $29/mo (premium features)..."

import asyncio

# Define specialist agents that all react to the same event
AGENTS = {
    "sentiment": "Analyze the sentiment of this customer message. Return: positive/negative/neutral + confidence.",
    "category": "Categorize this support ticket. Return: billing/technical/feature-request/other.",
    "urgency": "Rate the urgency of this message: low/medium/high/critical. Return urgency + reasoning.",
}

async def broadcast(message: str) -> dict:
    # All agents process the SAME message in parallel
    # Each agent sees the same input, but does something different with it
    tasks = {
        name: run_agent_async(name, prompt, message)
        for name, prompt in AGENTS.items()
    }
    results = await asyncio.gather(*tasks.values())
    return dict(zip(tasks.keys(), results))

# One customer message → three specialist analyses in parallel
ticket = "I've been charged twice and my account is locked. This is urgent!"
results = asyncio.run(broadcast(ticket))
# → {"sentiment": "negative (0.95)", "category": "billing", "urgency": "critical"}

# The blackboard is shared state that all agents can read and write
# In production, this would be a database (Supabase, Redis, etc.)
blackboard = {
    "task": "Write a blog post about AI agents",
    "status": "planning",
    "sections": {},       # agents write their contributions here
    "feedback": [],       # review comments from other agents
}

def research_agent(board: dict) -> dict:
    # Read the task from the blackboard
    topic = board["task"]
    # Do research, then write findings BACK to the blackboard
    board["sections"]["research"] = run_agent("researcher", RESEARCH_PROMPT, topic)
    board["status"] = "research_complete"
    return board

def writer_agent(board: dict) -> dict:
    # Read research from the blackboard (written by research agent)
    research = board["sections"]["research"]
    # Write the draft, add it to the blackboard
    board["sections"]["draft"] = run_agent("writer", WRITER_PROMPT, research)
    board["status"] = "draft_complete"
    return board

def editor_agent(board: dict) -> dict:
    # Read BOTH research and draft — full context from the blackboard
    context = f"Research: {board['sections']['research']}\nDraft: {board['sections']['draft']}"
    board["feedback"].append(run_agent("editor", EDITOR_PROMPT, context))
    board["status"] = "review_complete"
    return board

# Orchestrate: each agent reads/writes the same blackboard
blackboard = research_agent(blackboard)   # writes research
blackboard = writer_agent(blackboard)     # reads research, writes draft
blackboard = editor_agent(blackboard)     # reads both, writes feedback

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

@dataclass
class AgentMessage:
    # Who sent it and who should receive it
    sender: str                # "research-agent"
    recipient: str             # "analysis-agent"
    message_type: str          # "research_complete"

    # The actual data being passed
    payload: dict[str, Any]    # {"topic": "...", "findings": [...]}

    # Metadata for debugging and tracing
    confidence: float = 0.0    # How reliable is this data? (0-1)
    tokens_used: int = 0      # Cost tracking
    timestamp: str = field(    # When was this created?
        default_factory=lambda: datetime.now().isoformat()
    )

# Usage: research agent sends findings to analysis agent
msg = AgentMessage(
    sender="research-agent",
    recipient="analysis-agent",
    message_type="research_complete",
    payload={
        "topic": "competitor pricing Q4",
        "findings": ["Competitor A: $29/mo", "Competitor B: $49/mo"],
        "sources_checked": 12,
    },
    confidence=0.87,           # High confidence — 12 sources verified
    tokens_used=4200,
)
# Structured, typed, traceable. The receiving agent knows
# exactly what it got, who sent it, and how reliable it is.