vs-graph

JSON-driven LangGraph wrapper for building agentic workflows — define graph topology in JSON, implement nodes and edges in Python, and get a fully wired, observable, checkpointed execution graph.

---

Overview

vs-graph wraps LangGraph and eliminates the boilerplate of building agentic graphs: wiring nodes, routing edges, setting up checkpointers, and propagating trace events. You declare the graph structure in a JSON file, implement node logic by extending VsBaseNode, and wire routing logic by extending VsBaseConditionalEdge or VsDynamicFanOutEdge. The library handles everything else.

The @node and @edge decorators self-register implementations at import time. The @graph decorator on a graph class specifies which packages to auto-import so no manual import list is required.

---

The Problem It Solves

Without vs-graph

# Manual LangGraph wiring — every graph requires this boilerplate
from langgraph.graph import StateGraph, START, END

sg = StateGraph(MyState)
sg.add_node("classifier", classifier_node.invoke)
sg.add_node("executor", executor_node.invoke)
sg.add_node("responder", responder_node.invoke)
sg.add_edge(START, "classifier")
sg.add_conditional_edges("classifier", routing_fn, {"execute": "executor", "default": "responder"})
sg.add_edge("executor", "responder")
sg.add_edge("responder", END)
checkpointer = MemorySaver()
graph = sg.compile(checkpointer=checkpointer)

Every graph has this same boilerplate. Adding a node requires updating both code and wiring.

With vs-graph

# graph topology is in JSON
# Python only contains business logic

@graph(nodes=["myapp.graph.nodes"], edges=["myapp.graph.edges"])
class MyGraph(VsBaseGraph[MyState]):
    def __init__(self):
        with open("my_graph.json") as f:
            graph_json = json.load(f)
        super().__init__(graph_json=graph_json, state_class=MyState)
        self.build()

graph = MyGraph()
result = await graph.invoke({"trace_id": "", "error": None, "user_input": "hello"})

---

How It All Fits Together

@graph(nodes=[...], edges=[...])
class MyGraph(VsBaseGraph[MyState])
    └── __init__: load JSON, call build()

build()
    ├── _auto_import_packages()     # walks node/edge packages, triggers @node/@edge registration
    ├── _instantiate_nodes()        # VsNodeRegistry.create() for each JSON node definition
    ├── _instantiate_edges()        # VsEdgeRegistry.create() for conditional/dynamic_fan_out
    ├── _validate_graph()           # checks all node ids, edge targets, and exit nodes
    ├── _load_checkpointer()        # reads config.ini [graph] section
    └── _build_langgraph()          # compiles StateGraph with checkpointer

invoke(state)
    ├── auto-generates trace_id if not set
    └── StateGraph.ainvoke(state)
            └── per node: VsBaseNode.invoke()
                    ├── trace: "node started"
                    ├── execute(state)          ← your business logic
                    └── trace: "node completed" / "node failed"
            └── per conditional / dynamic_fan_out edge: async VsBaseEdge.route()
                    ├── _evaluate(state) / get_sends(state)   ← your routing logic
                    └── trace: "routing to next_node" / "fanning out to N branch(es)"
            (direct and fan_out edges are plain LangGraph edges — no wrapper, no trace)

---

Graph JSON

The JSON file declares the graph topology. Python classes are referenced by the name field in each node/edge definition — matching the name passed to @node(name="...") or @edge(name="...").

{
  "version": "1.0",
  "name": "my_graph",
  "entry_node": "classifier",
  "exit_nodes": ["responder"],
  "nodes": [
    { "id": "classifier", "name": "classifier_node", "config": {} },
    { "id": "executor",   "name": "executor_node",   "config": {} },
    { "id": "responder",  "name": "responder_node",  "config": {} }
  ],
  "edges": [
    {
      "type": "conditional",
      "source": "classifier",
      "edge_class": "classifier_routing_edge",
      "route": {
        "execute": "executor",
        "default": "responder"
      }
    },
    { "type": "direct", "source": "executor", "target": "responder" }
  ]
}

JSON Fields

