# LangGraph and OpenAI Agents SDK

> Add runstate coordination to LangGraph graphs and OpenAI Agents SDK workers without changing how the agents run.

Source: https://docs.getrunstate.com/guides/integrations/

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**

```ts
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**

```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](https://docs.getrunstate.com/guides/budgets/) 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**

```ts
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**

```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](https://docs.getrunstate.com/sdk/python/#blocking-client)); from other languages, use the [HTTP API](https://docs.getrunstate.com/api/) directly.