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_status to get its current cursor.
Call poll with 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 — use get_or_create_agent instead of create_agent.
ValueError: sender not in thread — add the sender with join_thread first.
get_thread_info(...) is None — the thread is missing from the selected store or database path.
has_unread(...) is True after polling — call mark_read with 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/