The Functional API: @entrypoint, @task, and When to Use It (LangGraph in Production, Module 13)

This is Module 13 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

---

A colleague shows you twenty lines of Python

You’re reviewing a pull request. The description says it rewrites Forge’s test-fix loop. You open the diff expecting nodes, edges, add_conditional_edges, and a build_graph() function. Instead you see this:

from langgraph.func import entrypoint, task

@task
def run_tests_task(repo_path: str):
    return sandbox.run_tests(repo_path)

@entrypoint(checkpointer=checkpointer)
def forge_app(payload: dict) -> dict:
    result = run_tests_task(payload["repo_path"]).result()
    attempts = 1
    while not result.passed and attempts < MAX_ATTEMPTS:
        edit_file(...).result()
        result = run_tests_task(payload["repo_path"]).result()
        attempts += 1
    return {"passed": result.passed, "attempts": attempts}

No StateGraph. No add_node. No conditional edges. Just a while loop. But the PR description says persistence works, interrupt() works, and streaming works. The implementation is LangGraph. It just looks like plain Python.

Two questions surface immediately. First: if this does everything the graph does, why did you spend twelve modules building a StateGraph? Second: when does this style actually win? This module answers both — by building the Functional API variant of Forge’s test-fix loop in forge/app.py, running it on the same Tally fixture as the Graph API build, and comparing them concretely.

---

In this module

You’ll learn

• How @task and @entrypoint work and what they share with the Graph API at the runtime level (CA13.1)
• How previous carries short-term state across invocations on the same thread (CA13.2)
• How entrypoint.final(value=, save=) decouples what the caller receives from what is persisted (CA13.3)
• How durability, HITL, and streaming work in the Functional API — identically to the Graph API (CA13.4)
• How to choose between the two APIs using a concrete decision table (CA13.5 + CA13.6)

You’ll build A Functional API variant of Forge’s test-fix loop in forge/app.py — @entrypoint + @task units for editing files and running tests — running on the same Tally sandbox, the same checkpointer, and the same Pregel runtime as the Graph API.

Concepts covered CA13 — The Functional API (core). This module BUILDs CA13.1–CA13.6 of the LangGraph theory inventory. It also re-exercises CA7.1 (interrupt() + Command(resume=)) and CA9.3/CA9.4 (RetryPolicy/CachePolicy on tasks) in the new context.

Prerequisites Modules 1–12 complete; the cumulative forge/ package through M12 (ForgeState with all fields through review, subgraphs, Supervisor, SqliteSaver, the Store, streaming). A .env with ANTHROPIC_API_KEY for live runs, or FORGE_MODEL_PROVIDER=ollama for a free local alternative.

---

Where you are

M1–M12  ✅  StateGraph, persistence, HITL, time-travel, durable execution,
            long-term memory, streaming, subgraphs, Supervisor
M13     👉  The Functional API — @entrypoint/@task variant of the test-fix loop
M14–M16 ⬜  Deploy Forge: langgraph.json, LangGraph Platform/Studio, LangSmith eval

Forge before M13: 100% Graph API. All logic lives in build_graph() — StateGraph + conditional edges + subgraphs + Supervisor, compiled with the checkpointer and the Store.

Forge after M13: forge/app.py adds a Functional API variant of the coder loop. The main graph in forge/graph.py is untouched and stays the primary path. app.py is a parallel variant, not a replacement.

Next up in M14: Forge ships — langgraph.json, langgraph dev, LangGraph Platform and Studio.

---

@task and @entrypoint: Pure Python, Same Pregel (CA13.1)

The Functional API exists because not every workflow is a good fit for a declarative graph. Nodes, edges, and add_conditional_edges are powerful precisely because they are explicit and visualizable — but they can be verbose when the control flow is just a while loop with an if inside it. The Functional API lets you write that control flow in Python, with all the same runtime guarantees underneath.

The canonical import in v1.x — the only correct one — is:

from langgraph.func import entrypoint, task

Do not use from langgraph.experimental import .... That path existed in pre-1.0 betas and no longer works. Do not capitalize these names — Entrypoint and Task are not v1.x symbols.

@task: a unit of work that returns a future

@task decorates a regular Python function and makes it a unit of work in the Pregel runtime. When you call a @task-decorated function, it does not execute immediately. It returns a future — you get the result by calling .result() (synchronous) or by await-ing the future in an async context.

That sounds subtle, but the consequence is important: every @task call is tracked by the runtime. If the workflow crashes and is resumed from a checkpoint, tasks that already completed are not re-run — their results are replayed from the checkpoint. This is the same idempotence model as adding retry_policy= and cache_policy= to a node in the Graph API (CA9.3/CA9.4, from M9), and you can attach those here too:

from langgraph.func import task
from langgraph.types import RetryPolicy, CachePolicy

@task(retry_policy=RetryPolicy(max_attempts=3, retry_on=Exception))
def run_tests_task(repo_path: str):
    """Sandboxed pytest run — retries on transient subprocess failure."""
    return sandbox.run_tests(repo_path)

