agent-mesh

Why It Exists

Most multi-agent demos are synchronous chains: one call waits for the next call, and one slow or failing agent stalls or voids the whole run. They often have no retry boundary, no isolation between specialist roles, and no reliable way to reconstruct what happened after a bad run.

agent-mesh is an event-driven backend for coordinating specialist agents through a small, explicit event contract. Work is submitted as a job, decomposed into role-specific tasks, processed by independent workers, and joined by an aggregator that can complete with partial results when a specialist fails permanently.

The default provider is fake, so the full test suite and demo run with no API keys.

Architecture

submit
  |
  v
JOB_SUBMITTED
  |
  v
supervisor
  |
  v
TASK_REQUESTED fan-out
  |                         |
  v                         v
researcher worker       analyst worker
  |                         |
  v                         v
TASK_SUCCEEDED/FAILED   TASK_SUCCEEDED/FAILED
  \                         /
   \                       /
    v                     v
        aggregator fan-in
              |
              v
      JOB_COMPLETED or JOB_FAILED

TASK_REQUESTED retry budget exhausted
              |
              v
        DEAD_LETTER
              |
              v
      dead-letter sweeper
              |
              v
         TASK_FAILED

Event Flow

Supervisor.submit(goal) publishes JOB_SUBMITTED with the job id as the correlation id. The supervisor consumes that event, stores the job state, and publishes one TASK_REQUESTED event per role. Each worker is in its own consumer group, so every role receives a copy of the task stream while work remains exclusive within that role. Workers publish TASK_SUCCEEDED or TASK_FAILED. The aggregator consumes both, joins by job id, and publishes JOB_COMPLETED or JOB_FAILED once every expected role has settled.

Every child event preserves the same correlation id and records the causation id of the event that produced it. GET /jobs/{job_id}/events returns the causal trace for debugging.

Reliability Semantics

The bus is a Protocol with two implementations. InMemoryBus is the default used by the examples and the ordinary test suite. RedisStreamsBus provides the same contract on Redis Streams for real deployments and is available with:

pip install -e .[redis]

tests/test_bus_contract.py runs one conformance suite against both implementations, using fakeredis for the Redis-backed case, so the abstraction is verified rather than assumed.

Guarantee	Mechanism
Fan-out by role	One topic copied to each subscribed consumer group
Exclusive work within a role	A pending set per topic and group
At-least-once delivery	Messages remain pending until acked
Crash recovery	Visibility-timeout reclaim returns unacked work to the queue
Idempotency	Bus-level seen set skips already-acked event ids
Retry budget	nack redelivers until max_deliveries is reached
Poison-message isolation	Exhausted messages move to DEAD_LETTER
Joins do not hang	Dead-letter sweeper converts poison tasks into TASK_FAILED
Partial failure	Aggregator marks jobs degraded when some roles fail permanently
Traceability	Correlation and causation ids on every event

Providers

FakeProvider is deterministic and needs no credentials. It can cycle through fixed responses, fail transiently for a configured number of calls, or fail permanently for degradation-path tests.

SDK-backed providers read credentials from the process environment, or from a gitignored .env file loaded with agent_mesh.config.load_dotenv(). Values already present in the real environment always win over .env, so platform-injected secrets are not overridden by a stale file baked into an image. describe_credentials() reports whether credentials are present without ever revealing a value.

OpenAIProvider uses the official OpenAI SDK and reads OPENAI_API_KEY when an API key is not passed directly. OpenAIProvider.DEFAULT_MODEL is gpt-4o-mini, overridable with OPENAI_MODEL or a model= argument.

GrokProvider uses the same SDK against xAI's OpenAI-compatible endpoint, https://api.x.ai/v1, so both vendors share one client path. It reads XAI_API_KEY, falling back to GROK_API_KEY. GrokProvider.DEFAULT_MODEL is grok-4.5, overridable with XAI_MODEL or a model= argument.

xAI retires model ids aggressively and does not keep bare-major aliases alive. A plausible-looking id such as grok-3 returns 404 at the first live call rather than failing at import. The live test suite therefore asserts that each provider default model id appears in that provider's GET /v1/models listing.

Real keys must never be committed. .env is gitignored.

MCP Integration

agent_mesh.mcp_server.create_mcp_server(supervisor, store, mesh) exposes three MCP tools:

Tool	Result
submit_job(goal)	Submits a goal, drains the mesh, and returns the job id
get_job(job_id)	Returns job status and output
list_jobs()	Returns known jobs

Install MCP support with:

pip install 'agent-mesh[mcp]'

Quickstart

pip install -e .[dev]
python examples/research_brief.py

To use SDK-backed providers:

pip install -e .[llm]
export OPENAI_API_KEY=...
export XAI_API_KEY=...

Running against live models

examples/live_brief.py runs the full mesh with Grok serving the researcher role and GPT serving the analyst role in the same job, which is the point of the provider abstraction.

python examples/live_brief.py

An optional goal can be passed as command line arguments. For the no-credentials equivalent, run examples/research_brief.py, which uses the deterministic fake provider.

HTTP API

Create an app with agent_mesh.api.create_app(bus, store, supervisor, mesh).

Method	Path	Description
GET	/healthz	Health check
POST	/jobs	Submit { "goal": "..." }; returns 202 and job_id
GET	/jobs	List all jobs
GET	/jobs/{job_id}	Get status, output, results, and failures
GET	/jobs/{job_id}/events	Get the full causal event trace
GET	/dead-letters	Inspect dead-lettered messages

Testing

ruff check .
pytest -q

The default suite has 26 tests and uses FakeProvider, so it runs with no credentials. That keeps CI green.

Live integration tests are opt-in and skipped otherwise:

AGENT_MESH_LIVE=1 pytest tests/test_live_providers.py

Both tiers are needed. Unit tests with a fake provider structurally cannot catch a retired model id, a changed base URL, or a moved auth scheme. The first live run of this suite caught exactly that class of bug.

Design Notes

The in-memory bus is the reference implementation and mirrors production queue semantics closely enough to make failure behavior testable. A deployment can replace it with Redis Streams, Kafka, or another durable transport while keeping the event envelope and component APIs stable.

The worker treats permanent provider failures as settled task failures and lets transient failures propagate to the bus retry path. That split keeps policy out of provider adapters and prevents a single bad request from blocking a fan-in.