PBXMCP

A telephone exchange (PBX) for LLM agents, exposed as an MCP server.

Each system gets an area code and each agent an extension. Agents place calls to each other with dial and reply with answer_call; the exchange routes the call, enforces a hop budget so call chains can't loop forever, and records every call in a ledger.

PBXMCP owns the protocol, routing, hop accounting, and MCP surface. The actual "ring the callee" step is host-specific — a kernel that pushes a turn to an agent's stdin, a message queue, an HTTP webhook — so delivery goes through a pluggable transport. A working in-memory loopback transport ships for tests and demos; a real host implements the same small contract.

Why

An agent normally can't talk to another agent. If a host can push a message to an agent (deliver it as that agent's next turn), then one agent calling another is just: route the call, deliver it as a turn, and carry a correlation id so the reply finds its way back. PBXMCP is that routing-and-correlation layer, packaged as an MCP server so any MCP-speaking agent can dial.

The receiver never has to "listen" on an open connection — it is rung by the host when it is ready for its next turn — so there is no socket to hold open and no per-call timeout on the callee's side.

Quickstart

pip install -e ".[server]"          # server extra pulls in fastmcp
python -m pbxmcp serve              # stdio MCP server
python -m pbxmcp serve --transport http --port 7420

Point an MCP client at it and call dial:

{"from_extension": "alice", "to_extension": "212-bob", "body": "status?"}

dial returns immediately with a run_id and a status (delivered / unknown_target / ttl_exceeded / failed). The reply arrives later as a separate turn on the caller; recognize it with the telephone envelope and answer with answer_call(reply_to_run_id, body).

Use it as a library

The exchange works with no MCP dependency at all:

from pbxmcp import Exchange, RoutingConfig, CallLedger, LoopbackTransport

transport = LoopbackTransport()
transport.register("alice", "bob")
exchange = Exchange(RoutingConfig(), transport, CallLedger())

call = exchange.dial(from_extension="alice", to_extension="bob", body="ping")
# bob's inbox now holds the call envelope:
env = transport.inbox("bob")[0]
exchange.answer(from_extension="bob", reply_to_run_id=call.run_id, body="pong")
# alice's inbox now holds the reply.

Tools

Tool	Purpose
dial(from_extension, to_extension, body, to_area?, ttl_hops?)	Place a call. Fire-and-forget; returns a run_id.
answer_call(from_extension, reply_to_run_id, body, status?)	Reply to a call you received.
list_calls()	Every call the exchange has recorded.
directory()	The area-code map and dialing defaults.

Docs

• docs/architecture.md — exchange, transport, ledger, hop budget.
• docs/protocol.md — the wire protocol: envelope shape, tool schemas, statuses.
• docs/roadmap.md — status and what's deferred (real transports, federation, auth).

Status

V0.1 — the core (routing, hop budget, ledger, envelope) is implemented and tested; the MCP server binds it via fastmcp; the loopback transport proves the mechanism end to end. Production delivery needs a host transport adapter (see the roadmap). 33 tests, no network required for the core suite.

License

MIT. See LICENSE.