This is exactly the pattern Forge uses for the Tester in the Graph API (add_node("tester", nodes.tester, retry_policy=...)). The Functional API exposes the same knobs on @task.

@entrypoint: the workflow

@entrypoint is the workflow container. The function it decorates becomes an object that exposes the full LangGraph workflow interface: .invoke(), .stream(), .get_state(), .get_state_history(), and astream_events(). It accepts a checkpointer and, optionally, a store and a cache:

from langgraph.func import entrypoint
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

conn = sqlite3.connect("forge_functional.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
checkpointer.setup()

@entrypoint(checkpointer=checkpointer)
def forge_app(payload: dict) -> dict:
    # Plain Python — no StateGraph, no add_node, no add_edge.
    repo = payload["repo_path"]
    result = run_tests_task(repo).result()
    ...
    return {"passed": result.passed}

Notice that .setup() is called on the checkpointer before the decorator — the Functional API shares the same SqliteSaver that build_graph() uses. They go to the same database, the same tables. That is the point: the runtime underneath is identical.

The shared Pregel runtime

The key insight of the Functional API is that both APIs run on the same Pregel engine. Pregel is the graph-processing model that LangGraph uses internally — a superstep-based execution model where work proceeds in rounds separated by synchronization barriers. When you use the Graph API, LangGraph compiles your StateGraph into a Pregel execution plan. When you use the Functional API, LangGraph compiles the @task calls in your @entrypoint into the same kind of Pregel execution plan. The checkpoint tables, the thread_id-based state, the streaming infrastructure — all shared.

graph TB
    subgraph "Graph API — forge/graph.py"
        direction TB
        A[intake] --> B[triage]
        B --> C[recall]
        C --> D[planner]
        D --> E[plan_approval<br>interrupt]
        E --> F[editor × N<br>Send fan-out]
        F --> G[apply_edits<br>defer=True]
        G --> H[tester<br>RetryPolicy]
        H -->|passed| I[reviewer]
        H -->|fail| J[fixer]
        J --> H
        I --> K[pr_approval<br>interrupt]
    end

    subgraph "Functional API — forge/app.py"
        direction TB
        P["@entrypoint forge_app(payload)"] --> Q["@task edit_file(repo, path, ...)"]
        Q --> R["@task run_tests_task(repo)"]
        R -->|passed| S[return result]
        R -->|fail| T["@task edit_file(...fix...)"]
        T --> R
    end

    subgraph "Shared Pregel Runtime"
        direction LR
        U[(SqliteSaver<br>checkpointer)]
        V[(InMemoryStore<br>or PostgresStore)]
        W[Streaming<br>infrastructure]
    end

    G -.->|same DB| U
    P -.->|same DB| U
    C -.->|same Store| V
    P -.->|same Store| V

Both topologies draw on the same infrastructure. The graph on the left is what you built in M1–M12. The workflow on the right is what this module adds in forge/app.py.

---

previous: Short-Term State Across Runs (CA13.2)

The previous parameter is the Functional API’s most useful ergonomic feature. In the Graph API, continuity between two invocations on the same thread comes automatically from the checkpointer — the graph state is restored. In the Functional API, you express the same idea explicitly: declare a previous keyword argument, and the runtime injects the value returned by the last invocation on that thread.

@entrypoint(checkpointer=checkpointer)
def forge_app(payload: dict, *, previous: dict | None = None) -> dict:
    # First call on a thread: previous is None.
    # Subsequent calls on the same thread: previous is the dict this function returned last time.
    prior_attempts = (previous or {}).get("total_attempts", 0)
    repo = payload["repo_path"]

    result = run_tests_task(repo).result()
    attempts = 1
    while not result.passed and attempts < 3:
        edit_file(repo, payload["targets"][0], f"fix: {result.output_tail[:80]}", payload["issue"]).result()
        result = run_tests_task(repo).result()
        attempts += 1

    return {
        "passed": result.passed,
        "attempts": attempts,
        "total_attempts": prior_attempts + attempts,
    }

Two things to keep straight:

• previous is per-thread, short-term state. It carries state from run N to run N+1 on the same thread_id. If you start a new thread, previous is None.
• The Store (from M10) is cross-thread, long-term memory. If you want to share Forge’s fix history across two different issues being worked in parallel, you use the Store with a namespace, not previous. previous and the Store serve distinct purposes and you often want both.

The test in test_module_13.py demonstrates this clearly:

def test_entrypoint_previous_accumulates():
    @entrypoint(checkpointer=checkpoint.in_memory())
    def counter(n: int, *, previous=None) -> int:
        total = (previous or 0) + n
        return entrypoint.final(value=total, save=total)

    cfg = {"configurable": {"thread_id": "c"}}
    assert counter.invoke(1, cfg) == 1
    assert counter.invoke(2, cfg) == 3   # previous=1, +2
    assert counter.invoke(10, cfg) == 13

On the second .invoke(2, cfg) call, previous equals 1 (the return value of the first call). On the third, previous is 3. The thread carries the accumulation. Switch to a different thread_id and you restart from None.

---

entrypoint.final: Decoupling Output from Persistence (CA13.3)

By default, whatever your @entrypoint function returns is also what gets saved as previous for the next run. entrypoint.final lets you separate those two things: value is what the caller of .invoke() receives, and save is what becomes previous next time.

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def forge_app(payload: dict, *, previous=None) -> dict:
    prior_attempts = (previous or {}).get("total_attempts", 0)
    # ... run the loop ...
    result = run_tests_task(payload["repo_path"]).result()

    # The caller gets a short summary. The runtime saves richer context for the next run.
    return entrypoint.final(
        value={"status": "done", "passed": result.passed},
        save={
            "total_attempts": prior_attempts + attempts,
            "last_issue": payload["issue"],
            "last_result": result.model_dump(),
        },
    )

The caller of forge_app.invoke(...) receives {"status": "done", "passed": True}. The next call to forge_app on the same thread receives {"total_attempts": 2, "last_issue": "...", "last_result": {...}} as previous.

When is this useful?

• Lean API surface: callers don’t need or want the full internal context — they want a verdict, not a state dump.
• Security: avoid persisting secrets, tokens, or large intermediate results that would be wasteful to re-inject.
• Versioning: the save payload is the serialized form that future invocations will receive. You can evolve it independently of what you expose to the caller.

Common misconception: @entrypoint replaces the checkpointer.

It does not. @entrypoint requires a checkpointer to make previous work at all. Without checkpointer= in the decorator call, previous is always None and state is never persisted between runs — exactly like a StateGraph.compile() with no compile(checkpointer=...). The Functional API is a style for writing workflows. The checkpointer is the persistence mechanism. You need both.

(v0 → v1 freshness trap): Pre-1.0 documentation described the Functional API as “LangGraph Functional API (beta)” with different class names and import paths. In v1.x, the only correct import is from langgraph.func import entrypoint, task — lowercase, from langgraph.func. Any documentation that mentions from langgraph.experimental import Entrypoint or uses capitalized class names is describing an API that no longer exists.

---

Durability, HITL, and Streaming in the Functional API (CA13.4)

This is where the shared Pregel runtime pays off concretely. The three capabilities you built in M7, M9, and M11 work identically in @entrypoint-based workflows.

Durability

Durability mode — the durability= parameter from M9 — applies to .invoke() and .stream() calls on an @entrypoint the same way it applies to a compiled graph:

result = forge_app.invoke(
    payload,
    config={
        "configurable": {"thread_id": "forge-run-1"},
        "durability": "sync",    # checkpoint after every task, not just on exit
    },
)

The three modes are "exit" (checkpoint only on clean finish), "async" (checkpoint after each superstep, non-blocking), and "sync" (checkpoint after each superstep, blocking). Never checkpoint_during= — that parameter was deprecated and removed.

Human-in-the-loop

interrupt() and Command(resume=) from langgraph.types work in the body of an @entrypoint. The re-execution trap (CA7.2) applies exactly as in the Graph API: when the workflow resumes from an interrupt, the @entrypoint body re-executes from the beginning up to the interrupt() call. Every @task called before the interrupt replays its result from the checkpoint without re-running the underlying function. Everything must be idempotent.

from langgraph.types import interrupt, Command

@entrypoint(checkpointer=checkpointer)
def forge_app(payload: dict, *, previous=None) -> dict:
    repo = payload["repo_path"]
    issue = payload["issue"]

    # Propose a plan (this @task result is replayed on resume — not re-run).
    plan = propose_plan_task(repo, issue).result()

    # Pause for human approval.
    decision = interrupt({"action": "approve_plan", "summary": plan.summary})
    if not (decision.get("approved") if isinstance(decision, dict) else decision):
        return entrypoint.final(value={"status": "rejected"}, save=None)

    # Continue with the approved plan.
    futures = [edit_file(repo, t["path"], t["instruction"], issue) for t in plan.targets]
    edits = [f.result() for f in futures]
    result = run_tests_task(repo).result()
    ...

To resume: forge_app.invoke(Command(resume={"approved": True}), config=config). The same thread_id in config["configurable"] is how the runtime finds the right checkpoint.

One constraint: interrupt() must be called in the @entrypoint body, not inside a @task. A @task is a unit of atomic work — it either completes or fails; it cannot pause mid-execution for human input. Place your interrupt gates in the @entrypoint body, between task calls.

Streaming

get_stream_writer from langgraph.config works inside @task functions. You emit custom events the same way you do in graph nodes:

from langgraph.config import get_stream_writer

@task
def edit_file(repo_path: str, target_path: str, instruction: str, issue: str):
    writer = get_stream_writer()
    writer({"progress": f"editing {target_path}"})
    # ... do the edit ...
    return FileEdit(path=target_path, new_content=content, summary=instruction)

On the caller side, .stream(payload, config, stream_mode="custom") or astream_events(version="v2") — same API as on a compiled graph:

for chunk in forge_app.stream(payload, config, stream_mode="custom"):
    print(chunk)  # {"progress": "editing report.py"}, {"progress": "running tests"}, ...

@task with retry_policy and cache_policy

The actual forge/app.py in the lab uses @task without additional arguments because the test-fix loop is the focus. But the full option set is available — and the syntax maps directly to the Graph API equivalent:

Graph API	Functional API
add_node("tester", nodes.tester, retry_policy=RetryPolicy(...))	@task(retry_policy=RetryPolicy(...))
add_node("planner", nodes.planner, cache_policy=CachePolicy(...))	@task(cache_policy=CachePolicy(...))

from langgraph.types import RetryPolicy, CachePolicy
from langgraph.cache.memory import InMemoryCache

@task(
    retry_policy=RetryPolicy(max_attempts=3, retry_on=Exception),
    cache_policy=CachePolicy(key_func=lambda r, **_: r, ttl=120),
)
def run_tests_task(repo_path: str):
    return sandbox.run_tests(repo_path)

Each @task is an independent unit of replay for the durability model (CA9.5): if a workflow crashes mid-run and resumes from a checkpoint, only the tasks that had not yet completed re-execute. Tasks that already finished replay their results from the checkpoint.

---

Functional API vs Graph API: a Decision Table (CA13.5 + CA13.6)

Here is the comparison you actually need when choosing between the two.

Criterion	Functional API (@entrypoint/@task)	Graph API (StateGraph)
Code style	Plain Python — if, while, for, functions — readable by any Python developer	Declarative — nodes + wired edges; explicit, inspectable, self-documenting
Visualization	No node-level graph (Studio shows a generic “entrypoint” outline)	Full graph rendered in Studio — visual debugging, per-node time-travel
Complex branching	Natural — branching logic lives in Python	Explicit but can become verbose (add_conditional_edges + path map)
Dynamic routing	if/match in Python — trivial	add_conditional_edges + routing function
Fan-out / map-reduce	[task(...).result() for ...] or asyncio.gather — fine for N ≤ ~20	the Send API (Send + defer=True) — more explicit, scales to large N
Persistence	Checkpointer required (@entrypoint(checkpointer=))	compile(checkpointer=)
HITL	interrupt() + Command(resume=) in @entrypoint body — identical	interrupt() + Command(resume=) in node body — identical
Streaming	.stream()/astream_events() — identical	.stream()/astream_events() — identical
Durability	durability= on .invoke()/.stream() — identical	durability= on .invoke()/.stream() — identical
Retry / cache per unit	@task(retry_policy=, cache_policy=)	add_node(..., retry_policy=, cache_policy=)
Short-term session state	previous — native, injected automatically	Via the graph state + checkpointer (implicit)
Output/save decoupling	entrypoint.final(value=, save=) — native	Via output schema filtering or state pruning
Team readability	Requires reading the implementation to understand the flow	Flow is visible in the graph topology; Studio shows it without reading code
When to choose	Pure algorithmic flow; rapid prototyping; batch/ETL jobs; solo developer who prefers reading Python	Complex topologies with multi-step branching, cycles, multi-agent coordination (Supervisor), subgraphs; Studio debugging; team collaboration
Forge case	forge/app.py — the test-fix loop is linear, simple, a good fit	forge/graph.py — the full pipeline (subgraphs, Supervisor, multi-step HITL) stays here

The shared Pregel runtime (CA13.6)

The table above may suggest that the two APIs are rivals. They are not — they run on the same Pregel runtime. An @entrypoint with a SqliteSaver checkpointer and a compiled StateGraph with the same SqliteSaver share the same checkpoint tables. The Store (InMemoryStore or PostgresStore) can be shared too: @entrypoint(checkpointer=checkpointer, store=store) and build_graph(checkpointer=checkpointer, store=store) point at the same underlying storage.

This means you can mix styles in the same project without friction. forge/graph.py handles the full pipeline with its Supervisor, subgraphs, and multi-step HITL. forge/app.py handles the inner coder loop as a lightweight functional variant. They share the same checkpointer and the same Store. Nothing in the persistence layer needs to change when you switch styles.

An opinionated recommendation

For Forge, the Graph API stays the primary path. The Functional variant in app.py exists to teach the comparison — and to cover cases where you want a lightweight alternative for a specific sub-workflow. In a team context, the graph wins: it is self-documenting, Studio-debuggable, and every teammate can visualize the topology without reading the implementation. The Functional API is the right tool when the flow is genuinely simple Python logic and you want to keep it that way.

---

Hands-on Lab: Build forge/app.py

Objective: implement forge/app.py — a Functional API variant of Forge’s coder loop — that shares the same Tally sandbox and checkpointer as forge/graph.py, and passes the same tests.

Observable result:

$ uv run pytest tests/test_module_13.py -v
tests/test_module_13.py::test_functional_forge_app_fixes_bug PASSED
tests/test_module_13.py::test_entrypoint_previous_accumulates PASSED
2 passed in 2.31s

$ uv run pytest tests/ -q
...................................    # M1–M12 and M13 tests
35 passed in 5.40s

Step 1: Start from the M12 codebase

Copy the cumulative forge/ package from M12 exactly. The following files are inherited unchanged: config.py, state.py, models.py, graph.py, nodes.py, tools.py, agents.py, sandbox.py, memory.py, subgraphs.py, supervisor.py, checkpoint.py. The only new file in M13 is forge/app.py.

Confirm your pins in pyproject.toml:

[project]
requires-python = ">=3.12"

[tool.uv.sources]
# Pins as of June 2026 — re-verify on PyPI before running
langgraph = {version = "~=1.2"}         # tested at 1.2.5
langchain = {version = "~=1.3"}         # tested at 1.3.9
langchain-core = {version = "~=1.4"}    # tested at 1.4.7
langchain-anthropic = {version = "~=1.4"}  # tested at 1.4.6
langgraph-checkpoint-sqlite = {version = "~=3.1"}  # tested at 3.1.0

Step 2: The @task workers

Open forge/app.py. The module docstring makes the relationship to the Graph API explicit — this is a parallel variant, not a replacement:

"""The Functional API (Module 13): the test-fix loop as @entrypoint + @task.

An alternative to the Graph API for plain-Python control flow. Same runtime (Pregel),
same durability/HITL/streaming primitives. Reach for it when the flow is straightforward
Python; reach for StateGraph (the main Forge build) when you want a declarative graph,
visualization, or complex branching.
"""
from __future__ import annotations
from pathlib import Path

from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.func import entrypoint, task

from forge import config, sandbox
from forge.models import FileEdit
from forge.nodes import EDIT_SYSTEM, MAX_ATTEMPTS, extract_code

The import is from langgraph.func import entrypoint, task — lowercase, from langgraph.func. This is the only correct import in v1.x.

The first @task edits a single file. Notice that config.get_model("smart") is the only way to get a model — no model IDs in this file:

@task
def edit_file(repo_path: str, target_path: str, instruction: str, issue: str) -> FileEdit:
    """A unit of work: edit one file and write it to the sandbox."""
    current = (Path(repo_path) / target_path).read_text()
    model = config.get_model("smart")
    resp = model.invoke([
        SystemMessage(content=EDIT_SYSTEM),
        HumanMessage(content=(
            f"Issue:\n{issue}\n\nEdit file {target_path}: {instruction}\n"
            f"Current contents:\n{current}"
        )),
    ])
    content = extract_code(resp.content)
    sandbox.write_file(repo_path, target_path, content)
    return FileEdit(path=target_path, new_content=content, summary=instruction)

The FileEdit schema is frozen from M5: path, new_content, summary, applied=False. No new fields. The edit is written to the sandbox copy of Tally — never to the original. sandbox.write_file and sandbox.run_tests from M9 handle that constraint.

The second @task runs the test suite:

@task
def run_tests_task(repo_path: str):
    """A unit of work: run the sandboxed suite."""
    return sandbox.run_tests(repo_path)

The return type is TestResult — frozen from M3: passed, total, failures, output_tail. No new fields. This is a deterministic node — a subprocess call to pytest — not a tool call to an LLM.

Step 3: The @entrypoint workflow

The build_forge_app factory compiles the functional workflow. Using a factory (rather than a module-level @entrypoint) makes it easy to inject different checkpointers for tests:

def build_forge_app(checkpointer=None):
    """Compile the functional-API version of Forge's coder loop."""

    @entrypoint(checkpointer=checkpointer)
    def forge_app(payload: dict) -> dict:
        repo = payload["repo_path"]
        issue = payload["issue"]
        targets = payload["targets"]

        # Map: edit every target file (tasks are futures; call .result() to block).
        futures = [edit_file(repo, t["path"], t["instruction"], issue) for t in targets]
        edits = [f.result() for f in futures]

        # Test-fix loop, bounded by MAX_ATTEMPTS from forge/nodes.py.
        result = run_tests_task(repo).result()
        attempts = 1
        while not result.passed and attempts < MAX_ATTEMPTS:
            edit_file(repo, targets[0]["path"], f"fix: {result.output_tail[:80]}", issue).result()
            result = run_tests_task(repo).result()
            attempts += 1

        return {"passed": result.passed, "attempts": attempts, "files": [e.path for e in edits]}

    return forge_app

This is the complete forge/app.py — 67 lines including imports and docstring. Compare that to forge/graph.py, which manages the full pipeline. The Functional API is more concise here because the flow is simpler: no triage, no Planner, no HITL gates, no Reviewer. Just the edit-test-fix-test loop.

The fan-out pattern [edit_file(...) for t in targets] followed by [f.result() for f in futures] is the Functional API equivalent of the Send fan-out from M5. The tasks are dispatched and then their results collected synchronously. For a small number of files (typical for a single-issue fix), this is fine. For large-scale fan-out (N >> 20), the Send API in the Graph API is more efficient.

Step 4: entrypoint.final in practice

The test in test_module_13.py shows the entrypoint.final pattern with a minimal counter example. Here is the pattern applied to Forge’s context — what the caller sees vs. what is saved for the next run:

@entrypoint(checkpointer=checkpointer)
def forge_app(payload: dict, *, previous=None) -> dict:
    prior_attempts = (previous or {}).get("total_attempts", 0)
    repo, issue, targets = payload["repo_path"], payload["issue"], payload["targets"]

    futures = [edit_file(repo, t["path"], t["instruction"], issue) for t in targets]
    edits = [f.result() for f in futures]
    result = run_tests_task(repo).result()
    attempts = 1
    while not result.passed and attempts < MAX_ATTEMPTS:
        edit_file(repo, targets[0]["path"], f"fix: {result.output_tail[:80]}", issue).result()
        result = run_tests_task(repo).result()
        attempts += 1

    return entrypoint.final(
        # What the caller of .invoke() receives:
        value={"passed": result.passed, "files": [e.path for e in edits]},
        # What becomes `previous` on the next invocation on this thread:
        save={"total_attempts": prior_attempts + attempts, "last_issue": issue},
    )

Step 5: The offline tests

The test suite in tests/test_module_13.py uses monkeypatching to run completely offline. The key pattern is replacing config.get_model with a routed fake that returns pre-baked file content for specific target paths:

def test_functional_forge_app_fixes_bug(monkeypatch, make_routed_model):
    tally_src = Path(__file__).resolve().parents[1] / "tally"
    repo = sandbox.make_sandbox(tally_src)
    sandbox.write_file(repo, "tests/test_issue_empty_category.py", REPRO_TEST)
    # Pre-bake the fixed file content that the fake model will return:
    fixed = (Path(repo) / "report.py").read_text().replace("totals[cat]", "totals.get(cat, 0.0)")
    routed = make_routed_model([("report.py", fixed)], default="# noop\n")
    monkeypatch.setattr(config, "get_model", lambda tier="smart", **k: routed)

    app = build_forge_app()  # no checkpointer = in-memory ephemeral
    out = app.invoke({
        "issue": "report crashes on empty category",
        "repo_path": repo,
        "targets": [{"path": "report.py", "instruction": "handle empty category"}],
    })
    assert out["passed"] is True and out["files"] == ["report.py"]

The REPRO_TEST is a real pytest test. sandbox.run_tests(repo) actually runs pytest against the Tally sandbox. The fake model provides the fix, and the real test suite confirms it works. No API key needed. This is the observable signal of the lab: TestResult(passed=True) from the actual pytest run.

The test_entrypoint_previous_accumulates test verifies previous behavior independently — no Tally involved, no monkeypatching needed, because the function itself (a counter) is trivial:

def test_entrypoint_previous_accumulates():
    @entrypoint(checkpointer=checkpoint.in_memory())
    def counter(n: int, *, previous=None) -> int:
        total = (previous or 0) + n
        return entrypoint.final(value=total, save=total)

    cfg = {"configurable": {"thread_id": "c"}}
    assert counter.invoke(1, cfg) == 1
    assert counter.invoke(2, cfg) == 3   # previous=1, +2
    assert counter.invoke(10, cfg) == 13

Step 6: Try it yourself

(a) Switch to a SqliteSaver and observe that the API does not change. Replace build_forge_app() with build_forge_app(checkpointer=checkpoint.sqlite("forge_m13.db")), run a live invocation, kill it mid-run, and resume with the same thread_id. The @entrypoint resumes from the last checkpoint exactly like a compiled graph would.

(b) Add previous to the production version. Extend forge_app to track cumulative attempts across runs on the same thread. Verify that previous["total_attempts"] increments correctly across two calls with the same thread_id.

(c) Add a CachePolicy to run_tests_task. Use key_func=lambda r, **_: r (keyed on repo_path). Time two consecutive calls on the same repo — the second should return the cached TestResult without re-running pytest.

---

In Production

The Functional API earns its place in a production system in specific circumstances. Here is where it actually helps:

Rapid prototyping. When you are exploring a new sub-workflow — a new triage heuristic, a reporting pipeline, a data-cleaning step — the Functional API lets you write 20 lines of Python without thinking about nodes, edges, or state schemas. Iterate quickly, then migrate to the Graph API if the flow grows complex enough to need visualization.

Batch and ETL jobs. The Functional API fits naturally inside a Python loop: asyncio.gather([forge_app.ainvoke(p, config) for p in payloads]). The Graph API can do this too, but the Functional API removes the friction of expressing “just run this function N times in parallel” as a graph topology.

Lightweight sub-workflows. When a sub-workflow has three or four steps in a straight line, a @entrypoint with three @task calls is more readable than a subgraph with nodes, edges, and a build_* function. The Functional API’s @task granularity also gives you per-task retry and cache without a full node declaration.

Granular replay. Each @task is an independent replay unit (CA9.5). If your workflow has ten tasks and crashes on step eight, steps one through seven replay from the checkpoint without re-executing. This is the same guarantee as node-level checkpointing in the Graph API, but expressed at the @task boundary instead of the node boundary.

Limits to keep in mind. The Functional API does not give you per-node visualization in Studio — you see an outline, not a graph. The Send fan-out from M5 (dynamic dispatch to parallel workers via the Send API) is not the natural pattern here — use asyncio.gather for small N, or reach for the Graph API when you need large-scale dynamic fan-out. There is no equivalent of stream_mode="tasks" at per-node granularity. For Forge’s full pipeline — Supervisor, subgraphs, multi-step HITL — the Graph API stays the right choice.

One sentence toward M14: LangGraph Platform supports both APIs — langgraph.json can deploy a build_graph() function or a forge_app entrypoint with the same configuration.

---

Mastery Corner

What to really understand here

CA13.1: @task and @entrypoint come from from langgraph.func import entrypoint, task (lowercase, v1.x only). A @task returns a future (.result() or await). An @entrypoint is the workflow — it exposes .invoke(), .stream(), .get_state(), and astream_events().

CA13.2: previous carries the value returned by the last invocation on the same thread. It is per-thread short-term state, distinct from the Store (cross-thread, long-term) and from the checkpointer’s internal state (which is what makes previous work at all).

CA13.3: entrypoint.final(value=, save=) decouples what the caller receives from what is saved as previous. Use it when the caller needs a summary and the next run needs context.

CA13.4: interrupt() + Command(resume=), durability=, .stream(), and get_stream_writer() work identically in the Functional API. The re-execution trap applies: code in the @entrypoint body before interrupt() re-runs on resume, so it must be idempotent (task results are replayed, not re-executed). interrupt() must be in the @entrypoint body, not inside a @task.

CA13.5 + CA13.6: Choose the Functional API for algorithmic flows in plain Python; choose the Graph API for complex topologies, visualization, team collaboration, and multi-agent systems. Both share the same Pregel runtime, the same checkpointer tables, and the same Store.

---

Five mastery questions

Q1 — Import and runtime (interview)

A developer writes from langgraph.func import Entrypoint, Task (capitalized) in a v1.x project and gets an ImportError. What is the correct import, and what runtime underlies both the Functional API and the Graph API?

• A) Import from langgraph.experimental; the runtime is a custom async event loop.
• B) Import entrypoint, task (lowercase) from langgraph.func; the runtime is the Pregel engine.
• C) Import from langchain.agents; the runtime is LangChain’s agent executor.
• D) No import needed; @entrypoint and @task are built-in Python decorators.

