LangGraph and OpenAI Agents SDK

runstate doesn’t replace your agent framework and has no framework-specific adapter to install. You call the SDK at the points where agents would otherwise conflict: before touching a shared resource, when picking up work, when spending money. The two patterns below cover most integrations.

LangGraph: claim a shared resource inside a node

Parallel branches of a graph (or copies of the same graph running on different machines) can contend for one resource. Wrap the critical section of the node in a claim so only one branch works on it at a time.

TypeScript

import { Annotation, END, START, StateGraph } from '@langchain/langgraph';
import { Runstate } from 'runstate-sdk';

const State = Annotation.Root({
  sections: Annotation<string[]>({ reducer: (a, b) => [...a, ...b], default: () => [] }),
});

const coordinator = new Runstate({ holder: 'report-coordinator' });
const run = await coordinator.scopes.create();

// Each writer uses its own holder so the console shows who owns the report.
function writer(name: string) {
  return async () => {
    const rs = new Runstate({ holder: name });
    const section = await rs.scope(run.id).claim('file:quarterly-report.md').run(
      async (lease) => `${name} wrote a section (generation ${lease.generation})`,
      { wait: true, timeoutMs: 60_000 }, // queue behind the current owner
    );
    return { sections: [section] };
  };
}

const graph = new StateGraph(State)
  .addNode('writer_a', writer('writer-a'))
  .addNode('writer_b', writer('writer-b'))
  .addEdge(START, 'writer_a')
  .addEdge(START, 'writer_b')
  .addEdge('writer_a', END)
  .addEdge('writer_b', END)
  .compile();

const result = await graph.invoke({ sections: [] });
console.log(result.sections);
await run.complete();

Python

import asyncio
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from runstate import AsyncRunstate


class State(TypedDict):
    sections: Annotated[list[str], operator.add]


async def main() -> None:
    coordinator = AsyncRunstate(holder="report-coordinator")
    run = await coordinator.scopes.create()

    # Each writer uses its own holder so the console shows who owns the report.
    def writer(name: str):
        async def node(state: State) -> dict:
            rs = AsyncRunstate(holder=name)

            async def write(lease):
                return f"{name} wrote a section (generation {lease.generation})"

            section = await rs.scope(run.id).claim("file:quarterly-report.md").run(
                write, wait=True, timeout_ms=60_000  # queue behind the current owner
            )
            await rs.aclose()
            return {"sections": [section]}

        return node

    builder = StateGraph(State)
    builder.add_node("writer_a", writer("writer-a"))
    builder.add_node("writer_b", writer("writer-b"))
    builder.add_edge(START, "writer_a")
    builder.add_edge(START, "writer_b")
    builder.add_edge("writer_a", END)
    builder.add_edge("writer_b", END)
    graph = builder.compile()

    result = await graph.ainvoke({"sections": []})
    print(result["sections"])
    await run.complete()
    await coordinator.aclose()


asyncio.run(main())

The same shape works for any node that calls a rate-limited tool (await run.quota('search-api').take() before the call) or an expensive model (reserve and settle a budget around it).

OpenAI Agents SDK: run an agent per task from a work queue

Keep the agent runtime on your own infrastructure and let runstate hand out the work. consume() renews the task’s lease while the agent runs, records the result, and requeues the task if the agent throws, so a crashed worker’s task is picked up by another.

TypeScript

import { Agent, run as runAgent } from '@openai/agents';
import { Runstate } from 'runstate-sdk';

const summarizer = new Agent({
  name: 'summarizer',
  instructions: 'Summarize the given document in one short paragraph.',
});

const rs = new Runstate({ holder: `summarizer-${process.pid}` });
const queue = rs.scope(process.env.RUNSTATE_SCOPE_ID!).mailbox('documents');

const stop = new AbortController();
process.on('SIGTERM', () => stop.abort());

await queue.consume<{ text: string }>(
  async (data, delivery) => {
    const result = await runAgent(summarizer, data.text);
    await delivery.complete({ payload: { summary: result.finalOutput } });
  },
  { concurrency: 4, leaseSeconds: 120, signal: stop.signal },
);

Python

import asyncio
import os

from agents import Agent, Runner
from runstate import AsyncRunstate

summarizer = Agent(
    name="summarizer",
    instructions="Summarize the given document in one short paragraph.",
)


async def main() -> None:
    async with AsyncRunstate(holder=f"summarizer-{os.getpid()}") as rs:
        queue = rs.scope(os.environ["RUNSTATE_SCOPE_ID"]).mailbox("documents")

        async def handle(data, delivery):
            result = await Runner.run(summarizer, data["text"])
            await delivery.complete(result={"payload": {"summary": result.final_output}})

        await queue.consume(handle, concurrency=4, lease_seconds=120)


asyncio.run(main())

Pick leaseSeconds comfortably longer than one renewal round trip; the lease is renewed about every third of that interval while the agent runs, and it is how long a crashed worker’s task waits before another worker gets it.

Other frameworks

runstate works from any TypeScript or Python code, so the same calls fit into other agent frameworks at the same points: before a shared resource, when taking work, around spend. If your framework runs synchronous Python, use the blocking Runstate client (see the Python SDK reference); from other languages, use the HTTP API directly.