Field	Required	Description
name	Yes	Graph name — used in logs and error messages
version	No	Version string (default "1.0")
entry_node	Yes	id of the node that receives the initial state
exit_nodes	Yes	List of node ids that connect to END
nodes	Yes	Array of node definitions
nodes[].id	Yes	Unique node id within this graph — used in edge source/target fields
nodes[].name	Yes	Registry name — must match the @node(name="...") decorator value
nodes[].config	No	Arbitrary dict passed to the node constructor as self.config
edges	No	Array of edge definitions
edges[].type	Yes	direct, fan_out, conditional, or dynamic_fan_out
edges[].source	Yes	Node id this edge originates from
edges[].target	For direct	Target node id
edges[].targets	For fan_out	Array of target node ids (minimum 2)
edges[].edge_class	For conditional, dynamic_fan_out	Registry name matching @edge(name="...")
edges[].route	For conditional	Map of route keys to target node ids — must include "default"
edges[].config	No	Arbitrary dict passed to the edge constructor as self.config

Edge Types

Type	Python class needed	Description
direct	No	Fixed single target — fully declared in JSON
fan_out	No	Fixed parallel targets — fully declared in JSON
conditional	Yes — extends VsBaseConditionalEdge	_evaluate(state) returns a route key; key is looked up in route map
dynamic_fan_out	Yes — extends VsDynamicFanOutEdge	get_sends(state) returns a list of Send objects for runtime branching

---

State

All graph state must extend VsBaseGraphState:

from typing import Optional
from vs_graph.schema.vs_base_graph_state import VsBaseGraphState

class MyState(VsBaseGraphState):
    user_input: str
    intent: Optional[str]
    response: Optional[str]

VsBaseGraphState is a TypedDict providing:

Field	Type	Description
trace_id	str	Auto-generated per invoke() call if not set. Links all trace events for this execution.
error	Optional[VsGraphError]	Populated by the framework when a node raises VsUserException. Contains message, cause, stack_trace.
current_message	Optional[str]	Convenience field for the current user-facing message. Not set by the framework.

---

Nodes

Extend VsBaseNode and implement execute. Decorate with @node(name="...") to self-register in VsNodeRegistry.

from vs_graph.node.vs_base_node import VsBaseNode
from vs_graph.decorator.vs_node_decorator import node
from vs_graph.exception.vs_graph_exceptions import VsUserException, VsNodeExecutionException

@node(name="classifier_node")
class ClassifierNode(VsBaseNode[MyState]):

    async def execute(self, state: MyState) -> dict:
        if not state.get("user_input"):
            raise VsUserException("Input is required.")
        state["intent"] = "execute"
        await self.trace(state["trace_id"], "intent classified", data={"intent": state["intent"]})
        return state

@node decorator

@node(name="classifier_node")   # registers under key "classifier_node"
@node                           # registers under the class name

The name value must match the "name" field in the JSON node definition.

What invoke() does (called by the graph, not you)

VsBaseNode.invoke() wraps your execute():

1. Logs and publishes trace: "node '{id}' started"
2. Calls await self.execute(state)
3. On success: logs and publishes trace: "node '{id}' completed" with duration_seconds
4. On VsUserException: stores error in state["error"] and returns state (graph continues)
5. On VsNodeExecutionException: logs error and re-raises (graph stops)
6. On any other exception: logs error and re-raises (graph stops)

Publishing trace events

async def execute(self, state: MyState) -> dict:
    await self.trace(
        state["trace_id"],
        "retrieved context",
        data={"num_chunks": 8},
    )
    return state

self.trace() is a no-op if no trace_publisher was passed to the graph.

Node constructor args

VsBaseNode.__init__ receives:

Parameter	Source
node_id	The id field from the JSON node definition
config	The config dict from the JSON node definition
trace_publisher	The publisher passed to VsBaseGraph.__init__

Access them as self.node_id, self.config, self.trace_publisher.

---

Edges

Direct edge

No Python class. Declared fully in JSON:

{ "type": "direct", "source": "executor", "target": "responder" }

Fan-out edge

No Python class. Declares fixed parallel targets in JSON (minimum 2):

{
  "type": "fan_out",
  "source": "splitter",
  "targets": ["worker_a", "worker_b", "worker_c"]
}

All targets receive the same state and execute in parallel via LangGraph's native fan-out.

Conditional edge

Extend VsBaseConditionalEdge and implement _evaluate. Decorate with @edge(name="..."):

from vs_graph.edge.vs_base_conditional_edge import VsBaseConditionalEdge
from vs_graph.decorator.vs_edge_decorator import edge

