Coming from LangGraph

Build the graph in the body of a Pyrula @agent (or @workflow), wrap its chat model in durable_chat_model, and run any tool’s outside effect through ctx.step. The run’s journal replaces the checkpointer: after a crash the body runs again from the top, the graph walks the same path, and finished model calls are answered from the journal.

To keep LangGraph in charge instead, make one node a durable Pyrula run with as_langgraph_node. Both shapes are in LangGraph.

Scaffold

pip install "pyrula[agents,langgraph,langchain]"
pyrula init --langgraph
python app.py

See Start from a scaffold.

Concept map

LangGraph	Pyrula	Change
StateGraph, nodes, edges	Unchanged, built inside an @agent or @workflow body	Direct swap
Chat model (ChatOpenAI, …)	durable_chat_model(model); pass model_id= if it has no model_name	Direct swap
model.invoke(), graph.invoke()	ainvoke; sync invoke raises UnjournaledSurface in a run	Small rewrite
model.astream()	Supported; journaled as one call	Operational change
Checkpointer	None; the run’s journal is the checkpoint	Conceptual redesign
thread_id	A Pyrula thread; read earlier turns with ctx.thread.history(...) (HTTP)	Conceptual redesign
interrupt()	await ctx.interrupt(...), optional timeout=	Small rewrite
Command(resume=...)	WorkflowClient.resume(name, run_id, action, value), /resume, or Pyrula Cloud	Small rewrite
ToolNode tool with an effect	Route the effect through ctx.step, or it runs again on resume	Small rewrite
LangChain tool in a native @agent	from_langchain_tool(tool)	Direct swap
recursion_limit	AgentLimits; a cycling graph ends with budget_exceeded	Small rewrite
Parallel model branches	Supported; give each wrapped model its own name=	Direct swap
Compiled subgraph as a node	Supported	Direct swap
create_react_agent	Supported	Direct swap
astream(stream_mode="updates")	Supported	Direct swap
astream(stream_mode="messages")	Not supported through durable_chat_model	Not supported
Node RetryPolicy	Use it on the node that calls the model	Direct swap
with_fallbacks, with_retry on the model	Not supported; a fallback would call live during replay	Not supported
Routing on model output	Must follow the recorded responses, not a clock or random value	Small rewrite
Time travel	A what-if run started from Pyrula Cloud	Conceptual redesign
Cancelling a graph run	Cancel or pause the run; each model call is a stop point	Direct swap
LangGraph as orchestrator	as_langgraph_node(...) on a checkpointed graph	Conceptual redesign
LangSmith tracing	pyrula dev, Pyrula Cloud, or OpenTelemetry (Observability)	Operational change
Where the graph runs	A worker process against a store	Operational change

Example

The graph is a plain LangGraph agent. The model keeps asking for the tool, and max_llm_calls ends the run after three calls with budget_exceeded.

import asyncio
import operator
from typing import Annotated, Any, TypedDict

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool as lc_tool
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode

from pyrula.agents import AgentContext, AgentLimits, agent
from pyrula.agents.integrations import durable_chat_model
from pyrula.agents.testing import AgentRunner

provider_calls: list[int] = []


class LoopingChat(BaseChatModel):
    """Stands in for a provider that keeps asking for the same tool."""

    model_name: str = "looping-model"

    @property
    def _llm_type(self) -> str:
        return "looping"

    def bind_tools(self, tools, **kwargs):
        return self.bind(tools=[{"name": t.name} for t in tools], **kwargs)

    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
        provider_calls.append(len(messages))
        call = {"id": f"c{len(messages)}", "name": "lookup", "args": {"order": "A-1"}}
        usage = {"input_tokens": 100, "output_tokens": 20, "total_tokens": 120}
        reply = AIMessage(content="", tool_calls=[call], usage_metadata=usage)
        return ChatResult(generations=[ChatGeneration(message=reply)])

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        raise NotImplementedError("durable_chat_model journals the async path only")


class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]


@lc_tool
async def lookup(order: str) -> str:
    """Look up an order."""
    return f"{order}: shipped"


@agent(name="docs_from_langgraph_support", timeout=60, limits=AgentLimits(max_llm_calls=3))
async def support(ctx: AgentContext, message: str) -> str:
    model = durable_chat_model(LoopingChat()).bind_tools([lookup])

    async def call_model(state: State) -> dict[str, Any]:
        return {"messages": [await model.ainvoke(state["messages"])]}

    def route(state: State) -> str:
        return "tools" if state["messages"][-1].tool_calls else END

    graph = StateGraph(State)
    graph.add_node("model", call_model)
    graph.add_node("tools", ToolNode([lookup]))
    graph.add_edge(START, "model")
    graph.add_conditional_edges("model", route, ["tools", END])
    graph.add_edge("tools", "model")

    # No checkpointer: the run's journal is the checkpoint.
    out = await graph.compile().ainvoke({"messages": [HumanMessage(content=message)]})
    return out["messages"][-1].content


async def main() -> None:
    events = await AgentRunner(support).run(message="where is A-1?")
    assert events[-1].kind == "run:error"
    assert events[-1].payload["code"] == "budget_exceeded"
    assert len(provider_calls) == 3


asyncio.run(main())

A crash and resume on this graph shape is in LangGraph.