LangGraph and OpenAI Agents SDK

runstate doesn’t replace your agent framework and has no framework-specific adapter to install. You wrap the functions your agents call at the points where they would otherwise conflict: shared work, rate-limited tools, scarce resources and money. The two patterns below cover most integrations.

LangGraph: wrap the tools a node calls

Parallel branches of a graph, or copies of the same graph running on different machines, often ask for the same research and hit the same APIs. Wrap those functions once. The nodes call them as before.

TypeScript

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

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

// Every branch and every machine shares one result per company in this run.
const research = runstate.once(researchCompany);
// At most 60 searches a minute across the whole swarm.
const search = runstate.limit('search-api', searchWeb, { max: 60, per: 'minute' });

const graph = new StateGraph(State)
  .addNode('profile', async ({ company }) => ({ sections: [(await research(company)).profile] }))
  .addNode('news', async ({ company }) => ({ sections: [(await search(`${company} news`)).join('\n')] }))
  .addEdge(START, 'profile')
  .addEdge(START, 'news')
  .addEdge('profile', END)
  .addEdge('news', END)
  .compile();

const result = await graph.invoke({ company: 'acme' });
console.log(result.sections);
await runstate.finish();

Python

import operator
from typing import Annotated, TypedDict

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


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


# Every branch and every machine shares one result per company in this run.
@runstate.once
def research_company(company: str) -> dict: ...


# At most 60 searches a minute across the whole swarm.
@runstate.limit("search-api", max=60, per="minute")
def search_web(query: str) -> list[str]: ...


def profile(state: State) -> dict:
    return {"sections": [research_company(state["company"])["profile"]]}


def news(state: State) -> dict:
    return {"sections": ["\n".join(search_web(f"{state['company']} news"))]}


builder = StateGraph(State)
builder.add_node("profile", profile)
builder.add_node("news", news)
builder.add_edge(START, "profile")
builder.add_edge(START, "news")
builder.add_edge("profile", END)
builder.add_edge("news", END)
graph = builder.compile()

result = graph.invoke({"company": "acme", "sections": []})
print(result["sections"])
runstate.finish()

The same shape works for scarce resources (slots) and expensive models (a budget). For async graphs, decorate async def functions and await them.

OpenAI Agents SDK: run an agent per job from a queue

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

TypeScript

summarize.ts

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

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

export const summarize = runstate.queue(
  'summarize',
  async (input: { text: string }) => {
    const result = await run(summarizer, input.text);
    return { summary: result.finalOutput };
  },
  { concurrency: 4, lease: '120s' },
);

worker.ts

import runstate from 'runstate-sdk';
import './summarize.ts';

await runstate.serve(); // drains and exits on SIGTERM

producer.ts

import { readFile } from 'node:fs/promises';
import { summarize } from './summarize.ts';

const text = await readFile('report.txt', 'utf8');
const { summary } = await summarize({ text });

Python

summarize.py

import runstate
from agents import Agent, Runner

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


@runstate.queue("summarize", concurrency=4, lease="120s")
async def summarize(input: dict) -> dict:
    result = await Runner.run(summarizer, input["text"])
    return {"summary": result.final_output}

worker.py

import asyncio

import runstate

import summarize  # noqa: F401  (defines the queue)

asyncio.run(runstate.aserve())  # drains and exits on SIGTERM

producer.py

import asyncio

from summarize import summarize


async def main() -> None:
    with open("report.txt", encoding="utf-8") as f:
        text = f.read()
    result = await summarize({"text": text})
    print(result["summary"])


asyncio.run(main())

Pick a lease comfortably longer than one renewal round trip. The SDK renews it while the agent runs, and it is how long a crashed worker’s job waits before another worker gets it.

Other frameworks

runstate works from any TypeScript or Python code, so the same wrappers fit into other agent frameworks at the same points: around shared work, rate-limited tools, scarce resources and spend. Synchronous and async Python are both supported (see Sync and async). From other languages, use the HTTP API directly.