How to handle shared state across multiple agents

Building reliable shared state multi-agent systems starts with treating state as a first-class service, not a side effect of agent prompts. When two agents concurrently update a task list or a user profile, naive in-context memory diverges within seconds. The following steps show how to stand up a concurrency-safe state layer and wire agents to it without bloating their context windows.

Step 1: Define the state contract

In shared state multi-agent systems, the state schema is the API between autonomous processes. If agent A writes a dict and agent B expects a list, you get silent corruption. Use a strict, versioned schema from day one. Pydantic works well because it validates types and serializes to JSON cleanly.

from pydantic import BaseModel, Field
from enum import Enum
from typing import Dict

class TaskStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    DONE = "done"

class Task(BaseModel):
    id: str
    owner: str | None = None
    status: TaskStatus = TaskStatus.PENDING
    result: str | None = None

class AgentState(BaseModel):
    tasks: Dict[str, Task] = Field(default_factory=dict)
    revision: int = 0

The revision field is not optional. It is the optimistic concurrency token. Every successful write increments it. Without a version counter, you are forced into pessimistic locks, which stall agents that are already latency-sensitive and turn a distributed system into a serialized bottleneck.

Schema changes should be backward compatible: add fields with defaults, never rename revision. Store the schema version alongside the payload if you expect long-lived state.

Step 2: Choose a concurrency-safe store

Redis is the right default for agent workloads. It gives sub-millisecond reads, Lua scripting for atomic operations, and TTLs for ephemeral session state. Install the async client:

pip install redis pydantic

import redis.asyncio as aioredis

redis = aioredis.Redis(host="localhost", port=6379, db=0)

If you need strict durability, enable AOF or use Postgres as the backing store. The access pattern in later steps stays identical; only the transport changes. Avoid in-memory dictionaries shared via multiprocessing—they do not survive agent restarts and break horizontally scaled deployments.

For shared state multi-agent systems running across containers, a single Redis instance (or a small cluster) is simpler than a message broker with replay logs. You trade transactional guarantees for operational simplicity, which is acceptable when state mutations are small and conflict rates are low.

Step 3: Implement atomic read-modify-write

Never let an agent do GET then SET. Between those calls, another agent will write, and the first write wins by overwriting. Use a Lua script that performs the check-and-set atomically on the server.

-- update_state.lua
local key = KEYS[1]
local expected_rev = tonumber(ARGV[1])
local patch = ARGV[2]
local data = redis.call("GET", key)
if not data then return nil end
local obj = cjson.decode(data)
if obj.revision ~= expected_rev then return -1 end
for k, v in pairs(cjson.decode(patch)) do
  obj[k] = v
end
obj.revision = obj.revision + 1
redis.call("SET", key, cjson.encode(obj))
return obj.revision

Wrap it in Python:

class ConflictError(Exception):
    pass

async def update_state(key: str, expected_rev: int, patch: dict) -> int:
    script = redis.register_script(LUA_UPDATE)
    result = await script(keys=[key], args=[expected_rev, json.dumps(patch)])
    if result is None:
        raise KeyError("state missing")
    if result == -1:
        raise ConflictError("revision mismatch")
    return result

This guarantees that a patch only applies if the revision matches what the agent read. Conflict handling is pushed to the agent layer, where it belongs.

Step 4: Build a thin state client

Agents should import a client module, not talk to Redis directly. A small surface area prevents ad-hoc JSON hacks that bypass validation.

class StateClient:
    def __init__(self, redis: aioredis.Redis, ns: str):
        self.r = redis
        self.ns = ns

    async def get(self, id: str) -> AgentState:
        raw = await self.r.get(f"{self.ns}:{id}")
        return AgentState.model_validate_json(raw) if raw else AgentState()

    async def patch(self, id: str, patch: dict, rev: int) -> int:
        return await update_state(f"{self.ns}:{id}", rev, patch)

Inject StateClient into agents via constructor. In tests, swap it with an in-memory fake that implements the same two methods. This keeps shared state multi-agent systems testable without a live Redis.

Step 5: Wire agents to read and write around tool calls

