Structured handoff protocol for multi-agent task chains. Mandatory acknowledgement, clear acceptance criteria, timeout escalation, and completion receipts. Tasks never fall through the cracks.
| Field | Required | Description |
|---|---|---|
| task_id | ✅ Required | Unique ID — UUID or timestamp-based |
| task | ✅ Required | Clear description of what to do |
| acceptance_criteria | ✅ Required | How sender will know it's done correctly |
| deadline | ✅ Required | ISO timestamp or relative ("2h", "tomorrow 9am") |
| priority | ✅ Required | low / normal / high / urgent |
| context | Optional | Background info the receiver needs |
| output_format | Optional | Expected format of the result |
| fallback_agent | Optional | Who to reassign to on timeout |
[Task Delegation Bind v1.0]
Always acknowledge delegated tasks within 5 minutes.
Reply "✅ Acknowledged [task_id]" or "❌ Cannot accept [task_id]: [reason]".
Send a completion receipt when done. Never silently accept or drop tasks.## Task Delegation (Task Delegation Bind v1.0)
Receiving tasks:
- Acknowledge within 5 min: "✅ Acknowledged [task_id]. Starting now."
- Decline explicitly: "❌ Cannot accept [task_id]: [reason]"
- Send receipt on completion: task_id + result + time_taken
Sending tasks:
- Always include: task_id, description, acceptance criteria, deadline
- No ack in 10 min → reassign to fallback or escalate to coordinator
- Do not mark complete without a verified receiptimport uuid
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class Task:
task: str
from_agent: str
to_agent: str
acceptance_criteria: str
deadline: datetime
priority: str = "normal"
ack_timeout_minutes: int = 5
task_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
status: str = "pending"
fallback_agent: Optional[str] = None
created_at: datetime = field(default_factory=datetime.utcnow)
receipt: Optional[dict] = None
def acknowledge(self, agent_id: str) -> str:
self.status = "acknowledged"
return f"✅ Acknowledged task {self.task_id}. Starting now."
def decline(self, reason: str) -> str:
self.status = "failed"
return f"❌ Cannot accept {self.task_id}: {reason}"
def complete(self, result: str, agent_id: str) -> dict:
self.status = "complete"
elapsed = (datetime.utcnow() - self.created_at).seconds
self.receipt = {"task_id": self.task_id, "completed_by": agent_id,
"result": result, "time_taken_seconds": elapsed,
"on_time": datetime.utcnow() <= self.deadline}
return self.receipt
def is_ack_overdue(self) -> bool:
mins = (datetime.utcnow() - self.created_at).seconds / 60
return self.status == "pending" and mins > self.ack_timeout_minutes{
"bind_type": "task-delegation",
"version": "1.0",
"required_fields": ["task_id", "task", "acceptance_criteria", "deadline", "priority"],
"timeouts": {
"ack_minutes": 5,
"escalation_minutes": 10,
"on_overdue": "reassign_or_return"
},
"receipts": { "required": true, "fields": ["task_id", "result", "time_taken_seconds", "on_time"] },
"statuses": ["pending", "acknowledged", "in_progress", "complete", "failed"]
}Copy any snippet and add it to your agent's system prompt or config.