GraphOS

The Service Mesh for AI Agents.

npm version npm downloads PyPI version PyPI downloads License: MIT Node >= 20 Python >= 3.10

GraphOS is an open-source governance and observability layer for LangGraph.js.

Wrap your compiled graph in one line, get policy enforcement (loops, budgets) and a local-first live dashboard with time-travel replay. No SaaS, no signup, no telemetry leaving your machine.

GraphOS Demo

As agents move from demos to production, three things bite:

• Infinite loops — the agent ping-pongs between nodes, burning tokens silently.
• Runaway cost — one bad prompt eats your monthly OpenAI budget before you notice.
• The black-box problem — no way to see what happened inside a 20-step run until it's finished.

GraphOS fixes this by wrapping your CompiledGraph with a policy-driven interceptor and streaming every step to a local dashboard.

• LoopGuard — halt when a node revisits with identical state (mode: "state") or simply visits N times (mode: "node", for agents whose state grows on every iteration).
• BudgetGuard — kill the run when cumulative cost exceeds your USD ceiling.
• MCPGuard — allow-list / deny-list MCP servers and tools, and cap MCP call volume before an agent drifts into unsafe tool usage.
• tokenCost() — drop-in cost extractor that reads usage_metadata off LangChain messages and applies a built-in price table for OpenAI + Anthropic models.

• Live graph — nodes glow as the agent traverses; halted nodes flash red.
• Per-step detail panel — click a step or scrub the timeline to see messages, tool calls, token usage, and the policy halt reason.
• Session switcher + time-travel — every run persists to SQLite (~/.graphos/traces.db); replay any past session step-by-step.

TypeScript / Node:

npm install @graphos-io/sdk
# or
pnpm add @graphos-io/sdk

Python:

pip install graphos-io

TypeScript:

import {
 GraphOS,
 LoopGuard,
 BudgetGuard,
 tokenCost,
 createWebSocketTransport,
 PolicyViolationError,
} from "@graphos-io/sdk";
import { myLangGraphApp } from "./agent";
const managed = GraphOS.wrap(myLangGraphApp, {
 projectId: "my-agent",
 policies: [
 new LoopGuard({ mode: "node", maxRepeats: 10 }),
 new BudgetGuard({ usdLimit: 2.0, cost: tokenCost() }),
 ],
 onTrace: createWebSocketTransport(),
});
try {
 const result = await managed.invoke({
 messages: [{ role: "user", content: "Analyze the market." }],
 });
 console.log(result);
} catch (err) {
 if (err instanceof PolicyViolationError) {
 console.log(`halted by ${err.policy}: ${err.reason}`);
 } else {
 throw err;
 }
}

Python:

import asyncio
from graphos_io import (
 wrap, LoopGuard, BudgetGuard, token_cost,
 create_websocket_transport, PolicyViolationError,
)
from my_agent import build_graph # your compiled LangGraph
async def main():
 managed = wrap(
 build_graph(),
 project_id="my-agent",
 policies=[
 LoopGuard(mode="node", max_repeats=10),
 BudgetGuard(usd_limit=2.0, cost=token_cost()),
 ],
 on_trace=create_websocket_transport(),
 )
 try:
 result = await managed.invoke({"messages": [{"role": "user", "content": "Analyze the market."}]})
 print(result)
 except PolicyViolationError as err:
 print(f"halted by {err.policy}: {err.reason}")
asyncio.run(main())

invoke() returns the merged final state. stream() is also available if you want to consume per-step updates yourself.

Both SDKs ship into the same dashboard over the same JSON-over-WebSocket protocol — point a Python agent and a TypeScript agent at it and watch both in one UI.

npx @graphos-io/dashboard graphos dashboard

Open http://localhost:4000. Run anything that calls createWebSocketTransport() and watch the graph execute live.

The dashboard persists every event to ~/.graphos/traces.db. By default it keeps the 200 most-recent sessions and prunes older ones; tune via GRAPHOS_RETENTION_SESSIONS.

GraphOS architecture: your code → @graphos-io/sdk → @graphos-io/dashboard, with SQLite persistence

The SDK runs in your process — zero network calls unless you point a transport at one. The dashboard is a separate local process started with graphos dashboard.

pnpm install
pnpm dev # dashboard + WS telemetry
pnpm demo:loop # LoopGuard halts an A↔B cycle
pnpm demo:budget # BudgetGuard halts a 4-node pipeline

Open http://localhost:4000.

• LoopGuard (state + node modes)
• BudgetGuard + tokenCost() price-table cost extractor
• WebSocket telemetry transport
• Live graph view with active / halted node states
• SQLite persistence + retention
• Session switcher + time-travel scrubber
• Per-step detail panel (messages, tool calls, usage)
• graphos dashboard CLI
• MCPGuard + MCP proxy
• Python SDK parity (graphos-io)

Bug reports and PRs welcome at github.com/ahmedbutt2015/graphos.

MIT — © Ahmed Butt