Migrating from langgraph

CubePi and langgraph both build tool-using LLM agents, but they have different mental models. This page maps langgraph concepts onto CubePi so you can port code without having to re-learn from scratch.

Mental-model shift

langgraph	CubePi	Why
State graph with nodes, edges, channels	Agent loop that's a plain while loop you can read	A linear loop is easier to reason about than a graph; CubePi never branches at runtime — control flow lives in middleware
Channels (typed state slots)	AgentContext.extra + AgentState.messages	A single dict + a single message list cover every state shape we've seen
StateGraph.add_node(name, fn)	A middleware hook or a tool	Functions in langgraph nodes split into two roles in CubePi: tool execution (when the model decides) vs. middleware (always-on transforms)
add_edge(a, b) / add_conditional_edges	Built-in: tools → next turn → tools → …	The conditional shape (tool calls → re-prompt) is the loop; you don't reify it
MemorySaver / SqliteSaver / PostgresSaver	MemoryCheckpointer / SQLiteCheckpointer / PostgresCheckpointer	Same idea, append-only schema instead of full snapshots
config: {"configurable": {"thread_id": …}}	Agent(thread_id=…)	First-class agent parameter
stream_mode="messages" / "values" / "updates"	agent.subscribe(listener) — one event stream	One pattern, eleven event types
Tools as @tool decorated functions	AgentTool with Pydantic params + async execute	Closer to OpenAI/Anthropic native shape
HumanMessage, AIMessage	UserMessage, AssistantMessage	Same role-tagged messages, just renamed
Interrupts via interrupt_before / interrupt_after	agent.steer(...), agent.follow_up(...), agent.abort()	Imperative control instead of declarative interrupt points
config_schema	Constructor parameters on Agent	No separate schema layer

Side-by-side: a tool-using agent

langgraph

from typing import TypedDict
from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool


@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"72°F and sunny in {city}"


llm = ChatAnthropic(model="claude-sonnet-4-5-20250929").bind_tools([get_weather])

class State(TypedDict):
    messages: list

def call_model(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

def should_continue(state: State):
    last = state["messages"][-1]
    return "tools" if last.tool_calls else END

graph = StateGraph(State)
graph.add_node("llm", call_model)
graph.add_node("tools", ToolNode([get_weather]))
graph.add_edge("__start__", "llm")
graph.add_conditional_edges("llm", should_continue)
graph.add_edge("tools", "llm")
app = graph.compile()

for chunk in app.stream({"messages": [("user", "Weather in Tokyo?")]}):
    print(chunk)

CubePi

import asyncio
from pydantic import BaseModel
from cubepi import Agent, AgentTool, AgentToolResult, Model, TextContent
from cubepi.providers.anthropic import AnthropicProvider


class GetWeatherParams(BaseModel):
    city: str


async def get_weather(tool_call_id, params: GetWeatherParams, *, signal=None, on_update=None):
    return AgentToolResult(content=[TextContent(text=f"72°F and sunny in {params.city}")])


agent = Agent(
    provider=AnthropicProvider(api_key="…"),
    model=Model(id="claude-sonnet-4-5-20250929", provider="anthropic"),
    tools=[AgentTool(
        name="get_weather",
        description="Get current weather for a city.",
        parameters=GetWeatherParams,
        execute=get_weather,
    )],
)
agent.subscribe(lambda e, s=None: print(e.type))
asyncio.run(agent.prompt("Weather in Tokyo?"))

CubePi version removes:

• The StateGraph, edges, nodes, END sentinel, conditional edges.
• The ToolNode registry — tools go directly to the Agent.
• The should_continue function — the loop knows when there are tool calls.
• The State TypedDict — state lives on the agent.

Mapping common patterns

Checkpointing

# langgraph
from langgraph.checkpoint.sqlite import SqliteSaver
graph.compile(checkpointer=SqliteSaver.from_conn_string(":memory:"))

# CubePi
from cubepi.checkpointer import SQLiteCheckpointer
async with SQLiteCheckpointer("agent.db") as cp:
    agent = Agent(..., checkpointer=cp, thread_id="conv-1")

CubePi's append-only model is O(1) per message, regardless of conversation length. langgraph saves full snapshots, which scales linearly with history.

Streaming

# langgraph
for chunk in app.stream(state, stream_mode="messages"):
    if chunk["event"] == "on_chat_model_stream":
        print(chunk["data"]["chunk"].content, end="")

# CubePi
def on_event(event, signal=None):
    if event.type == "message_update" and event.stream_event.type == "text_delta":
        print(event.stream_event.delta, end="")

agent.subscribe(on_event)
await agent.prompt("…")

One subscriber, one stream — no mode flag.

Interrupting / human-in-the-loop

# langgraph
graph.compile(interrupt_before=["tools"])

# CubePi
class HumanApproval(Middleware):
    async def before_tool_call(self, ctx, *, signal=None):
        approved = await ask_human(f"Run {ctx.tool_call.name}({ctx.args})?")
        if not approved:
            return BeforeToolCallResult(block=True, reason="rejected")
        return None

Imperative interrupts via middleware. You decide per call instead of configuring graph-level interrupt points.

Branching

# langgraph
graph.add_conditional_edges("llm", lambda s: "tools" if s["messages"][-1].tool_calls else "summary")
graph.add_node("summary", summarize)
graph.add_edge("summary", END)

# CubePi
class SummariseAtEnd(Middleware):
    async def should_stop_after_turn(self, ctx) -> bool:
        msg = ctx.message
        if not any(isinstance(c, ToolCall) for c in msg.content):
            # No more tool calls; we're done. Inject a summary turn first.
            ...
            return True
        return False

There's no built-in branching primitive; flow control happens through should_stop_after_turn and after_model_response.

What langgraph does that CubePi doesn't (yet)

• Multi-agent supervisor patterns. No first-class "agents spawning agents" abstraction. You can build it by running multiple Agent instances with shared tools.
• Visual graph rendering. No app.get_graph().draw_mermaid() equivalent. CubePi's flow is linear so the picture would be a single line anyway.
• Time travel / fork at arbitrary checkpoints. The Postgres schema has fork columns but no API surface in v0.3.
• LangSmith / Langfuse integrations. Bring your own tracing via middleware + on_response.

What CubePi does that langgraph doesn't

• Native async-first — every entry point is async. No app.invoke vs. app.ainvoke split.
• Append-only persistence — O(1) DB writes, JSONB-queryable messages.
• 3 core deps vs. langchain-core + langgraph-sdk + transitives.
• Streaming-realistic test provider (FauxProvider) ships in the box.
• MCP loaders for HTTP + stdio transports.

Porting checklist

1. Replace StateGraph construction with a single Agent(...) call.
2. Move @tool-decorated functions to AgentTool instances (Pydantic models for params, async execute).
3. Replace MemorySaver / SqliteSaver / PostgresSaver with MemoryCheckpointer / SQLiteCheckpointer / PostgresCheckpointer.
4. Replace stream_mode callbacks with agent.subscribe(...).
5. Convert custom nodes that did message transforms → Middleware hooks.
6. Convert interrupt_before/after → before_tool_call / after_model_response middleware.
7. If you had a summary or route node — fold it into after_model_response with decision="stop" or "loop_to_model".

See also

• Core Concepts — the building blocks you're mapping to.
• Middleware → Composition — where flow-control logic lives.
• Checkpointing — the new persistence story.