agent-mesh

Synchronous agent chains stall when one specialist fails, so agent-mesh coordinates role-specific workers through an event contract with retries, dead letters, causal traces, and partial-result fan-in.

Run the offline proof

From a clone of this repository, the deterministic example needs no credentials or network access:

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

It runs two jobs: one completes with researcher and analyst results, and one finishes as degraded with the research result retained after a simulated permanent analyst failure.

Related MCP server: agent-orchestrator

What the tests prove

python -m pytest -q
ruff check .

The full local suite produced 74 passed, 4 skipped, 1 warning in 34.02s; Ruff reported All checks passed!. The passing tests exercise fan-out and exclusive consumer-group delivery, acknowledgements, retries, dead-letter exhaustion, visibility-timeout reclaim in memory, correlation/causation history, normal and degraded fan-in, HTTP endpoints, the common in-memory/Redis bus contract using fakeredis, and Redis-backed fan-in recovery across reconstructed store and aggregator objects.

They also cover the model-driven planner: a nacked and redelivered job re-plans to byte-identical task ids instead of doubling the fan-out, a role with no worker is refused where the error is legible rather than dead-lettering later, and a plan that fails to parse dead-letters the job instead of running it.

Not implemented / not proven

• The default suite uses FakeProvider; four live OpenAI/xAI tests were skipped, so the result above does not prove current provider credentials, model IDs, latency, or availability.

• Redis behavior is tested with fakeredis, not a real Redis server under network partitions, process crashes, or production load.

• Delivery is intentionally at-least-once, not exactly-once; consumers must keep their own side effects idempotent.

• There is no published throughput, latency, or scale benchmark.

• The MCP adapter exists, but the default suite has no end-to-end MCP transport test.

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.

Planning the fan-out with a model

default_decomposer is a fixed three-role split. LLMDecomposer lets a model plan it instead — and the interesting part is not the prompt, it is that delivery here is at-least-once.

Supervisor.run_once decomposes on every delivery of JOB_SUBMITTED, so the same goal gets planned more than once in normal operation: after a nack, after a crash between publish and ack, after a consumer-group rebalance. TaskSpec generates a random task_id by default, so even the built-in decomposer republishes tasks under fresh identities on redelivery. Point that path at a model whose wording varies between calls and one job becomes two different fan-outs, billed twice, against an expected map the workers no longer match.

So the planner makes the plan stable rather than trusting the model to be:

• Task identity is derived, not generated — task_id is a digest of the goal and the role. A test re-plans the same goal with a model that rewords every instruction and asserts the ids are identical.

• Roles come from an allow-list — a model naming a specialist nobody runs produces work that can only dead-letter, so it is refused at planning time.

• Fan-out is capped, and a duplicate role is refused because expected is keyed by role and the twin would settle the job early.

• It fails closed — unparseable output raises, the delivery is nacked, and the job dead-letters where an operator can see it.

from agent_mesh.planner import LLMDecomposer

supervisor = Supervisor(
    bus, store, decomposer=LLMDecomposer(model, roles=("researcher", "analyst"))
)

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.

Fan-in state has the same split. JobStore is the in-memory reference and RedisJobStore persists expected roles, partial results, failures, and the final status. Redis updates use optimistic transactions, so only the process that atomically changes running to a terminal state emits the completion event. Restarting an aggregator between two role results does not forget the first.

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:

Install MCP support with:

pip install 'agent-mesh[mcp]'

Deployment quickstart

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

Run the deployable Redis-backed API and workers (fake providers by default):

docker compose up --build
curl -X POST http://localhost:8000/jobs \
  -H "content-type: application/json" \
  -d '{"goal":"Compare two implementation options"}'

Set AGENT_MESH_PROVIDER_MODE=mixed with OPENAI_API_KEY and XAI_API_KEY to run Grok as researcher and OpenAI as analyst. Redis uses append-only persistence and a named Compose volume.

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).

Testing

ruff check .
pytest -q

The default suite uses FakeProvider, so it runs with no credentials. See the measured result and its limits at the top of this README.

Live integration tests are opt-in and skipped otherwise:

AGENT_MESH_LIVE=1 pytest tests/test_live_providers.py

The live tier can catch failures the fake provider structurally cannot, including a retired model id, a changed base URL, or a moved auth scheme.

Design Notes

The in-memory bus and job store are reference implementations. The supplied server and Compose stack use Redis Streams plus RedisJobStore; transport and fan-in state therefore survive process replacement instead of mixing a durable queue with an in-memory join that forgets half-completed jobs.

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.