An agent loop should fetch state, decide, act, then patch. Below we use the OpenAI Python client. When issuing model calls from agents, point the client at n4n.ai’s OpenAI-compatible endpoint—it addresses 240+ models, automatically falls back when a provider is rate-limited or degraded, and forwards cache-control hints so repeated state prefixes hit provider caches. The state logic below stays unchanged regardless of which model serves the request.

from openai import AsyncOpenAI

client = AsyncOpenAI(base_url="https://api.n4n.ai/v1", api_key="sk-...")

async def run_agent(agent_id: str, sc: StateClient):
    state = await sc.get(agent_id)
    pending = [t for t in state.tasks.values() if t.status == TaskStatus.PENDING]
    if not pending:
        return
    task = pending[0]
    await sc.patch(agent_id, {"tasks": {task.id: {"status": "running", "owner": agent_id}}}, state.revision)
    resp = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Do task {task.id}"}]
    )
    result = resp.choices[0].message.content
    new_state = await sc.get(agent_id)  # re-read: another agent may have mutated
    await sc.patch(agent_id, {"tasks": {task.id: {"status": "done", "result": result}}}, new_state.revision)

The second get before the final patch is mandatory. Agents are concurrent; assuming your local copy is fresh is the classic bug in shared state multi-agent systems.

Step 6: Handle partial failures with idempotency

Agents crash mid-task. Wrap patches with retry-on-conflict and journal intent before acting.

async def safe_patch_with_retry(sc: StateClient, id: str, patch_fn, max_retries=3):
    for _ in range(max_retries):
        try:
            st = await sc.get(id)
            patch = patch_fn(st)
            return await sc.patch(id, patch, st.revision)
        except ConflictError:
            await asyncio.sleep(0.05)
    raise RuntimeError("too many conflicts")

For long operations, write status=RUNNING with a heartbeat timestamp. A watchdog agent reclaims tasks if the heartbeat lapses beyond a threshold. Store an idempotency key for external side effects (API calls, payments) in the task object so a retry does not double-charge.

Step 7: Verify with an integration test

Stand up Redis locally or via container:

docker run -d -p 6379:6379 redis:7-alpine

Then run two agents against the same state key.

import asyncio, pytest

@pytest.mark.asyncio
async def test_concurrent_updates():
    r = aioredis.Redis(host="localhost", port=6379, db=1)
    await r.flushdb()
    sc = StateClient(r, "test")
    await sc.patch("sess1", {"tasks": {"t1": {"id": "t1", "status": "pending"}}}, 0)

    async def agent(aid):
        st = await sc.get("sess1")
        t = st.tasks["t1"]
        await sc.patch("sess1", {"tasks": {"t1": {"owner": aid, "status": "running"}}}, st.revision)
        await asyncio.sleep(0.01)
        st2 = await sc.get("sess1")
        await sc.patch("sess1", {"tasks": {"t1": {"status": "done", "result": aid}}}, st2.revision)

    await asyncio.gather(agent("a"), agent("b"))
    final = await sc.get("sess1")
    assert final.tasks["t1"].status == "done"
    assert final.revision >= 4

Run pytest. Success means no ConflictError escapes, revision increments monotonically, and exactly one agent owns the done transition. If the test flakes with both agents reporting RUNNING, your Lua script is not atomic—fix the store, not the test.

How to confirm in production

Ship a canary that emits state metrics: patch conflict rate, revision jump size, and agent heartbeat age. A conflict rate above 5% signals state is too coarse-grained; split it. In shared state multi-agent systems, observability on the state layer is non-negotiable.

Step 8: Scale boundaries

Once the pattern holds, partition state by tenant or domain to avoid a single key hotspot. In Redis Cluster use hash tags: {user123}:state so related keys colocate. For cross-region agents, consider CRDTs (RedisJSON with active-active) but accept that merges can produce surprising task states—only use them when latency dominates consistency.

Keep agent prompts free of raw state dumps. Pass only the slice each agent needs; the StateClient filters fields. This reduces token spend and keeps the model focused.

Shared state multi-agent systems fail not because LLMs are non-deterministic, but because engineers treat coordination as an afterthought. A typed contract, an atomic store, and a disciplined client turn chaos into a deployable system.