Communication Between Agents › Coordination Protocols

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