---

Q2 — previous vs the Store (scenario/debugging)

A developer wants to share Forge’s history of fixes between two parallel Forge instances working on two different issues (different thread_id values). They plan to use previous to share this data. Why is that the wrong tool, and what should they use instead?

• A) previous is cross-thread and will work, but it is slow.
• B) The Store, because it is cross-thread and addressable by namespace.
• C) The checkpointer, because it stores all thread state in a shared table.
• D) entrypoint.final(save=...), because the save payload is broadcast to all threads.

---

Q3 — When to use entrypoint.final (scenario)

Forge’s Functional API variant should return {"status": "done", "passed": True} to the caller, but the next invocation on the same thread needs to know the cumulative attempt count and the last file edited. How do you express this?

• A) Put all data in value= — the caller can filter it.
• B) Put all data in save= — the caller receives previous directly.
• C) Return entrypoint.final(value={"status": "done", "passed": True}, save={"total_attempts": n, "last_file": f}).
• D) Use a second, nested @entrypoint to handle the persistence.

---

Q4 — interrupt() placement (debugging)

A developer calls interrupt({"action": "approve_plan", "plan": plan}) inside a @task function. The workflow never pauses — it runs straight through. What is wrong, and where does interrupt() belong?

• A) interrupt() does not work in the Functional API at all; use NodeInterrupt instead.
• B) interrupt() must be called in the @entrypoint body, not inside a @task. A @task is an atomic work unit that cannot pause mid-execution.
• C) The developer should pass interrupt_before=["approve_plan"] to the @entrypoint decorator.
• D) interrupt() only works if durability="sync" is set on the .invoke() call.

