Appendix C: API Quick Reference

The core signatures this book relies on, grouped by concern, each one pointing back to the chapter that verified it and explains why it’s shaped the way it is.

These are the core signatures, not every one the book prints. Each entry was verified against the pinned packages while drafting the chapter listed, and that chapter carries the reasoning behind the shape. Treat this appendix as a lookup table, not a substitute for the argument.

Graph construction and execution

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AtlasState)
builder.add_node(
    name, fn, *,
    retry_policy=RetryPolicy(...), timeout=float | TimeoutPolicy(...),
    error_handler=fn,
)
builder.add_edge(source, target)
builder.add_conditional_edges(source, path_fn, path_map=None)
# path_fn may return list[Send] instead of a single node name (Ch17)
graph = builder.compile(checkpointer=..., store=...)
graph.invoke(input, config, *, context=..., durability="sync", control=...)

Introduced/verified: add_node and compile in Ch4, checkpointer= in Ch9, store= in Ch13. On invoke/ainvoke, durability= ("sync", "async", or "exit") and control= come from Ch10, context= from Ch17 and Ch23. recursion_limit via config={"recursion_limit": N}, default 10007 for a graph you compile (create_agent sets 9,999, Ch2), raises GraphRecursionError (Ch6).

create_agent and middleware

from langchain.agents import create_agent

agent = create_agent(
    model, tools=[...], system_prompt=..., middleware=[...],
    response_format=..., context_schema=..., checkpointer=..., store=..., name=...,
)

Introduced/verified across Ch2 (create_agent), Ch7 (response_format=), Ch8 (middleware=), Ch20 (name=), Ch23 (context_schema=).

from langchain.agents.middleware import (
    AgentMiddleware, ToolCallRequest, ModelRequest, ModelResponse,
    PIIMiddleware, SummarizationMiddleware, HumanInTheLoopMiddleware,
)

Hooks: before_agent, before_model, after_model, after_agent, wrap_model_call, wrap_tool_call (Ch8).

create_deep_agent

from deepagents import create_deep_agent
from deepagents.backends.store import StoreBackend

agent = create_deep_agent(
    model, tools=[...], system_prompt=..., middleware=[...],
    subagents=[{"name": ..., "description": ..., "system_prompt": ..., "tools": [...]}],
    backend=StoreBackend(store=store, namespace=namespace_fn),  # fn of the Runtime
    interrupt_on={...}, permissions=[...], checkpointer=..., store=...,
)

Introduced Ch18, on deepagents 0.6.x, a beta package. A sub-agent spec needs name, description, and system_prompt; one with no tools key inherits the main agent’s tools, and middleware= on the call does not reach sub-agents, so each spec lists its own.

A spec named "general-purpose" replaces the harness’s default sub-agent (Appendix E). With no backend=, files live in the thread’s own state; StoreBackend puts them in the store, under the namespace its function returns.

Persistence: checkpointer and store

from langgraph.checkpoint.memory import InMemorySaver          # dev only
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver  # prod
from langgraph.store.memory import InMemoryStore                # dev only
from langgraph.store.postgres import PostgresStore               # prod

Both introduced in Ch9 (checkpointer) and Ch13 (store). With a checkpointer, thread_id is required in config["configurable"]. graph.get_state(config) returns a StateSnapshot (.values, .next); graph.get_state_history(config) returns an iterable of snapshots for time travel (Ch9).

from langgraph.store.base import IndexConfig, TTLConfig

store.put(namespace, key, value, index=None, *, ttl=minutes)  # no compare-and-swap
store.get(namespace, key)  # one Item, or None
store.search(namespace_prefix, query=..., filter=..., limit=10)  # list[SearchItem]
InMemoryStore(index=IndexConfig(embed=embeddings, dims=1536, fields=["$"]))
with PostgresStore.from_conn_string(  # a context manager, not a store
    dsn, index=..., ttl=TTLConfig(default_ttl=...)
) as store: ...

Introduced Ch13; get/put as an upsert in Ch14; TTLConfig and the sweeper in Ch24. A namespace is a tuple of strings. search matches by namespace prefix, which PostgresStore runs as LIKE, so keep only items whose .namespace equals the one you meant (Ch13).

query= ranks by similarity only on a store built with index=. ttl= is in minutes, and InMemoryStore raises NotImplementedError on it. An expired item is deleted by a sweep (sweep_ttl(), or the background sweeper start_ttl_sweeper() starts), not hidden on read (Ch24).

Control flow: interrupt, Command, Send

from langgraph.types import interrupt, Command, Send

decision = interrupt(payload)  # suspends the run; resuming needs a checkpointer
# resume: graph.invoke(Command(resume=decision_value), config)
# Command(update=, goto=) updates state and routes in one move
# Send(node, arg, *, timeout=None) fans out one superstep across parallel branches

interrupt/Command(resume=) introduced in Ch11; Command(update=, goto=) for routing in Ch6; Send in Ch17. A resumed node re-runs from the top, not from the line after interrupt() (Ch11).

Streaming

from langgraph.config import get_stream_writer

for chunk in graph.stream(
    input, config,
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,  # also yield events from inside subgraphs
    version="v2",  # one dict shape for every combination of modes
):
    kind, ns, data = chunk["type"], chunk["ns"], chunk["data"]  # mode, ns, payload

get_stream_writer()({"progress": "..."})  # inside a node or tool: the "custom" mode

stream_mode introduced Ch4; the rest Ch19. The modes are values, updates, messages, custom, debug, checkpoints, and tasks; messages data is a (chunk, metadata) pair.

version defaults to "v1", where one mode yields the bare chunk, a list yields (mode, chunk), and subgraphs=True makes it (ns, mode, chunk), so pass version="v2" explicitly. ns is () for the streamed graph’s own nodes and ("node:<task_id>",) inside a subgraph that node started. astream takes the same arguments.

