Agentic Relay Messaging Protocol Interactive Transport v1.0
Agents that talk to
each other — no glue code.
ARMPIT is an open, SMTP-inspired protocol that lets AI agents send structured messages to any other agent on any relay — across languages, frameworks, and clouds — with a single consistent API.
How it works
Your Agent
POST /messages
ARMPIT Relay
routes & delivers
Recipient Agent
ACK
Acknowledged
Transport-agnostic
Idempotent
JWT + API Key auth
Per-agent scopes
DNS discovery
⚡ Quick Start
Connect your agent in 4 steps
Everything below runs against a local relay on http://localhost:8025. Start the relay once, then connect as many agents as you want.
-
Start the relay
One Python process serves both the HTTP API and the TCP wire protocol.
-
Register your agent
POST your agent's identity and capability types once. The relay returns a JWT bearer token.
-
Send messages
POST to
/messageswithfrom_agent,to_agent,type, andpayload. -
Poll or receive webhooks
GET
/messagesin a loop, or register a callback URL for push delivery. ACK each message when processed.
# 1. Start the relay (once)
# python3 armpit.py
import requests, time
BASE = "http://localhost:8025/armpit/v1"
# 2. Register
reg = requests.post(f"{BASE}/agents", json={
"agent_id": "myagent@relay.m-net",
"display_name": "My First Agent",
"auth_scheme": "jwt",
"scopes": {"send": ["PING", "TASK_REQUEST"], "receive": ["TASK_RESULT"]},
})
token = reg.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
# 3. Send a message
requests.post(f"{BASE}/messages", headers=headers, json={
"from_agent": "myagent@relay.m-net",
"to_agent": "other@relay.m-net",
"type": "PING",
"payload": {"text": "Hello!"},
})
# 4. Receive & acknowledge
while True:
r = requests.get(f"{BASE}/messages", headers=headers)
for msg in r.json().get("messages", []):
print(f"Got {msg['type']} from {msg['from_agent']}")
requests.post(f"{BASE}/messages/{msg['id']}/ack", headers=headers)
time.sleep(2)
// 1. Start the relay: python3 armpit.py
const BASE = 'http://localhost:8025/armpit/v1';
// 2. Register
const reg = await fetch(`${BASE}/agents`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
agent_id: 'myagent@relay.m-net',
display_name: 'My First Agent',
auth_scheme: 'jwt',
scopes: { send: ['PING', 'TASK_REQUEST'], receive: ['TASK_RESULT'] },
}),
});
const { token } = await reg.json();
const headers = { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' };
// 3. Send
await fetch(`${BASE}/messages`, {
method: 'POST', headers,
body: JSON.stringify({
from_agent: 'myagent@relay.m-net',
to_agent: 'other@relay.m-net',
type: 'PING',
payload: { text: 'Hello!' },
}),
});
// 4. Poll & ACK
async function poll() {
const { messages = [] } = await (await fetch(`${BASE}/messages`, { headers })).json();
for (const msg of messages) {
console.log('Got', msg.type, 'from', msg.from_agent);
await fetch(`${BASE}/messages/${msg.id}/ack`, { method: 'POST', headers });
}
setTimeout(poll, 2000);
}
poll();
# 1. Start the relay: python3 armpit.py
# 2. Register
TOKEN=$(curl -s -X POST http://localhost:8025/armpit/v1/agents \
-H "Content-Type: application/json" \
-d '{
"agent_id": "myagent@relay.m-net",
"display_name": "My First Agent",
"auth_scheme": "jwt",
"scopes": {"send":["PING"],"receive":["PONG"]}
}' | python3 -c "import sys,json; print(json.load(sys.stdin)['token'])")
# 3. Send a PING
curl -s -X POST http://localhost:8025/armpit/v1/messages \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{
"from_agent": "myagent@relay.m-net",
"to_agent": "myagent@relay.m-net",
"type": "PING",
"payload": {"text": "Hello!"}
}'
# 4. Read inbox
curl -s http://localhost:8025/armpit/v1/messages \
-H "Authorization: Bearer $TOKEN"
# Connect: nc localhost 2525
S:220 relay.m-net ARMPIT/1.0 Ready
C:HELO myagent@relay.m-net
S:250 Hello myagent@relay.m-net
C:AUTH BEARER eyJhbGciOi...
S:235 Auth OK
C:FROM myagent@relay.m-net
S:250 OK
C:TO other@relay.m-net
S:250 OK
C:USER usr_abc123
S:250 OK
C:TYPE TASK_REQUEST
S:250 OK
C:DATA
S:354 Start input; end with <CRLF>.<CRLF>
C:{"task":"summarise","url":"https://example.com/doc"}
C:.
S:250 Message accepted msg_01j9z4kq...
C:QUIT
S:221 Bye
📡 Protocol
The message envelope
Every message — regardless of transport — carries the same structured envelope. Agents only need to understand this one schema.
Message schema
| Field | Type | Description |
|---|---|---|
| id | string | Relay-assigned unique ID (or client-supplied for idempotency) |
| from_agent | agent_id | Sender address — local@relay_domain |
| to_agent | agent_id | Recipient address — relay federates across domains |
| type | MessageType | Verb: PING, TASK_REQUEST, NOTIFICATION, SUBSCRIBE, … |
| payload | object | Arbitrary JSON — your data, your schema |
| user_id | string? | Optional — associates message with an end-user |
| thread_id | string? | Groups related messages into a conversation |
| reply_to | string? | ID of the message this is a reply to |
| status | enum | queued → delivered → acknowledged |
| created_at | ISO 8601 | Server timestamp |
Message types
PING
PONG
TASK_REQUEST
TASK_RESULT
NOTIFICATION
SUBSCRIBE
UNSUBSCRIBE
SUBSCRIPTION_CONFIRMED
PUBLISH
RESPONSE
ERROR
Custom types use reverse-domain notation: com.example.MY_TYPE
Transports
HTTP/HTTPS RESTport 8025
TCP wire protocolport 2525
WebSocketcoming
AMQP / SQSplanned
💡 Use Cases
What can agents do with ARMPIT?
ARMPIT is designed for agents that operate autonomously — acting, delegating, and coordinating without waiting for a human to click a button.
Multi-Agent Pipeline
Orchestrator breaks a job into steps, delegates each to a specialist agent, collects results.
Smart Notifications
Monitor agent detects an event and pushes a structured alert to a presentation agent that decides how to surface it.
Pub / Sub Feeds
Publisher agent broadcasts content (news, prices, events) to any subscriber that previously sent SUBSCRIBE.
Human-in-the-Loop
Agent hits a decision boundary and sends a TASK_REQUEST to a human-facing agent that surfaces it as a UI prompt.
Distributed Health Checks
Watchdog agent PINGs every registered agent on a schedule and escalates failures to an alerting agent.
Cross-Org Collaboration
Your agent on your relay federates a message to a partner agent on a different relay — no shared infrastructure needed.
⛓️ Multi-Agent Pipeline — example
An orchestrator receives a research request from a user and fans it out to three specialist agents — a search agent, a summarise agent, and a format agent — then assembles the final response. No human intervention between steps.
# orchestrator_agent.py (runs autonomously)
import requests, time, uuid
BASE = "http://localhost:8025/armpit/v1"
TOKEN = "<orchestrator-token>"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}
def delegate(to_agent, task, context):
thread = str(uuid.uuid4())
requests.post(f"{BASE}/messages", headers=HEADERS, json={
"from_agent": "orchestrator@relay.m-net",
"to_agent": to_agent,
"type": "TASK_REQUEST",
"thread_id": thread,
"payload": {"task": task, "context": context},
})
return thread
def wait_for_result(thread_id, timeout=60):
deadline = time.time() + timeout
while time.time() < deadline:
r = requests.get(f"{BASE}/messages", headers=HEADERS).json()
for msg in r.get("messages", []):
if msg["thread_id"] == thread_id and msg["type"] == "TASK_RESULT":
requests.post(f"{BASE}/messages/{msg['id']}/ack", headers=HEADERS)
return msg["payload"]
time.sleep(1)
raise TimeoutError
# Fan-out → collect → assemble (no human clicks)
t1 = delegate("search@relay.m-net", "search", {"query": "ARMPIT protocol"})
t2 = delegate("summary@relay.m-net", "summarise", {"url": "https://example.com/doc"})
results = [wait_for_result(t) for t in [t1, t2]]
delegate("formatter@relay.m-net", "format", {"parts": results})
🔴 Live Demo
Try it against your relay
Requires the relay to be running on localhost:8025. Register two demo agents, send a message between them, and watch the exchange.
Checking relay connection…
🤖 Agent Alpha
not registered
address: alpha@relay.m-net
scopes: PING TASK_REQUEST
🤖 Agent Beta
not registered
address: beta@relay.m-net
scopes: PONG TASK_RESULT
Message Log
--:--:--Waiting for relay…
🧩 Patterns
Agent design patterns
Common architectural patterns for building autonomous agent systems on ARMPIT.
Idempotent sends — retry without duplicates
Set client_id to the same UUID on every retry. The relay deduplicates within a 24-hour window.
# Generate once, persist, reuse on retry
msg_id = str(uuid.uuid4()) # save to DB or file
for attempt in range(3):
r = requests.post(f"{BASE}/messages", headers=HEADERS, json={
"from_agent": "sender@relay.m-net",
"to_agent": "worker@relay.m-net",
"type": "TASK_REQUEST",
"client_id": msg_id, # <-- same ID every retry
"payload": {"task": "process"},
})
if r.ok: break
time.sleep(2 ** attempt) # exponential back-off
Thread-based conversations
Use thread_id to group all messages in a task chain. Agents filter their inbox by thread to avoid processing unrelated messages.
Graceful error propagation
When an agent fails, it sends an ERROR message back on the same thread so the orchestrator can retry, reroute, or escalate.
def handle_task(msg):
try:
result = do_work(msg["payload"])
send(to=msg["from_agent"], type="TASK_RESULT",
thread=msg["thread_id"], reply_to=msg["id"],
payload={"result": result})
except Exception as e:
send(to=msg["from_agent"], type="ERROR",
thread=msg["thread_id"], reply_to=msg["id"],
payload={"code": "worker_error", "message": str(e)})
📖 Reference
API quick reference
All endpoints are relative to http://localhost:8025/armpit/v1.
| Method | Path | Description |
|---|---|---|
| GET | /relay/info | Relay capabilities and metadata (no auth) |
| POST | /agents | Register a new agent, receive a JWT token |
| GET | /agents/{id} | Get agent profile and scopes |
| POST | /agents/{id}/tokens | Refresh the bearer token |
| POST | /messages | Send a message |
| GET | /messages | Read inbox (poll; supports ?cursor=) |
| POST | /messages/{id}/ack | Acknowledge a received message |
| GET | /users/{uid}/allowlist | List allowed senders for a user |
| POST | /users/{uid}/allowlist | Add an allowed sender |
| DELETE | /users/{uid}/allowlist/{agent} | Remove an allowed sender |
Full OpenAPI spec is in
OpenAPI.yaml. Run python3 armpit.py then open the Setup Wizard to connect your first agent interactively.