# 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:

```bash
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:

```bash
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](https://github.com/Abhishekq10/agent-comms/releases) 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)`.

## Try it

Two runnable examples, no setup:

```bash
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

| Module | Purpose |
|--------|---------|
| `mailbox` | One thread-safe inbox per agent: peer messages and typed events |
| `hitl` | Agent asks a human and blocks; human answers; agent resumes |
| `messaging_tools` | LLM-ready tool schemas + dispatch over the inbox |
| `mcp_server` | MCP tools over streamable HTTP — the agent surface |
| `api` | FastAPI server exposing both channels — the human surface |

## The inbox

```python
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.

```python
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:

```bash
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.

```mermaid
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:

```python
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.

```python
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:

```python
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.

```bash
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"
```

| Tool | What it does |
|------|--------------|
| `send_message(to, message, type)` | Deliver to another agent's inbox |
| `check_messages()` | Read *your* unread entries, mark them read |
| `peek_inbox()` | Unread count, consumes nothing |
| `list_peers()` | Agents you can message |
| `ask_human(question, …)` | Block until a human answers, return the answer |
| `notify_agent(agent, type, payload)` | Deliver a typed event — **supervisor only** |

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:

```bash
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`.

| Method | Path | Purpose |
|--------|------|---------|
| POST | `/agents/{name}` | Create an inbox |
| DELETE | `/agents/{name}` | Archive an inbox |
| POST | `/agents/{name}/messages` | Deliver a message or typed event |
| GET | `/agents/{name}/messages` | Everything in the inbox (consumes nothing) |
| POST | `/agents/{name}/messages/check` | Unread, and mark them read |
| GET | `/agents/{name}/messages/peek` | Unread count, nothing consumed |
| GET | `/mailboxes` | Every known inbox |
| GET | `/questions` | Questions awaiting a human |
| GET | `/questions/history` | Answered questions |
| POST | `/questions/{id}/answer` | Answer one, unblocking the agent |
| GET | `/status` | Broker summary |

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

```bash
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

```python
# 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

```mermaid
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](LICENSE) and [NOTICE](NOTICE).

Contributions welcome — see [CONTRIBUTING.md](CONTRIBUTING.md).