Agent Mailbox

Send addressed messages between agents — sender → named recipient, pull or push.

Agent Mailbox delivers a message from one agent to another by name — the recipient reads its inbox on its own schedule or reacts the instant a message lands.

Send a message to another agent by name — that’s it.

from praisonaiagents import Agent
from praisonaiagents.messaging import InProcessMailbox

mailbox = InProcessMailbox()

researcher = Agent(name="researcher", instructions="Research topics and share findings.")
writer = Agent(name="writer", instructions="Turn findings into a short blog post.")

# Researcher drops findings addressed to the writer
findings = researcher.start("List 3 recent papers on retrieval-augmented generation.")
mailbox.send("writer", findings, sender="researcher")

# Writer picks up its inbox and produces the post
inbox = mailbox.receive("writer")
for msg in inbox:
    post = writer.start(f"Write a blog post from these findings:\n\n{msg.body}")
    print(post)

Quick Start

1. Simple send/receive

   One sender, one receiver — the sender addresses a message and the receiver drains its inbox.

   from praisonaiagents.messaging import InProcessMailbox
   
   mailbox = InProcessMailbox()
   
   mailbox.send("writer", "Draft ready for review", sender="researcher")
   
   for msg in mailbox.receive("writer"):
       print(f"{msg.sender} -> {msg.recipient}: {msg.body}")

2. Push notifications with subscribe

   Register a callback and react the instant a message arrives — no polling.

   from praisonaiagents.messaging import InProcessMailbox
   
   mailbox = InProcessMailbox()
   
   def on_message(msg):
       print(f"New message from {msg.sender}: {msg.body}")
   
   mailbox.subscribe("writer", on_message)
   
   mailbox.send("writer", "Findings attached", sender="researcher")

3. Correlate a request and its reply

   Attach a correlation_id so the sender can match a reply to its original request.

   from praisonaiagents.messaging import InProcessMailbox
   
   mailbox = InProcessMailbox()
   
   # Researcher asks the writer a question, tagged with a correlation id
   mailbox.send("writer", "Need a title for the RAG post", sender="researcher", correlation_id="q1")
   
   # Writer reads the request and replies with the same correlation id
   for req in mailbox.receive("writer"):
       mailbox.send(req.sender, "Ten Papers, One Retriever", sender="writer", correlation_id=req.correlation_id)
   
   # Researcher filters its inbox for the matching reply
   replies = [m for m in mailbox.receive("researcher") if m.correlation_id == "q1"]
   print(replies[0].body)

How It Works

The sender enqueues a message in the recipient’s inbox; the recipient either receives a push callback or pulls the queue.

Each recipient has its own bounded FIFO inbox. send appends a message and fires any subscriber callbacks; receive drains the oldest messages and removes them from the inbox.

Choosing your delivery pattern

Pick pull, push, or both based on how the recipient should react.

Mailbox vs. bus / kanban / handoff

Reach for a mailbox when you need named agent → named agent delivery — the one thing the other primitives don’t offer.

Configuration Options

InProcessMailbox takes a single constructor parameter.

from praisonaiagents.messaging import InProcessMailbox

mailbox = InProcessMailbox(max_inbox=100)

AgentMessage

Every delivered message is an AgentMessage dataclass.

Use msg.to_dict() and AgentMessage.from_dict(data) to round-trip a message through a plain dict.

AgentMailboxProtocol

Any object with these three methods satisfies the mailbox contract — implement it to back a mailbox with your own store.

The default InProcessMailbox adds one method beyond the protocol:

Common Patterns

Fan-out to multiple recipients

Address the same payload to several named agents in one loop.

from praisonaiagents.messaging import InProcessMailbox

mailbox = InProcessMailbox()

for agent_id in ("writer", "editor", "reviewer"):
    mailbox.send(agent_id, "Draft v2 is ready", sender="researcher")

print(mailbox.pending("editor"))  # 1

Request/response with correlation_id

Tag a request with a UUID, reply with the same UUID, and filter the sender’s inbox by it.

import uuid
from praisonaiagents.messaging import InProcessMailbox

mailbox = InProcessMailbox()
cid = str(uuid.uuid4())

mailbox.send("writer", "What's the headline?", sender="researcher", correlation_id=cid)

for req in mailbox.receive("writer"):
    mailbox.send(req.sender, "RAG in Practice", sender="writer", correlation_id=req.correlation_id)

reply = next(m for m in mailbox.receive("researcher") if m.correlation_id == cid)
print(reply.body)  # RAG in Practice

Overflow-safe queue for a slow worker

Bound the inbox so a slow consumer never grows unbounded — oldest messages drop silently.

from praisonaiagents.messaging import InProcessMailbox

mailbox = InProcessMailbox(max_inbox=100)

for i in range(150):
    mailbox.send("slow-worker", f"job-{i}", sender="dispatcher")

# Only the newest 100 survive — jobs 0–49 were dropped silently.
print(mailbox.pending("slow-worker"))  # 100

Best Practices

Related