---

Q5 — Choosing between APIs (interview)

An architect needs to orchestrate Forge on a pipeline with: a Supervisor routing four sub-agents, LangGraph Studio used by the team for debugging, and five engineers who maintain the system. Which API should they choose and why?

• A) The Functional API — it produces more Python code, which every developer already knows.
• B) The Graph API — the topology is complex, Studio provides visual debugging, and the graph is self-documenting for a team.
• C) Both simultaneously — there is no meaningful distinction for teams.
• D) Neither — use create_react_agent instead of building a custom pipeline.

---

Answers

Q1: B. In v1.x, the correct import is from langgraph.func import entrypoint, task (lowercase). Capitalized Entrypoint and Task existed in pre-1.0 betas and no longer exist. Both APIs run on the Pregel runtime — the same engine, the same checkpointing model.

Q2: B. previous is per-thread: it only flows from run N to run N+1 on the same thread_id. For cross-thread sharing, use the Store (M10) with a namespace that both threads can read and write.

Q3: C. entrypoint.final(value=..., save=...) is the right tool: value is what .invoke() returns to the caller; save is what becomes previous on the next invocation. Options A and B collapse the two roles into one, losing the decoupling. Option D is not how @entrypoint works.

Q4: B. interrupt() pauses the workflow at the @entrypoint level, not inside a @task. A @task is an atomic unit — it commits its result when it finishes. If you call interrupt() inside a @task, it runs as a normal function and the workflow continues. HITL gates belong in the @entrypoint body between task calls.

