The Service Mesh for AI Agents.

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.

---

🧐 Why GraphOS?

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.

---

✨ What you get

Policy enforcement

• 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.

Local dashboard

• 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.

---

🛠 Install

TypeScript / Node:

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

Python:

pip install graphos-io

---

🚀 Quick start

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.

---

🖥 Run the dashboard

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.

---

📦 Packages

Package	Language	What it does
@graphos-io/core	TS	Shared types (Policy, NodeExecution, TraceEvent)
@graphos-io/sdk	TS	GraphOS.wrap(), LoopGuard, BudgetGuard, tokenCost, transports
@graphos-io/dashboard	TS	Next.js + React Flow dashboard with graphos CLI
@graphos-io/mcp-proxy	TS	Proxy MCP tool calls, emit GraphOS traces, redact payloads, and enforce MCP allow/deny rules
graphos-io	Python	wrap(), LoopGuard, BudgetGuard, MCPGuard, token_cost, async-first transport

---

🏗 Architecture

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.

---

🧪 Run the demos from the monorepo

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.

---

🗺 Roadmap

• 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)

---

🤝 Contributing

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

License

MIT — © Ahmed Butt