One agent per node: running the Claude Agent SDK inside LangGraph

Our last post argued that LangGraph and the Claude Agent SDK sit at different layers — one owns the flow, the other owns the work. This post is the concrete join between them, the pattern we actually run in Hive: every graph node is a full agent, not a raw API call.

The framing here owes a debt to Khaled Elfakharany’s write-up on using the Claude Agent SDK inside LangGraph nodes — same core idea, our own implementation and a few corrections for the current SDK.

The gap in every LangGraph tutorial

Open almost any LangGraph example and a node looks like this: take the state, call the model API, return a response. Clean — and lossy. A raw completion gives you a block of text. It doesn’t read a file, run a command, search the web, or manage its own context. You’ve wired up orchestration and then handed each step a lobotomized model.

That’s fine until the work gets real. “Investigate why the build broke” isn’t a completion; it’s an agent loop with tools. So the question is: can a LangGraph node keep the deterministic flow and still get the full agent underneath?

Yes — make the node body a Claude Agent SDK call.

The pattern

LangGraph decides what runs when. The SDK decides how each step runs — with which tools, which model, which context. A node is just a Python function returning a state update; nothing stops that function from driving a complete agent.

flowchart LR
  subgraph LG[LangGraph — flow, state, routing]
    START((START)) --> N1[node: research]
    N1 --> N2[node: analyze]
    N2 --> R{route}
    R -->|ok| N3[node: synthesize]
    R -->|redo| N1
    N3 --> E((END))
  end
  N1 -. SDK call .-> A1[Agent: WebSearch, WebFetch]
  N2 -. SDK call .-> A2[Agent: Read, Grep]
  N3 -. SDK call .-> A3[Agent: Write]

The wrapper

One small async wrapper runs an SDK agent and returns its text. Note the imports — the package is claude_agent_sdk (the old claude_code_sdk / ClaudeCodeOptions names are gone), and messages arrive as typed objects you match on, not dicts with a .type string.

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def run_agent(prompt: str, system: str, tools: list[str], model: str) -> str:
    """Run one Claude Agent SDK turn, return the assistant text."""
    options = ClaudeAgentOptions(
        system_prompt=system,
        allowed_tools=tools,
        model=model,                 # "haiku" | "sonnet" | "opus" — per node
        permission_mode="acceptEdits",
    )
    text = ""
    async for msg in query(prompt=prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    text += block.text
    return text

A node that uses it

async def research(state: GState) -> dict:
    out = await run_agent(
        prompt=f"Research: {state['input']}",
        system="You are a diligent research agent. Cite sources.",
        tools=["WebSearch", "WebFetch"],
        model="sonnet",
    )
    return {"outputs": {"research": out}}

LangGraph doesn’t care what happened inside — it only sees the returned state update. Inside, the agent had its full toolbox. Different nodes get different tools, different models, different system prompts.

The part that makes it sing: parallel fan-in

A node returning {"outputs": {"research": ...}} is only safe under parallelism if the state merges updates instead of overwriting them. That’s a reducer. In Hive the graph state is exactly this:

from typing import Annotated, TypedDict
import operator

def merge(a: dict, b: dict) -> dict:
    return {**a, **b}

class GState(TypedDict):
    input: str
    outputs: Annotated[dict, merge]         # each node writes outputs[name]; branches merge, not clobber
    usages: Annotated[list, operator.add]   # token/cost records accumulate across every node

Now fan-out is just two edges from the same source, and fan-in is a node with two inbound edges — LangGraph runs the branches concurrently and waits at the join:

g = StateGraph(GState)
for name, node in [("pros", pros), ("cons", cons), ("verdict", verdict)]:
    g.add_node(name, node)
g.add_edge(START, "pros")     # same source ⇒ pros and cons run in parallel
g.add_edge(START, "cons")
g.add_edge("pros", "verdict") # two inbound edges ⇒ verdict waits for both (fan-in)
g.add_edge("cons", "verdict")
g.add_edge("verdict", END)
app = g.compile()

flowchart TB
  START((START)) --> pros[pros agent
model: sonnet]
  START --> cons[cons agent
model: sonnet]
  pros --> verdict[verdict agent
model: opus]
  cons --> verdict
  verdict --> E((END))
  pros -. writes .-> O[(outputs · merge reducer)]
  cons -. writes .-> O
  O -. reads both .-> verdict

Two agents argue in parallel, a third weighs the merged result. Each is a separate SDK call with its own tools and model — and because the reducer merges, neither branch overwrites the other.

Why it’s worth it

Benefit	What you get
Tool isolation	Each agent gets exactly the tools it needs — the researcher has web access, the summarizer has none. Smaller blast radius, cleaner prompts.
Model per node	Route with haiku, analyze with sonnet, reason with opus. You pay for capability only where it matters.
Parallel execution	Fan out five agents at once; the reducer collects their results. Wall-clock is the slowest branch, not the sum.
State visibility	outputs shows exactly what each node produced. Inspect, replay, debug one node in isolation.
Deterministic flow	The graph is code. No guessing what an orchestrator-model will decide next.

How we run it in Hive

Hive’s Brain (a Claude Agent SDK runtime) exposes a /graph endpoint. A workflow’s node/edge spec is compiled into exactly the StateGraph above; each node calls the SDK; the run streams per-node events back into the chat so you watch “pros ✓ · cons ✓ · verdict…” live. Same-source nodes run in parallel, multi-input nodes fan in, and every node’s tokens land in usages for cost accounting.

It’s the graph orchestration style sitting next to the plain delegate-orchestrator — pick a deterministic graph when the flow is known, an orchestrator when it isn’t.

The tradeoffs (this is not free)

• Two systems to run. LangGraph for flow, the SDK for execution. More moving parts, more to reason about.
• Latency stacks. Every node is a full agent turn. In a long chain the per-call overhead adds up — parallelize what you can.
• Errors need catching inside nodes. Timeouts, rate limits, a tool that fails. A node that raises drops its branch; decide whether that fails the run or degrades it.
• State can balloon. Accumulating full agent outputs gets heavy. Summarize what you pass forward instead of forwarding everything.

Reach for this when the flow has conditional branching, parallel steps that aggregate, or different tool/model needs per step — and you need to debug it later. Skip it when the task is a single agent or a straight line; there, just call the SDK directly. A twelve-node graph for something an agent loop solves in three turns is self-inflicted cost.

Closing

The trick isn’t clever — it’s a boundary. LangGraph holds the flow and the state; the Claude Agent SDK holds the action. Keep each node a function that returns a state update, put a full agent inside it, and let a reducer merge the parallel branches. You get deterministic workflows with genuinely capable agents at every step — which is exactly what “multi-agent system” should have meant all along.

---

References: Claude Agent SDK — Python · LangGraph · pattern inspired by Khaled Elfakharany, “I Found a Way to Use Claude Agent SDK Inside LangGraph Nodes.”