From kitaru
Migrate existing LangGraph and LangChain agent code to Kitaru's LangGraph adapter. Use when code mentions LangGraph, StateGraph, CompiledStateGraph, graph.invoke, graph.ainvoke, stream, astream, interrupt, Command, thread_id, checkpointer, InMemorySaver, MemorySaver, create_agent, KitaruGraphRunner, LangGraphRunRequest, LangGraphRunResult, KitaruLangGraphMiddleware, build_resume_request, wait_for_interrupt, checkpoint_strategy, graph_call, calls, callbacks, event streams, Deep Agents, or checkpointers. Helps classify direct, approximate, and absent mappings, choose graph_call vs middleware-backed calls boundaries, preserve LangGraph ownership of graph state, and produce a migration report.
How this skill is triggered — by the user, by Claude, or both
Slash command
/kitaru:kitaru-langgraph-migrationThe summary Claude sees in its skill listing — used to decide when to auto-load this skill
Use this skill to inspect LangGraph, LangChain `create_agent(...)`, or Deep
Use this skill to inspect LangGraph, LangChain create_agent(...), or Deep
Agents code and move only the supportable durability boundary to Kitaru's
kitaru.adapters.langgraph surfaces.
The output should be conservative:
direct,
approximate, or absent.KitaruGraphRunner, LangGraphRunRequest,
LangGraphRunResult, and, when safe, KitaruLangGraphMiddleware.MIGRATION_REPORT.md section or file that names replay, interrupt,
checkpointer, state, stream, and side-effect risks.Do not promise that Kitaru replaces LangGraph persistence or checkpoints every node inside the graph.
LangGraph still owns graph state, checkpointers, stores, interrupts,
thread_id, graph-local replay, and any Deep Agents sandbox or filesystem
backend.
Kitaru can record either:
checkpoint_strategy="graph_call"; orKitaruLangGraphMiddleware while a KitaruGraphRunner is active with
checkpoint_strategy="calls".A useful picture: LangGraph is the house. Kitaru can put a camera at the front door and, if LangChain middleware is installed, at a few visible interior doorways. It cannot see through walls into checkpointers, stores, hosted tools, or async model/tool internals that do not pass through a true Kitaru checkpoint.
Use it when the user asks to:
graph.invoke(...) / graph.ainvoke(...) entrypoints with
KitaruGraphRunner.invoke(...) / ainvoke(...);StateGraph / CompiledStateGraph;checkpoint_strategy="graph_call" and "calls";Command(resume=...) handling;thread_id, InMemorySaver, or
MemorySaver assumptions;KitaruLangGraphMiddleware to a LangChain create_agent(...) agent;stream, astream, stream_events, or Deep
Agents under Kitaru;Do not use it to:
graph_call and calls.Classify each source pattern before editing:
direct: Kitaru has a clear adapter surface for the same outer behavior.
Example: graph.invoke(input, config) becomes
runner.invoke(LangGraphRunRequest.start(input, thread_id=...)).approximate: The migration is possible, but replay, observability,
interrupt, streaming, state, or persistence behavior differs. Example:
callbacks remain observability, not replay boundaries.absent: There is no safe automatic migration until the source is redesigned.
Example: a claim that Kitaru will replace a production LangGraph checkpointer.Unsupported patterns must not be silently approximated. Add a concrete
# TODO(migration): ... comment near proposed code, list it in the report, and
explain the redesign needed.
thread_id handling, raw invoke/ainvoke/stream calls, interrupts,
Command(resume=...), callbacks, middleware, Deep Agents usage, and existing
Kitaru flows/checkpoints.references/concept-map.md and
references/gaps-and-flags.md. Count direct, approximate, flagged, and
blocked items.checkpoint_strategy="graph_call" unless
the source clearly uses synchronous LangChain model/tool handlers that can be
wrapped by KitaruLangGraphMiddleware.KitaruGraphRunner and
serializable LangGraphRunRequest objects. Add explicit result status
handling for completed vs interrupted.MIGRATION_REPORT.md. Include the pattern inventory,
classifications, chosen boundary, flags, behavior differences, and
verification plan.Look for:
StateGraph(...), .compile(...), CompiledStateGraph, graph.invoke(...),
graph.ainvoke(...), stream(...), astream(...), or stream_events(...).config={"configurable": {"thread_id": ...}}, checkpoint_id, or
checkpoint_ns.interrupt(...), Command(resume=...), wait_for_interrupt(...), and
result status branches.InMemorySaver, MemorySaver, Postgres/Redis/custom checkpointers, stores,
or memory APIs.create_agent(...), middleware=[...], and whether
KitaruLangGraphMiddleware() is installed.@kitaru.flow or @kitaru.checkpoint wrappers.KitaruGraphRunner(graph, checkpoint_strategy="graph_call") for a
clean outer checkpoint around one completed or interrupted invocation.checkpoint_strategy="calls" only when synchronous LangChain model/tool
calls pass through KitaruLangGraphMiddleware while the runner is active.thread_id handling. If the source has no stable thread ID,
flag it before migration.wait() calls inside graph nodes or checkpoints. Bridge
LangGraph interrupts at flow scope when human input is needed.Minimal outer graph call:
from typing import TypedDict
from langgraph.graph import END, START, StateGraph
from kitaru.adapters.langgraph import KitaruGraphRunner, LangGraphRunRequest
class State(TypedDict):
topic: str
summary: str
def summarize(state: State) -> dict[str, str]:
return {"summary": f"Summary for {state['topic']}"}
builder = StateGraph(State)
builder.add_node("summarize", summarize)
builder.add_edge(START, "summarize")
builder.add_edge("summarize", END)
graph = builder.compile()
runner = KitaruGraphRunner(graph, name="summary_graph")
request = LangGraphRunRequest.start(
{"topic": "durable agents", "summary": ""},
thread_id="thread-001",
)
result = runner.invoke(request)
if result.status != "completed":
raise RuntimeError(f"Expected completed graph, got {result.status!r}")
print(result.output)
For full examples, load references/code-patterns.md.
When a pattern is unsafe or unsupported:
# TODO(migration): ... comment near proposed code.LOW, MEDIUM, HIGH, or BLOCKER.interrupt() in that node may run again."Every non-trivial migration must include or draft MIGRATION_REPORT.md with:
file:line locations;thread_id, checkpointers/stores,
interrupts/resume, middleware/calls mode, streams/events, Deep Agents, capture
policy, and side effects;Use references/report-template.md when a full report is needed.
Avoid these:
calls without KitaruLangGraphMiddleware and calling it granular
Kitaru checkpointing.awrap_model_call / awrap_tool_call metadata as true
checkpoints.InMemorySaver as production restart-durable storage.result.status == "interrupted".Load only the reference file needed for the current task:
references/concept-map.md — source-to-target mapping table and
classification guidance.references/code-patterns.md — import-complete migration examples.references/gaps-and-flags.md — severity definitions, upstream assumptions,
and must-flag patterns.references/report-template.md — LangGraph-specific migration report template.Kitaru source references:
kitaru/docs/content/docs/adapters/langgraph.mdxkitaru/src/kitaru/adapters/langgraph/__init__.pykitaru/src/kitaru/adapters/langgraph/langchain.pykitaru/examples/integrations/langgraph_agent/README.mdGuides creation, editing, and verification of skills for AI coding agents using test-driven development with subagent scenarios. Use when authoring or debugging skills.
npx claudepluginhub zenml-io/kitaru-skills --plugin kitaru