agent-comms

Agents that can message each other, and stop to ask a human a question.

Two channels, nothing more:

• Every agent has one private inbox. It holds messages from peer agents and typed events from an orchestrator. One queue per agent, read only by that agent.

• An agent can ask a human and block until answered. The call returns what the human said.

agent-comms is a communication medium. It transports messages and delivers events. It does not run your agent loop and never claims to control it.

Nothing here can stop, preempt, or interrupt a running agent. An agent that never reads its inbox is unaffected by anything put in it — honoring what arrives is the consumer's job. That boundary is what keeps the library small and its claims defensible.

Install

Not on PyPI — install straight from a tagged release:

pip install "agent-comms[mcp] @ git+https://github.com/Abhishekq10/agent-comms.git@v0.1.0"

Or run the broker with no install at all:

uvx --from "agent-comms[mcp] @ git+https://github.com/Abhishekq10/agent-comms.git@v0.1.0" \
    agent-comms-broker

A built wheel and sdist are attached to every release if you would rather not build from source.

The mcp extra adds the agent surface. Without it the broker still serves REST — run it with --no-mcp, or create_app(mcp=False).

Related MCP server: mcp-server-agent-comm

Try it

Two runnable examples, no setup:

python examples/multi_agent_demo.py    # three agents coordinating over inboxes
python examples/hitl_demo.py --auto    # an agent blocks on a human, gets answered

$ python examples/multi_agent_demo.py
coordinator: handing out work
coordinator: waiting for reports
  [worker-b] got assignment: batch 2
  [worker-a] got assignment: batch 1
  [coordinator] worker-a says: batch 1 complete
  [coordinator] worker-b says: batch 2 complete
coordinator: releasing workers
  [worker-a] shutting down
  [worker-b] shutting down

final inbox state:
  coordinator: 2 entries, 0 unread
  worker-a: 2 entries, 0 unread
  worker-b: 2 entries, 0 unread

The two workers run on real threads, so which one reports first varies between runs — that interleaving is the point.

$ python examples/hitl_demo.py --auto
broker on http://127.0.0.1:6123

agent asked: "Deploy to prod or staging?"
  context: CI is green on main; 3 commits since the last release.
  ...and is now blocked, waiting.

answer it from another terminal:

  curl http://127.0.0.1:6123/questions
  curl -X POST http://127.0.0.1:6123/questions/cb1d305b27d1/answer \
       -H 'content-type: application/json' \
       -d '{"answer":"staging","answered_by":"alice"}'

(--auto: answering it for you)

agent unblocked after 0.5s with: "staging" (from alice)

Without --auto it prints the same curl and waits — the blocked agent returns the moment you run it.

What's inside

The inbox

from agent_comms import MailboxRegistry

MailboxRegistry.create_mailbox("worker-a")

# a message from a peer agent
MailboxRegistry.send("coordinator", "worker-a", "batch 3 is yours", type="assignment")

# a typed event — no human sender, structured body
MailboxRegistry.notify("worker-a", "CONFIG_CHANGED", {"key": "retries"})

# cheap check at an iteration boundary, consumes nothing
count, summary = MailboxRegistry.peek_inbox("worker-a")   # (2, "coordinator (1), CONFIG_CHANGED (1)")

# read and mark read
for entry in MailboxRegistry.check_inbox("worker-a"):
    if entry.sender:
        handle_message(entry.sender, entry.content)
    else:
        handle_event(entry.type, entry.payload)

type is a free-form label. The library attaches no meaning to any value — there is no built-in PAUSE or SHUTDOWN, because what those should do is your decision, not the library's.

Inboxes are private by addressing. A delivery is stored in the recipient's inbox, and check_inbox(name) reads only that one. There is no global stream.

Asking a human

The agent blocks; the call returns the answer.

from agent_comms import ask_human

answer = ask_human("agent-1", "Deploy to prod or staging?", timeout=300)

if answer.timed_out:
    ...                    # nobody replied — proceed on your own judgment
else:
    print(answer.text)     # "staging"

Meanwhile, on the human side:

curl localhost:6123/questions
# {"questions":[{"question_id":"a1b2c3d4e5f6","agent":"agent-1",
#                "text":"Deploy to prod or staging?","answer":null,...}]}

curl -X POST localhost:6123/questions/a1b2c3d4e5f6/answer \
     -H 'content-type: application/json' \
     -d '{"answer":"staging","answered_by":"alice"}'

The blocked ask_human() returns the moment that POST lands.

sequenceDiagram
    participant A as agent
    participant B as broker
    participant H as human

    A->>B: ask_human("Deploy to prod or staging?")
    Note over B: question registered,<br/>answerable immediately
    B--)A: blocked

    H->>B: GET /questions
    B-->>H: question_id, text, context
    H->>B: POST /questions/ID/answer — answer "staging"
    Note over B: first answer wins —<br/>a second gets 409

    B-->>A: "staging"
    Note over A: unblocks the moment<br/>the POST lands

The question is registered before anything waits on it, which is why an answer arriving in that gap isn't lost.

No notification backend ships here, by design. GET /questions is the interface — poll it, or wire it into whatever you already run. Embedded users who prefer a push can pass a callback:

from agent_comms import HumanChannel

channel = HumanChannel(on_question=lambda q: notify_slack(q.text))

Answering is first-write-wins: a second answer raises AlreadyAnsweredError (HTTP 409), and answering a question whose asker already timed out raises UnknownQuestionError (HTTP 404).

