7.3 KiB
AgentMsgs
A small message bus for coordinating short-lived agents and tools.
Use it when independent Python processes or MCP clients need to:
- create shared threads;
- send ordered messages;
- track unread messages per agent;
- inspect thread activity;
- persist coordination state in SQLite.
AgentMsgs has no server process and no authentication layer. It can run entirely in memory for tests, or use a shared SQLite file for coordination between processes.
Quick start
Requirements: Python 3.14+ and uv.
uv sync
Python
from agentmsgs.core.app import App
from agentmsgs.stores.memory import InMemoryStore
app = App(store=InMemoryStore())
alice = app.get_or_create_agent("Alice")
bob = app.get_or_create_agent("Bob")
thread = app.create_thread(alice, bob)
app.append(thread.id, alice, "Hello from Alice")
messages = app.poll(thread.id)
for message in messages:
print(message.seq, message.sender.name, message.content)
app.mark_read(thread.id, bob, messages[-1].seq)
assert app.has_unread(thread.id, bob) is False
InMemoryStore disappears when the process exits.
For persistent storage:
from pathlib import Path
from agentmsgs.core.app import App
from agentmsgs.stores.sqlite import SQLiteStore
app = App(store=SQLiteStore(Path(".agentmsgs.db")))
MCP server
Start the MCP server over stdio with an explicit SQLite path:
uv run agentmsgs-mcp --transport stdio --db-path ./agentmsgs.db
Or use HTTP:
uv run agentmsgs-mcp --transport http --db-path ./agentmsgs.db
Choose a shared database file when separate clients or processes should communicate:
uv run agentmsgs-mcp --db-path ./shared-agentmsgs.db --transport stdio
For a temporary, process-local server, use the in-memory store:
uv run agentmsgs-mcp --in-memory --transport stdio
The in-memory mode is synchronized for concurrent calls within that server, but its data disappears on restart and is not shared with another server process. SQLite mode requires an explicit --db-path.
See all server options with:
uv run agentmsgs-mcp --help
Core concepts
Agent
An agent has a human-readable name and a stable UUID:
alice = app.get_or_create_agent("Alice")
get_or_create_agent reuses an existing name. create_agent raises ValueError if the name is already taken.
Thread
A thread groups at least two participants:
thread = app.create_thread(alice, bob)
Threads have their own UUIDs. Multiple threads may contain the same participants.
Message
Messages are ordered per thread:
message = app.append(thread.id, alice, "Starting work")
assert message.seq == 1
Only thread participants can append messages.
Read cursor
Each agent has a per-thread cursor identifying the last sequence it has read. Polling after a sequence avoids reprocessing old messages:
messages = app.poll(thread.id, after_seq=message.seq)
app.mark_read(thread.id, bob, message.seq)
A cursor of 0 means nothing has been marked as read.
Python API
App is the main Python interface:
| Method | Purpose |
|---|---|
get_or_create_agent(name) |
Get or create an agent by name. |
create_agent(name) |
Create an agent, rejecting duplicate names. |
create_thread(*agents) |
Create a thread with at least two agents. |
get_thread(thread_id) |
Fetch a thread or None. |
find_threads(*agents) |
Find threads containing all supplied agents. |
join_thread(thread_id, agent) |
Add a participant. |
leave_thread(thread_id, agent) |
Remove a participant. |
append(thread_id, sender, content) |
Append a message. |
poll(thread_id, after_seq=0) |
Return messages after a sequence. |
mark_read(thread_id, agent, seq) |
Advance an agent's read cursor. |
has_unread(thread_id, agent) |
Check whether messages remain unread. |
get_thread_info(thread_id) |
Get message count and latest activity. |
get_thread_read_state(thread_id, agent) |
Get an agent's cursor and unread counts. |
get_thread_info returns None for an unknown thread. get_thread_read_state returns None for an unknown thread and returns a valid state for an empty thread.
Example activity check:
info = app.get_thread_info(thread.id)
state = app.get_thread_read_state(thread.id, bob)
if info is not None and info.message_count:
print("Message count:", info.message_count)
if state is not None and state.unread_from_others_count:
print("Bob has unread messages from other agents")
MCP tools
The server exposes these tools:
| Tool | Purpose |
|---|---|
create_agent(name) |
Get or create an agent. |
delete_account(agent_id) |
Delete an agent. |
create_thread(participant_names) |
Create a thread from names. |
join_thread(thread_id, agent_name) |
Add a participant. |
thread_info(thread_id) |
Return participants, message count, and latest message. |
thread_status(thread_id, agent_name) |
Return cursor and unread counts. |
append(thread_id, sender_name, content) |
Append a message. |
poll(thread_id, after_seq) |
Poll messages after a sequence. |
has_unread(thread_id, agent_name) |
Check for unread messages. |
A typical MCP worker loop is:
- Create or find the worker agent.
- Receive or store the thread ID.
- Call
thread_statusto get its current cursor. - Call
pollwith that cursor. - Process the returned messages.
- Mark the last processed sequence as read.
- Repeat.
Storage and concurrency
InMemoryStore
Use it for tests, examples, and short-lived single-process workflows. Access is synchronized within one store instance, but state is local to the Python process and is not shared.
SQLiteStore
Use it when state must survive restarts or be shared between processes. The store uses SQLite tables for agents, threads, participants, messages, and read cursors, and enables WAL mode.
One SQLiteStore instance owns one SQLite connection. Separate processes should each create their own store and connection to the same database file. Do not share one store connection concurrently between threads without synchronization.
Derived reads such as thread information and read state use a read transaction so their values come from one consistent snapshot.
There is no automatic TTL or garbage collection. Delete agents and threads explicitly when they are no longer needed.
Common errors
ValueError: need >=2 participants— create a thread with at least two agents.ValueError: agent name taken— useget_or_create_agentinstead ofcreate_agent.ValueError: sender not in thread— add the sender withjoin_threadfirst.get_thread_info(...) is None— the thread is missing from the selected store or database path.has_unread(...) is Trueafter polling — callmark_readwith the last processed sequence.
Testing
uv run pytest -q
The test suite covers the Python API, both stores, persistence, concurrent sequence allocation, thread read models, read-snapshot consistency, and MCP payloads.
Layout
src/agentmsgs/
├── core/ # domain types, operations, store protocols, App
├── stores/ # InMemoryStore and SQLiteStore
├── mcp/ # FastMCP server
└── utils.py
examples/demo.py
tests/