message-protocol

Agent Message Protocol

Quand utiliser ce skill

Utilise ce skill dès que deux agents ou plus doivent s'échanger des tâches, des résultats ou des signaux de contrôle de façon fiable et traçable — qu'il s'agisse d'une architecture mono-processus (event bus local) ou distribuée (Redis Streams, RabbitMQ, Kafka).

---

Workflow en 10 étapes

1. Définir le format de message standard

Tout message doit contenir ces champs minimaux :

Champ	Type	Description
message_id	UUID v4	Identifiant unique du message
sender	string	ID de l'agent émetteur
recipient	string	ID de l'agent cible ou "broadcast"
type	enum	task_request / task_result / status_update / error / heartbeat / control
payload	dict	Données utiles sérialisées
timestamp	ISO 8601 UTC	Heure d'émission
correlation_id	UUID v4	Relie requête et réponse
schema_version	string	Ex. "1.2" — pour la compatibilité

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

@dataclass
class AgentMessage:
    message_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    sender: str = ""
    recipient: str = ""
    type: str = ""  # task_request | task_result | status_update | error | heartbeat | control
    payload: dict[str, Any] = field(default_factory=dict)
    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    correlation_id: str | None = None
    schema_version: str = "1.0"

2. Typer les messages — un type = un schéma

Chaque type possède un schéma de payload documenté et validé via Pydantic :

from pydantic import BaseModel, Field

class TaskRequestPayload(BaseModel):
    task_id: str
    task_type: str
    input_data: dict
    priority: int = Field(default=5, ge=0, le=9)
    deadline_seconds: int | None = None

class TaskResultPayload(BaseModel):
    task_id: str
    status: str  # "success" | "partial" | "failed"
    output_data: dict
    duration_ms: int

class ErrorPayload(BaseModel):
    error_code: str          # "TIMEOUT" | "VALIDATION_FAILED" | "AGENT_UNAVAILABLE"
    error_message: str
    retry_hint: bool
    retry_after_seconds: int | None = None
    fallback_suggestion: str | None = None
    stack_trace: str | None = None  # debug only, masquer en prod