Building with LangGraph

from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from langchain_anthropic import ChatAnthropic from langchain_mcp_adapters.client import MultiServerMCPClient from typing import TypedDict, Annotated import operator class State(TypedDict): question: str plan: list[str] findings: Annotated[list[dict], operator.add] draft: str verdict: str model = ChatAnthropic(model="claude-opus-4-7", temperature=0) async def plan_node(state: State): resp = await model.ainvoke( f"Break this research question into 3-5 sub-questions:\n{state['question']}" ) return {"plan": resp.content.split("\n")} async def research_node(state: State): mcp = MultiServerMCPClient({ "brave": {"command": "npx", "args": ["-y", "@mcp/brave-search"]}, }) tools = await mcp.get_tools() agent = create_react_agent(model, tools) findings = [] for sub in state["plan"]: r = await agent.ainvoke({"messages": [("user", sub)]}) findings.append({"sub": sub, "answer": r["messages"][-1].content}) return {"findings": findings} async def synthesize_node(state: State): context = "\n\n".join(f.get("answer", "") for f in state["findings"]) resp = await model.ainvoke( f"Write a cited answer to: {state['question']}\n\nNotes:\n{context}" ) return {"draft": resp.content} async def verify_node(state: State): resp = await model.ainvoke( f"Rate this answer's faithfulness to the notes (pass/fail + reason):\n" f"Q: {state['question']}\nA: {state['draft']}" ) return {"verdict": resp.content} graph = StateGraph(State) graph.add_node("plan", plan_node) graph.add_node("research", research_node) graph.add_node("synthesize", synthesize_node) graph.add_node("verify", verify_node) graph.set_entry_point("plan") graph.add_edge("plan", "research") graph.add_edge("research", "synthesize") graph.add_edge("synthesize", "verify") graph.add_edge("verify", END) app = graph.compile(checkpointer=MemorySaver())

from langgraph.types import interrupt, Command async def approve_node(state: State): response = interrupt({ "draft": state["draft"], "ask": "Approve to publish? Edit below if needed.", }) return {"draft": response["edited_draft"]} # In the client: config = {"configurable": {"thread_id": "run-42"}} async for event in app.astream({"question": ""}, config): # stream until interrupt # Inspect state, get human input, then resume: await app.ainvoke( Command(resume={"edited_draft": "<human's edits>"}), config )