Internal Messaging

AFK’s internal messaging system lets agents communicate via structured envelopes with at-least-once delivery and idempotency. Use it when agents in the same system need to exchange data reliably.

Message lifecycle

The InternalA2AEnvelope

Every message is wrapped in a typed envelope:

from afk.messaging import InternalA2AEnvelope

envelope = InternalA2AEnvelope(
    message_type="request",
    run_id="run-42",
    thread_id="thread-42",
    conversation_id="conversation-42",
    correlation_id="task-42",
    payload={"findings": ["fact 1", "fact 2"]},
    idempotency_key="task-42-research",
    source_agent="researcher",
    target_agent="writer",
)

Field	Type	Purpose
`message_type`	`str`	`request`, `response`, or `event`
`run_id`	`str`	Run identifier for tracing
`thread_id`	`str`	Memory thread identifier
`conversation_id`	`str`	Cross-run conversation identifier
`payload`	`dict`	Message data (any JSON-serializable content)
`correlation_id`	`str`	Groups related messages in a workflow
`idempotency_key`	`str`	Deduplication — same key = same message
`source_agent`	`str`	Name of the sending agent
`target_agent`	`str`	Name of the receiving agent
`metadata`	`dict`	JSON-safe tracing or routing metadata
`timestamp_ms`	`int`	Creation timestamp in milliseconds

Delivery behavior

Success path
Retry path
Failure path

Sender creates and submits the envelope
Delivery store checks the idempotency key (rejects duplicates)
Message is delivered to the receiver
Receiver processes and ACKs
Store marks as delivered

Result: Message processed exactly once.

Idempotency and correlation

Always set an idempotency_key. Without it, retry deliveries can cause duplicate processing. Use a deterministic key derived from the task context (e.g., f"{task_id}-{step_name}").

Correlation IDs group related messages across a workflow. When debugging, filter by correlation_id to see the full message chain for a task.

# All messages in this workflow share the same correlation_id
correlation = f"analysis-{run_id}"

await send(InternalA2AEnvelope(
    message_type="request",
    run_id=run_id,
    thread_id=thread_id,
    conversation_id=conversation_id,
    correlation_id=correlation,
    source_agent="coordinator",
    target_agent="researcher",
    payload={"query": "AI trends"},
    idempotency_key=f"{correlation}-research",
))

await send(InternalA2AEnvelope(
    message_type="request",
    run_id=run_id,
    thread_id=thread_id,
    conversation_id=conversation_id,
    correlation_id=correlation,
    source_agent="coordinator",
    target_agent="writer",
    payload={"topic": "AI trends"},
    idempotency_key=f"{correlation}-write",
))

Delivery store backends

In-memory
Persistent

Default. Fast, no setup. State lost on restart.

from afk.messaging import InMemoryA2ADeliveryStore

store = InMemoryA2ADeliveryStore()

Implement the DeliveryStore protocol for durable messaging:

from afk.messaging import A2ADeliveryStore, InternalA2AEnvelope

class RedisDeliveryStore(A2ADeliveryStore):
    async def get_success(self, idempotency_key: str): ...
    async def record_success(self, idempotency_key: str, response): ...
    async def record_dead_letter(self, dead_letter): ...
    async def list_dead_letters(self): ...

Internal Messaging

Message lifecycle

The InternalA2AEnvelope

Delivery behavior

Idempotency and correlation

Delivery store backends

Next steps

A2A Protocol

Task Queues

Documentation Index

​Message lifecycle

​The InternalA2AEnvelope

​Delivery behavior

​Idempotency and correlation

​Delivery store backends

​Next steps

A2A Protocol

Task Queues

Message lifecycle

The InternalA2AEnvelope

Delivery behavior

Idempotency and correlation

Delivery store backends

Next steps