> ## Documentation Index
> Fetch the complete documentation index at: https://afk.arpan.sh/llms.txt
> Use this file to discover all available pages before exploring further.

# Internal Messaging

> Agent-to-agent messaging with delivery guarantees and idempotency.

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

```mermaid theme={null}
sequenceDiagram
    participant Sender
    participant DeliveryStore
    participant Receiver

    Sender->>DeliveryStore: Send envelope (with idempotency key)
    DeliveryStore->>DeliveryStore: Check deduplication
    DeliveryStore->>Receiver: Deliver message
    alt Success
        Receiver-->>DeliveryStore: ACK
        DeliveryStore->>DeliveryStore: Mark delivered
    else Failure
        Receiver-->>DeliveryStore: NACK / timeout
        DeliveryStore->>DeliveryStore: Schedule retry
        DeliveryStore->>Receiver: Redeliver
    end
```

## The InternalA2AEnvelope

Every message is wrapped in a typed envelope:

```python theme={null}
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

<Tabs>
  <Tab title="Success path">
    1. Sender creates and submits the envelope
    2. Delivery store checks the idempotency key (rejects duplicates)
    3. Message is delivered to the receiver
    4. Receiver processes and ACKs
    5. Store marks as delivered

    **Result:** Message processed exactly once.
  </Tab>

  <Tab title="Retry path">
    1. Delivery fails (receiver timeout, transient error)
    2. Store schedules retry with exponential backoff
    3. Message is redelivered (same idempotency key)
    4. Receiver processes and ACKs on retry

    **Result:** At-least-once delivery. Receiver must be idempotent.
  </Tab>

  <Tab title="Failure path">
    1. All retry attempts exhausted
    2. Message moves to dead-letter queue
    3. Alert generated (if configured)

    **Result:** Message is not lost — it's in the DLQ for manual review.
  </Tab>
</Tabs>

## Idempotency and correlation

<Tip>
  **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}"`).
</Tip>

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

```python theme={null}
# 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

<Tabs>
  <Tab title="In-memory">
    Default. Fast, no setup. State lost on restart.

    ```python theme={null}
    from afk.messaging import InMemoryA2ADeliveryStore

    store = InMemoryA2ADeliveryStore()
    ```
  </Tab>

  <Tab title="Persistent">
    Implement the `DeliveryStore` protocol for durable messaging:

    ```python theme={null}
    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): ...
    ```
  </Tab>
</Tabs>

## Next steps

<CardGroup cols={2}>
  <Card title="A2A Protocol" icon="network-wired" href="/library/a2a">
    Cross-system agent communication.
  </Card>

  <Card title="Task Queues" icon="list-check" href="/library/task-queues">
    Async job processing with queue backends.
  </Card>
</CardGroup>