Q5: B. The Graph API wins for complex multi-agent topologies, Studio-based team debugging, and maintainability by a team. The Functional API is not wrong here but it gives up the visualization and self-documenting topology that a five-person team needs. create_react_agent (option D) is a legacy alias not used in v1.x — use create_agent from langchain.agents for the agent loop.

---

Traps to avoid

@entrypoint without a checkpointer. Without checkpointer= in @entrypoint(checkpointer=...), previous is always None and no state is ever persisted. The Functional API does not auto-persist. This is the same requirement as StateGraph.compile(checkpointer=...). An entrypoint without a checkpointer is a stateless function — perfectly valid for one-shot use, but not for resumable workflows or HITL.

(v0 → v1 freshness trap) langgraph.experimental and capitalized names. Pre-1.0 LangGraph documentation and tutorials (before October 2025) described a Functional API under langgraph.experimental with class names like Entrypoint and Task. That API no longer exists. Any code or documentation that uses from langgraph.experimental import ... or instantiates Entrypoint(...) is describing a removed API. In v1.x: from langgraph.func import entrypoint, task, lowercase, every time.

Believing the Functional API has no streaming or HITL. This is the most common misconception when developers first see the API. Because it looks like plain Python, it can seem like “just a function” without runtime superpowers. It is not. interrupt(), Command(resume=), durability=, .stream(), astream_events(), and get_stream_writer() all work identically. The Pregel runtime underneath provides the same guarantees regardless of which surface you use to express the workflow.