@edge(name="classifier_routing_edge")
class ClassifierRoutingEdge(VsBaseConditionalEdge[MyState]):

    def _evaluate(self, state: MyState) -> str:
        if state.get("intent") == "execute":
            return "execute"
        return "default"

_evaluate returns a route key. The base class's async route() wrapper looks it up in the route map from JSON, publishes the routing trace, and returns the target node id. If the key is not in the map, it falls back to "default". A "default" entry in the JSON route map is required.

JSON:

{
  "type": "conditional",
  "source": "classifier",
  "edge_class": "classifier_routing_edge",
  "route": {
    "execute": "executor",
    "default": "responder"
  }
}

Dynamic fan-out edge

Extend VsDynamicFanOutEdge and implement get_sends. Returns a list of LangGraph Send objects — each specifies a target node and a custom state for that branch:

from vs_graph.edge.vs_dynamic_fan_out_edge import VsDynamicFanOutEdge
from vs_graph.decorator.vs_edge_decorator import edge
from langgraph.types import Send

@edge(name="document_fan_out_edge")
class DocumentFanOutEdge(VsDynamicFanOutEdge[MyState]):

    def get_sends(self, state: MyState) -> list:
        return [
            Send("process_document", {**state, "document": doc})
            for doc in state["documents"]
        ]

JSON:

{
  "type": "dynamic_fan_out",
  "source": "splitter",
  "edge_class": "document_fan_out_edge"
}

@edge decorator

@edge(name="classifier_routing_edge")   # registers under key "classifier_routing_edge"
@edge                                   # registers under the class name

The name value must match the "edge_class" field in the JSON edge definition.

---

@graph Decorator — Auto-Import Packages

Instead of manually importing every node and edge module, use @graph on your graph class to declare which packages contain your implementations. The library scans and imports them automatically at build() time, triggering @node and @edge registration.

from vs_graph.decorator.vs_graph_decorator import graph
from vs_graph.vs_base_graph import VsBaseGraph

@graph(
    nodes=["myapp.graph.nodes"],
    edges=["myapp.graph.edges"],
)
class MyGraph(VsBaseGraph[MyState]):

    def __init__(self):
        with open("my_graph.json") as f:
            graph_json = json.load(f)
        super().__init__(graph_json=graph_json, state_class=MyState)
        self.build()

Both nodes and edges accept a list of package paths. Each package is walked recursively — all modules within it are imported. You can specify multiple packages:

@graph(
    nodes=["myapp.graph.nodes", "shared.common_nodes"],
    edges=["myapp.graph.edges"],
)
class MyGraph(VsBaseGraph[MyState]):
    ...

Without @graph, node and edge modules must be imported manually before build() is called:

import myapp.graph.nodes.classifier_node  # noqa
import myapp.graph.nodes.executor_node    # noqa

@graph with no packages, or subclasses without the decorator, are both valid — auto-import is skipped.

---

Registries

VsNodeRegistry

VsNodeRegistry is a static registry. @node populates it at import time.

Method	Description
register(name, node_class)	Called by @node — raises ValueError if name already registered
get(name)	Returns the class — raises ValueError if not found
create(name, node_id, config, **kwargs)	Instantiates the node class
is_registered(name)	Returns True if name is registered
list_nodes()	Returns sorted list of registered names
clear()	Removes all registered nodes — for testing only

VsEdgeRegistry

Same pattern as VsNodeRegistry for edge classes.

Method	Description
register(name, edge_class)	Called by @edge — raises ValueError if name already registered
get(name)	Returns the class — raises ValueError if not found
create(name, source, **kwargs)	Instantiates the edge class
is_registered(name)	Returns True if name is registered
list_edges()	Returns sorted list of registered names
clear()	Removes all registered edges — for testing only

---

Checkpointer

Configured via config.ini. Provides LangGraph thread-level state persistence across invoke() calls.

[graph]
checkpointer = memory

[graph]
checkpointer             = redis
redis_url                = redis://localhost:6379
checkpoint_key_prefix    = vs:checkpoint:
checkpoint_ttl_seconds   = 3600

Key	Default	Description
graph.checkpointer	memory	Checkpointer type: memory or redis
graph.redis_url	redis://localhost:6379	Redis connection URL (redis only)
graph.checkpoint_key_prefix	vs:checkpoint:	Key prefix for stored checkpoints (redis only)
graph.checkpoint_ttl_seconds	None	TTL for checkpoint keys in seconds (redis only)