LLM tool schemas

MessagingTools produces JSON tool schemas and dispatches the calls. Identity is bound at construction, so check_messages takes no arguments — the model cannot name another agent's inbox.

from agent_comms import MessagingTools

tools = MessagingTools(agent_name="worker-a", peers=["coordinator", "worker-b"])

schemas = tools.schemas()   # send_message, check_messages, list_peers

tools.dispatch("send_message", {
    "to": "coordinator",
    "message": "batch 3 complete",
    "type": "progress_update",
})

peers also takes a callable, for a roster that changes while the agent runs:

MessagingTools("worker-a", peers=lambda: registry.alive())

MCP — the agent surface

Run the broker, then point an MCP client at it. Identity comes from the connection, never from an argument — the agent name is a header, so a model cannot reach another agent's inbox by hallucinating a parameter. check_messages takes no arguments at all.

agent-comms-broker                      # REST + MCP on 127.0.0.1:6123

claude mcp add --transport http agent-comms http://127.0.0.1:6123/mcp/ \
    --header "X-Agent-Name: worker-a"

Add --header "X-Supervisor: 1" to unlock notify_agent. It carries no sender, so it reads as coming from the system rather than a peer — which is why ordinary agents cannot call it. It still only delivers; the recipient decides what the event means.

MCP is mounted on the same app as REST, so agents and humans share one broker holding one set of inboxes. A stdio-spawned server would get its own process and its own empty registry, which is why this is streamable HTTP.

Give each agent its own name and they can reach each other:

claude mcp add --transport http coord http://127.0.0.1:6123/mcp/ \
    --header "X-Agent-Name: coordinator" --header "X-Supervisor: 1"

HTTP endpoints

Served by start_api_server(port=6123) or agent-comms-broker.

One delivery endpoint covers both kinds — sender present means a peer message, absent means a typed event:

curl -X POST localhost:6123/agents/worker-a/messages \
     -d '{"sender":"coordinator","content":"batch 3 is yours","type":"assignment"}'

curl -X POST localhost:6123/agents/worker-a/messages \
     -d '{"type":"CONFIG_CHANGED","payload":{"key":"retries"}}'

Binding. start_api_server() binds 127.0.0.1 by default. This broker mediates human answers and ships no authentication, so exposing it beyond loopback is a deliberate act — pass host="0.0.0.0" explicitly, behind something that authenticates.

Extension points

# mask secrets before a human sees the question
class TextSanitizer(Protocol):
    def sanitize(self, text: str) -> str: ...

HumanChannel(sanitizer=MySecretMasker())

# push instead of poll
HumanChannel(on_question=lambda q: notify_slack(q.text))

Architecture

flowchart TB
    agents["agents<br/>worker-a · worker-b · …"]
    humans["humans &amp; ops tooling<br/>UI · CLI · curl"]
    embedded["embedded callers<br/>same process, no network"]

    subgraph broker["one process · one FastAPI app"]
        mcp["MCP surface<br/>/mcp"]
        rest["REST surface<br/>/agents/… · /questions"]
        inbox[("MailboxRegistry<br/>one inbox per agent")]
        human[("HumanChannel<br/>questions awaiting an answer")]

        mcp --> inbox
        mcp --> human
        rest --> inbox
        rest --> human
    end

    agents -- "identity from<br/>X-Agent-Name header" --> mcp
    humans -- "poll and answer" --> rest
    embedded --> inbox
    embedded --> human

Both surfaces are mounted on one FastAPI app in one process, over one set of inboxes. Embedded users skip the network entirely and call MailboxRegistry / ask_human() directly — same store, no server.

Design notes

Most of what's interesting here is what the library doesn't do.

It delivers; it never acts. There is no PAUSE, no SHUTDOWN, no built-in signal vocabulary — type is a free-form string the library attaches no meaning to. An earlier version shipped a separate signal bus with pause() / resume() and a POST /pause endpoint, and it was the one place the library reached in and changed a consumer's state. Deleting it made the boundary claim literally true instead of true-with-an-asterisk. If you want PAUSE semantics, deliver {"type": "PAUSE"} and decide for yourself what honoring it means.

One queue per agent, not two. Messages from peers and typed events used to live in separate structures with separate APIs. They're the same thing — an entry addressed to one agent — so they share one inbox, distinguished by whether sender is set. That removed a module, five concepts, and the question of which queue a given thing belongs in.

Identity comes from the connection, never from an argument. Over MCP the agent name is a header; check_messages takes no parameters at all, so a model cannot read another agent's inbox by hallucinating one. Note this is addressing plus tool-shape, not enforcement — anything with in-process access, or the REST surface, can reach any inbox. That's why the default bind is loopback.

No notification backend. GET /questions is the interface. Shipping Slack or webhook delivery would mean owning integrations and their failure modes forever, to save consumers a poll loop they can write in five lines.

ask_human blocks and returns the answer. The obvious alternative — fire a notification and let the caller figure out when a reply arrives — pushes the hard part (correlating answer to question, handling the race where the answer lands before you start waiting) onto every consumer. ask() and wait() are split internally precisely so a question is answerable before anyone waits on it.

In-memory, deliberately. Inboxes and pending questions live in the process. Adding persistence would mean picking a store, owning migrations, and defending a durability story. Consumers who need it hook on_question or poll /questions/history and write wherever they already write.

License

Apache-2.0. See LICENSE and NOTICE.

Contributions welcome — see CONTRIBUTING.md.