---

Key takeaways

• from langgraph.func import entrypoint, task is the only correct v1.x import. @task returns a future (.result() or await); @entrypoint is the workflow, exposing .invoke(), .stream(), .get_state(), and astream_events().
• previous injects the value returned by the previous invocation on the same thread. It is per-thread, short-term session state — distinct from the Store (cross-thread, long-term) and from the checkpointer’s internal state (which makes previous work in the first place).
• entrypoint.final(value=, save=) decouples what .invoke() returns to the caller from what is persisted as previous for the next run. Use it to serve a lean API surface while saving richer context internally.
• durability=, interrupt(), Command(resume=), .stream(), and astream_events() work identically in both APIs. The Pregel runtime is shared. So are the checkpointer and the Store.
• @task(retry_policy=, cache_policy=) attaches durability and caching at the unit-of-work level. Each task is an independent replay unit: if a workflow resumes from a checkpoint, completed tasks replay their results without re-running.
• Choose the Functional API for algorithmic flows, rapid prototyping, and batch jobs. Choose the Graph API for complex topologies, multi-agent Supervisors, Studio visualization, and team collaboration.
• Both APIs share the same checkpointer and the same Store. You can mix them in a single project — forge/graph.py for the full pipeline, forge/app.py for a specific sub-workflow — without any change to the persistence layer.

