Framework integrations

LangGraph Integration

Use MuBit as a persistent BaseStore for LangGraph StateGraphs. Each node reads or writes through PutOp / SearchOp.

The mubit-langgraph adapter implements LangGraph's BaseStore interface. Pass it to graph.compile(store=...) and your graph nodes can read or write memory through the standard PutOp / SearchOp ops without any LangGraph-specific MuBit knowledge.

pip install mubit-langgraph[langgraph]

Minimal usage

langgraph_minimal.py

from mubit_langgraph import MubitStore
from langgraph.graph import StateGraph
from langgraph.store.base import PutOp, SearchOp
 
store = MubitStore(api_key="mbt_...")
NS = ("memories", "user-1", "session-1")
 
# In graph nodes, use the store directly
store.batch([PutOp(namespace=NS, key="finding-1", value={"text": "...", "intent": "lesson"})])
results = store.batch([SearchOp(namespace_prefix=NS, query="security issues", limit=5)])
 
# MAS extensions
store.register_agent(NS, agent_id="reviewer", role="code-reviewer")
store.checkpoint(NS, snapshot="Review complete")
store.handoff(NS, from_agent_id="planner", to_agent_id="reviewer",
             content="...", requested_action="review")
 
graph.compile(store=store)

Full example: Code review pipeline

A StateGraph with planner, reviewer loop, and summarizer. The MuBit store persists findings across steps and across sessions.

code_review/main.py

from langgraph.graph import StateGraph, START, END
from langgraph.store.base import PutOp, SearchOp
from mubit_langgraph import MubitStore
 
store = MubitStore(endpoint="https://api.mubit.ai", api_key="mbt_...")
NAMESPACE = ("memories", "code-reviewer", "review-session")
 
# Register agents
for agent_id, role in [
    ("planner", "review-planner"),
    ("reviewer", "item-reviewer"),
    ("summarizer", "review-summarizer"),
]:
    store.register_agent(NAMESPACE, agent_id=agent_id, role=role)
 
def planner_node(state):
    past = store.batch([SearchOp(namespace_prefix=NAMESPACE, query="code review checklist", limit=3)])[0]
    # ... LLM call to generate checklist ...
    store.checkpoint(NAMESPACE, snapshot=f"Checklist created with {len(checklist)} items")
    return {"checklist": checklist, "current_idx": 0, "findings": []}
 
def reviewer_node(state):
    item = state["checklist"][state["current_idx"]]
    # ... LLM call to evaluate item ...
    store.batch([PutOp(
        namespace=NAMESPACE,
        key=f"finding-{state['current_idx']}",
        value={"text": finding, "intent": "lesson"},
    )])
    return {"findings": state["findings"] + [finding], "current_idx": state["current_idx"] + 1}
 
def summarizer_node(state):
    context = store.get_context(NAMESPACE, query="code review findings", max_token_budget=4096)
    # ... LLM call to synthesize final review ...
    store.record_outcome(NAMESPACE, reference_id="review-001",
                         outcome="success", rationale="Review completed.")
    return {"final_review": review}
 
graph = StateGraph(ReviewState)
graph.add_node("planner", planner_node)
graph.add_node("reviewer", reviewer_node)
graph.add_node("summarizer", summarizer_node)
graph.add_edge(START, "planner")
graph.add_edge("planner", "reviewer")
graph.add_conditional_edges("reviewer", should_continue,
                            {"reviewer": "reviewer", "summarizer": "summarizer"})
graph.add_edge("summarizer", END)
 
result = graph.compile().invoke({
    "code_diff": "...", "checklist": [], "current_idx": 0,
    "findings": [], "final_review": "",
})

Extended MuBit features

MubitStore exposes the full MAS surface. Every method takes the namespace as its first positional argument:

store.register_agent(NS, agent_id="...", role="...", read_scopes=[...], write_scopes=[...])
store.handoff(NS, from_agent_id="...", to_agent_id="...", content="...", requested_action="review")
store.feedback(NS, handoff_id="...", verdict="approve", from_agent_id="...")
store.checkpoint(NS, snapshot="...", label="...")
store.record_outcome(NS, reference_id="...", outcome="success", rationale="...")
store.surface_strategies(NS)
store.diagnose(NS, error_text="...")
store.archive(NS, content="...", artifact_kind="...")
store.dereference(NS, reference_id="...")
store.get_context(NS, query="...", max_token_budget=...)

Gotchas

Namespace is the source of truth for scope. Use a stable tuple shape across nodes ((memories, user_id, session_id)) so SearchOp finds writes from earlier nodes.
PutOp.value must be a dict. Cast strings before storing — the adapter rejects raw strings.
SearchOp returns a list of Item objects. Each Item.value is the dict you stored.
JS adapter: npm install @mubit-ai/langgraph provides the same surface async, with camelCase method names.

Version compatibility

`mubit-langgraph`	`mubit-sdk`	`langgraph`
`0.5.x`	`>= 0.6.0`	`>= 0.2.40`