Custom checkpointers can be registered via VsCheckpointerFactory:

from vs_graph.checkpointer.vs_checkpointer_factory import VsCheckpointerFactory

VsCheckpointerFactory.register(
    "postgres",
    lambda cfg: MyPostgresCheckpointer(config=cfg)
)

The factory lambda receives the same cfg dict that _load_checkpointer builds from config.ini.

---

Trace Publisher

Pass a VsBaseTracePublisher to the graph constructor to stream execution events to any external sink (queue, websocket, database):

from vs_common.publisher.vs_base_trace_publisher import VsBaseTracePublisher
from vs_common.schema.vs_trace_level import VsTraceLevel
from typing import Any, Dict, Optional

class MyTracePublisher(VsBaseTracePublisher):

    async def publish(
        self,
        trace_id: str,
        message: str,
        level: VsTraceLevel = VsTraceLevel.INFO,
        data: Optional[Dict[str, Any]] = None,
    ) -> None:
        await my_queue.push({
            "trace_id": trace_id,
            "message": message,
            "level": level.value,
            "data": data or {},
        })

graph = MyGraph(graph_json, state_class=MyState, trace_publisher=MyTracePublisher())

The publisher is automatically propagated to every node and edge instance. You do not pass it to nodes or edges directly.

Events published automatically by the framework:

Event	Source
"node '{id}' started"	VsBaseNode.invoke()
"node '{id}' completed" with duration_seconds	VsBaseNode.invoke()
"node '{id}' user error"	VsBaseNode.invoke() on VsUserException
"node '{id}' failed"	VsBaseNode.invoke() on exception
"edge '{source}' routing to '{next_node}'" with route_key, next_node, duration_seconds	VsBaseConditionalEdge.route()
"edge '{source}' fanning out to {n} branch(es)" with branch_count, targets, duration_seconds	VsDynamicFanOutEdge.route()

Direct and static fan_out edges are pure topology — they compile to plain LangGraph edges and emit no trace events. The routing wrappers are async and await the trace publish before returning the next node, so the routing event is always ordered ahead of the next node's "started" event.

---

Exceptions

Exception	When to raise	Effect
VsUserException	Error is safe to surface to the user	Stored in state["error"] — graph continues to next node
VsNodeExecutionException	Internal node failure	Logged and re-raised — graph stops
VsGraphBuildException	Invalid build state	Raised during build()
VsEdgeEvaluationException	Edge routing failure	Raised from edge evaluate()

from vs_graph.exception.vs_graph_exceptions import VsUserException, VsNodeExecutionException

async def execute(self, state: MyState) -> dict:
    if not state.get("user_input"):
        raise VsUserException("Please provide an input.")

    try:
        result = await call_external_service()
    except Exception as e:
        raise VsNodeExecutionException(f"External service failed: {e}") from e

    state["result"] = result
    return state

state["error"] shape when VsUserException is raised:

{
  "message": "Please provide an input.",
  "cause": null,
  "stack_trace": null
}

---

Full Example

import json
from typing import Optional

from vs_graph.vs_base_graph import VsBaseGraph
from vs_graph.schema.vs_base_graph_state import VsBaseGraphState
from vs_graph.node.vs_base_node import VsBaseNode
from vs_graph.edge.vs_base_conditional_edge import VsBaseConditionalEdge
from vs_graph.decorator.vs_node_decorator import node
from vs_graph.decorator.vs_edge_decorator import edge
from vs_graph.decorator.vs_graph_decorator import graph
from vs_graph.exception.vs_graph_exceptions import VsUserException


# --- State ---

class MyState(VsBaseGraphState):
    user_input: str
    intent: Optional[str]
    response: Optional[str]


# --- Nodes ---

@node(name="classifier_node")
class ClassifierNode(VsBaseNode[MyState]):

    async def execute(self, state: MyState) -> dict:
        if not state.get("user_input"):
            raise VsUserException("Input is required.")
        state["intent"] = "search" if "find" in state["user_input"] else "chat"
        await self.trace(state["trace_id"], "classified", data={"intent": state["intent"]})
        return state


@node(name="responder_node")
class ResponderNode(VsBaseNode[MyState]):

    async def execute(self, state: MyState) -> dict:
        state["response"] = f"Handling intent: {state['intent']}"
        return state


# --- Edges ---