---

What’s next

Want the full LangGraph mastery question bank (interview-style) and the next module in your inbox? Subscribe here — it’s free, like everything in this series.

Forge is now complete: Graph API pipeline, Functional API variant, persistence, HITL, streaming, durable execution, long-term memory, and multi-agent supervision all in place. Module 14 ships it — langgraph.json, langgraph dev, LangGraph Platform and Studio, the SDK, and the first real deployment of Forge in production.

---

Links

• Lab code — module-13
• Module 12 — Subgraphs and Multi-Agent Systems
• Module 14 — Deploy Forge: LangGraph Platform, the Server API, and Studio
• Course index — LangGraph in Production

---

References

1. LangGraph Functional API — official guide. langchain-ai.github.io/langgraph/how-tos/use-functional-api (as of June 2026)
2. LangGraph langgraph.func API reference — @entrypoint, @task, entrypoint.final, previous. langchain-ai.github.io/langgraph/reference/func (as of June 2026)
3. How to use the functional API with a checkpointer — LangGraph how-to guide. langchain-ai.github.io/langgraph/how-tos/functional-api-checkpointer (as of June 2026)
4. Malewicz, G. et al. (2010). Pregel: A System for Large-Scale Graph Processing. ACM SIGMOD 2010. The graph-processing model that underlies LangGraph’s superstep-based execution. dl.acm.org/doi/10.1145/1807167.1807184
5. Lab code for this module: github.com/dupuis1212/langgraph-prod-labs/tree/main/module-13