LangGraph Integration
If you already use LangGraph, Stroma can add contract validation to your existing graphs without rewriting them.
Install
uv add stroma[langgraph]
Decorating LangGraph nodes
Use @stroma_langgraph_node to attach contracts to your LangGraph node functions:
from pydantic import BaseModel
from stroma import (
ContractRegistry,
NodeContract,
StromaRunner,
)
from stroma.adapters.langgraph import (
LangGraphAdapter,
stroma_langgraph_node,
)
class InputState(BaseModel):
x: int
class OutputState(BaseModel):
y: int
# Define contracts
registry = ContractRegistry()
contract = NodeContract(
node_id="add_one",
input_schema=InputState,
output_schema=OutputState,
)
registry.register(contract)
@stroma_langgraph_node("add_one", contract) # (1)!
async def add_one(state: InputState) -> dict:
return {"y": state.x + 1}
- Works like
@stroma_nodebut is designed for LangGraph's state-passing pattern.
Wrapping a graph
The LangGraphAdapter discovers decorated nodes in your graph and wraps them with validation:
runner = StromaRunner.quick()
adapter = LangGraphAdapter(registry, runner)
# Assuming you have a compiled LangGraph graph
# graph = build_your_graph()
# wrapped = adapter.wrap(graph) # (1)!
wrap()finds all nodes decorated with@stroma_langgraph_node, replaces them with validating wrappers, and returns the modified graph. Non-decorated nodes are left untouched.
How wrapping works
When the adapter wraps a node, it:
- Extracts the state dict from LangGraph's state object
- Validates it against the node's input schema
- Calls the original node function
- Validates the result against the node's output schema
- Returns the validated output as a dict
If validation fails at either boundary, a ContractViolation is raised — just like with the standard runner.
State extraction
The adapter handles multiple state formats: plain dicts, Pydantic models, objects with .dict(), and objects with __dict__. This covers LangGraph's various state representations.
When to use the adapter vs. the runner
| Use case | Recommendation |
|---|---|
| New pipeline from scratch | Use StromaRunner directly |
| Existing LangGraph graph | Use LangGraphAdapter to add validation |
| Gradual migration | Start with the adapter, migrate nodes to the runner over time |
Tip
The adapter currently provides contract validation only. Full runner features (retries, checkpointing, cost tracking) require using StromaRunner directly.
Recap
- Install with
uv add stroma[langgraph] - Decorate nodes with
@stroma_langgraph_nodeto attach contracts - Use
LangGraphAdapter.wrap(graph)to add validation to an existing graph - The adapter validates inputs and outputs at every decorated node boundary
That's the end of the tutorial! For deeper dives, check the Concepts guide and the API Reference.