@edge(name="classifier_routing_edge")
class ClassifierRoutingEdge(VsBaseConditionalEdge[MyState]):

    def _evaluate(self, state: MyState) -> str:
        return state.get("intent", "default")


# --- Graph ---

@graph(nodes=["myapp.graph.nodes"], edges=["myapp.graph.edges"])
class MyGraph(VsBaseGraph[MyState]):

    def __init__(self):
        graph_json = {
            "version": "1.0",
            "name": "my_graph",
            "entry_node": "classifier",
            "exit_nodes": ["responder"],
            "nodes": [
                {"id": "classifier", "name": "classifier_node", "config": {}},
                {"id": "responder",  "name": "responder_node",  "config": {}},
            ],
            "edges": [
                {
                    "type": "conditional",
                    "source": "classifier",
                    "edge_class": "classifier_routing_edge",
                    "route": {"search": "responder", "chat": "responder", "default": "responder"}
                }
            ]
        }
        super().__init__(graph_json=graph_json, state_class=MyState)
        self.build()


# --- Run ---

import asyncio

async def main():
    g = MyGraph()
    result = await g.invoke({
        "trace_id": "",
        "error": None,
        "current_message": None,
        "user_input": "find me a product",
        "intent": None,
        "response": None,
    })
    print(result["response"])  # Handling intent: search

asyncio.run(main())

---

Class Reference

---

VsBaseGraph

Abstract base class for all graphs. Extend it and call build() in __init__.

Constructor:

Parameter	Type	Description
graph_json	Dict[str, Any]	Parsed graph JSON — validated on construction
state_class	Type[S]	TypedDict subclass used as LangGraph state type
trace_publisher	Optional[VsBaseTracePublisher]	Publisher injected into all nodes and edges

Methods:

Method	Description
build()	Auto-imports packages, instantiates nodes/edges, validates, loads checkpointer, compiles LangGraph. Call once at startup.
async invoke(state, config=None)	Runs the compiled graph. Auto-generates trace_id if not set.

---

VsBaseNode

Abstract base class for all nodes.

Constructor parameters (injected by VsNodeRegistry.create()):

Parameter	Type	Description
node_id	str	Node id from JSON
config	Optional[Dict[str, Any]]	Node config dict from JSON
trace_publisher	Optional[VsBaseTracePublisher]	Propagated from the graph

Instance attributes:

Attribute	Description
self.node_id	The node id from JSON
self.config	The config dict from JSON
self.trace_publisher	The graph-level trace publisher
self.logger	Logger instance keyed to the class name

Abstract method:

Method	Signature	Description
execute	async execute(state: S) -> dict	Implement your node logic here. Return the updated state dict.

Helper method:

Method	Signature	Description
trace	async trace(trace_id, message, data=None, level=INFO)	Publishes a trace event. No-op if no publisher.

---

VsBaseConditionalEdge

Abstract base class for conditional routing edges.

Constructor parameters:

Parameter	Type	Description
source	str	Source node id
route	Dict[str, str]	Map of route keys to target node ids — must include "default". Stored as self.route_map.
config	Optional[Dict[str, Any]]	Edge config dict from JSON
trace_publisher	Optional[VsBaseTracePublisher]	Propagated from the graph

Abstract method:

Method	Signature	Description
_evaluate	_evaluate(state: S) -> str	Return a route key. route() looks it up in route_map; falls back to "default".

Routing function (registered with LangGraph — you do not call it):

Method	Signature	Description
route	async route(state: S) -> str	Calls _evaluate, resolves the key against route_map (falling back to "default"), publishes the routing trace, returns the target node id. Raises if there is no default.

---

VsDynamicFanOutEdge

Abstract base class for runtime fan-out edges.

Abstract method:

Method	Signature	Description
get_sends	get_sends(state: S) -> List	Return a list of langgraph.types.Send objects. Each Send specifies a target node and a custom state dict for that branch.

Routing function (registered with LangGraph — you do not call it):

Method	Signature	Description
route	async route(state: S) -> List	Calls get_sends, publishes a fan-out trace with branch_count and targets, returns the Send list.

---

VsCheckpointerFactory

Static factory for checkpointer registration and creation.

Method	Description
register(name, factory_fn)	Register a custom checkpointer. factory_fn receives a config dict.
create(checkpointer_type, config)	Instantiate and return the checkpointer.

Built-in types: "memory", "redis".

---

Running Tests

./run_tests.sh