Fault tolerance

from langgraph.types import RetryPolicy, TimeoutPolicy

RetryPolicy(max_attempts=3)  # retry_on defaults to default_retry_on (Ch4)
RetryPolicy(max_attempts=3, retry_on=(RefundError,))  # narrowed on purpose (Ch10)
TimeoutPolicy(run_timeout=30.0, idle_timeout=10.0)  # async nodes only

RetryPolicy from Ch4, narrowed in Ch10; TimeoutPolicy from Ch10. The order is fixed: the node runs under its timeout, and an exception goes to RetryPolicy. error_handler= runs once the policy stops, either because the attempts are spent or because retry_on does not match the exception, which then gets one attempt and no retry.

from langgraph.runtime import RunControl
from langgraph.errors import GraphDrained

control = RunControl()
await graph.ainvoke(input, config, control=control)
control.request_drain()  # any thread; stops at the next superstep boundary

Introduced Ch10. GraphDrained is raised in the process that owns the run, so it is not observable across a process boundary. Behind the Agent Server there is no RunControl: the server drains itself on SIGTERM (Ch22), and one run is stopped with client.runs.cancel(thread_id, run_id) (Ch23).

Functional API: @entrypoint and @task

from langgraph.func import entrypoint, task

@task(retry_policy=RetryPolicy(max_attempts=3), cache_policy=...)
def step(arg: str) -> dict: ...

@entrypoint(checkpointer=InMemorySaver(), store=..., cache=...)
def workflow(inputs: list[str], *, previous: list | None = None) -> list:
    return [f.result() for f in [step(i) for i in inputs]]  # a task returns a future

Covered in Appendix F. A task can be called only inside a run, and called bare it raises RuntimeError. A bare @task has no retry and no cache, and a task’s cache_policy= does nothing without the entrypoint’s cache=.

previous is the last return value on the same thread_id, and only with a checkpointer. workflow.invoke(None, config) resumes a failed run and re-runs only the tasks that did not finish.

MCP

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {name: {"transport": "stdio", "command": ..., "args": [...]}}
)
tools = await client.get_tools()  # raises if a server cannot connect

Introduced Ch7. A server that fails to connect makes get_tools() raise. A server that connects but exposes fewer tools returns a short list without complaint, so assert the tools you need by name and refuse to start otherwise.

Tracing and evaluation

from langsmith import traceable, trace, Client

@traceable(run_type="tool", name=...)
with trace(name=..., tags=[...], metadata={...}):
    ...
from langsmith.anonymizer import create_anonymizer
import langsmith as ls
ls.configure(client=Client(anonymizer=create_anonymizer([...])))   # trace redaction

All introduced Ch20. The anonymizer masks every span’s inputs and outputs, and only an installed client is used. It is distinct from the stream-side redaction Ch19 covers - the two are separate pipelines and neither implies the other.

from langsmith.evaluation import evaluate
from agentevals.trajectory.match import create_trajectory_match_evaluator
from openevals.llm import create_llm_as_judge

evaluate(target, data=, evaluators=, experiment_prefix=...)
create_trajectory_match_evaluator(
    trajectory_match_mode="strict" | "unordered" | "subset" | "superset"
)
create_llm_as_judge(prompt=, feedback_key=, model=)

All introduced Ch21. agentevals and openevals are pre-1.0 langchain-ai packages - pin the version, expect churn, the same caveat this book gives LangMem (Ch14).

Deployment and scheduling

from langgraph_sdk import get_client

client = get_client(url="http://localhost:8123")   # matches a local `langgraph up`
await client.crons.create(assistant_id=, schedule=, input=, webhook=)   # stateless
await client.crons.create_for_thread(
    thread_id=, assistant_id=, schedule=, input=
)   # stateful
await client.runs.create(thread_id=, assistant_id=, input=, webhook=)
await client.runs.cancel(thread_id, run_id)  # stop one run (Ch23)

Introduced Ch22, except runs.cancel. Cron schedules are standard cron expressions, interpreted in UTC unless you pass timezone=. A stateless cron’s thread is deleted after the run unless you pass on_run_completed="keep".

A graph you export in langgraph.json is compiled without a checkpointer or store: the Agent Server supplies both, and langgraph dev refuses a graph that brings its own (Ch22). The CLI and the server are not part of the pinned stack, so that refusal was checked against the langgraph-api 0.14.0 source Ch22 runs, where it applies only under langgraph dev.

Agent Server auth

from typing import Any

from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate(authorization: str | None) -> Auth.types.MinimalUserDict:
    ...  # raise Auth.exceptions.HTTPException(status_code=401) to refuse
    return {"identity": ..., "permissions": [...]}

@auth.on  # every resource; @auth.on.threads, .crons, .store, ... are narrower
async def deny_by_default(ctx: Auth.types.AuthContext, value: Any) -> bool:
    return False

Introduced Ch23, registered in langgraph.json as "auth": {"path": "./atlas/auth.py:auth"}. ctx carries user, permissions, resource, and action. A handler returns None or True to allow, False to refuse with a 403, or a filter dict such as {"owner": ...} that limits what the caller sees. The SDK at the pin documents these returns, and the langgraph-api 0.14.0 source enforces them that way.

Stability notes

Beta or pre-1.0 surfaces referenced in this book, each paired with a stable fallback where one exists: DeltaChannel (LangGraph 1.2+, beta), stream_events(version="v3") (beta, version="v2" fallback), deepagents (0.6.x, beta classifier), LangMem/agentevals/openevals (all 0.0.x-to-pre-1.0, langchain-ai). Verify each against current release notes before depending on it in production - Chapter 26 is the reference for how to do that verification, not just what to verify.