LangGraph's Functional API: @entrypoint and @task as an Alternative to Graphs

Add persistence, human-in-the-loop, and streaming to ordinary Python functions — without drawing a StateGraph.

Most LangGraph tutorials teach the graph-building API: you define a shared state schema, add nodes, and wire them together with edges and conditional routing. It is powerful, but for many workflows it feels like a lot of scaffolding around what is really just a sequence of function calls with a loop or two. The Functional API is LangGraph's answer to that. It lets you keep ordinary Python control flow — if statements, for loops, function calls — and still get LangGraph's headline features: durable persistence, human-in-the-loop pauses, memory, and streaming. This article covers @entrypoint and @task, how checkpointing and interrupts work in this model, and when to reach for it instead of a StateGraph.

Two decorators, that's the surface area

The Functional API is just two decorators imported from langgraph.func. @entrypoint marks the top-level function that defines a workflow; @task marks a discrete unit of work inside it — an LLM call, an API request, a bit of processing — that LangGraph should track.

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver
 
@task
def write_essay(topic: str) -> str:
    # pretend this is a slow LLM call
    return f"An essay about {topic}..."
 
@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> str:
    essay = write_essay(topic).result()   # task returns a future
    return essay

Notice that write_essay(topic) does not return a string directly — it returns a future-like object, and you call .result() to resolve it synchronously. That indirection is the whole point: LangGraph records each task's result in the checkpoint, so on a resume it can replay completed tasks from the saved result instead of re-running them.

Invoking, and why tasks matter for determinism

You run an entrypoint like any Runnable — .invoke() or the async .ainvoke() — passing a config with a thread_id so the checkpointer knows which conversation or run this is:

config = {"configurable": {"thread_id": "essay-123"}}
result = workflow.invoke("cats", config)

Anything non-deterministic — random values, timestamps, network calls, LLM requests — must live inside a @task. When a workflow resumes, the entrypoint body re-executes from the top; completed tasks are served from the checkpoint, but bare inline side effects would run again. Wrapping them in tasks is what makes resumption safe.

Short-term memory with previous

An entrypoint can ask for the value it returned on its previous invocation for the same thread_id by declaring a keyword-only previous parameter. This gives you conversational or accumulating state without a state schema:

from typing import Any
 
@entrypoint(checkpointer=InMemorySaver())
def accumulator(number: int, *, previous: Any = None) -> int:
    previous = previous or 0
    return number + previous
 
config = {"configurable": {"thread_id": "acc-1"}}
accumulator.invoke(1, config)   # -> 1
accumulator.invoke(2, config)   # -> 3  (2 + previous 1)

By default the value returned to the caller is also the value saved as previous. When you need those to differ — return one thing to the user but persist another — use entrypoint.final:

@entrypoint(checkpointer=InMemorySaver())
def workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
    previous = previous or 0
    # return `previous` to the caller, but SAVE 2*number as next `previous`
    return entrypoint.final(value=previous, save=2 * number)

Injected parameters: config, store, writer

Beyond previous, the entrypoint can request several framework-provided values as keyword-only arguments, matched by name and type annotation:

from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.types import StreamWriter
 
store = InMemoryStore()
 
@entrypoint(checkpointer=InMemorySaver(), store=store)
def workflow(
    inp: dict,
    *,
    previous: Any = None,
    config: RunnableConfig,
    store: BaseStore,
    writer: StreamWriter,
) -> dict:
    ...
 

Running tasks in parallel

Because a task returns a future immediately, you can launch several before resolving any of them. Call the tasks first, then resolve their futures — LangGraph runs the outstanding tasks concurrently, which is the functional-equivalent of a fan-out/fan-in in a StateGraph, expressed as ordinary Python:

@task
def fetch(source: str) -> str:
    return f"data from {source}"
 
@entrypoint(checkpointer=InMemorySaver())
def gather(sources: list[str]) -> list[str]:
    # launch all tasks first (they don't block here)...
    futures = [fetch(s) for s in sources]
    # ...then resolve them; the tasks ran concurrently
    return [f.result() for f in futures]

This is where the future-based design pays off beyond just checkpointing: the same mechanism that lets LangGraph replay a completed task on resume also lets it schedule independent tasks in parallel without you touching threads or asyncio directly.

Human-in-the-loop with interrupt

Because entrypoints are durable, you can pause one mid-flight to collect human input and resume later — even in a different process or after a restart. Call interrupt() with a payload describing what you need; it raises out of the workflow, the checkpoint is saved, and .invoke() returns the interrupt to the caller. You resume by invoking again with a Command(resume=...):

from langgraph.types import interrupt, Command
 
@entrypoint(checkpointer=InMemorySaver())
def review_flow(topic: str) -> dict:
    essay = write_essay(topic).result()
    # pause here and surface the essay for approval
    decision = interrupt({"essay": essay, "action": "approve or reject?"})
    return {"essay": essay, "approved": decision}
 
config = {"configurable": {"thread_id": "review-1"}}
review_flow.invoke("cats", config)          # returns the interrupt payload
# ... later, after a human decides ...
review_flow.invoke(Command(resume=True), config)   # resumes, returns the dict

On resume, the essay is not regenerated — write_essay ran before the interrupt, so its result is replayed from the checkpoint. That is exactly why the expensive step is a @task and the pause is inline.

Streaming

Entrypoints support the same .stream()/.astream() interface as graphs. Use stream_mode="updates" to see each task's output as it completes, and stream_mode="custom" together with the injected writer to emit your own progress events:

@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str, *, writer: StreamWriter) -> str:
    writer({"status": "starting"})
    essay = write_essay(topic).result()
    writer({"status": "done"})
    return essay
 
for mode, chunk in workflow.stream(
    "cats", config, stream_mode=["updates", "custom"]
):
    print(mode, chunk)

Resuming after an error

If a workflow raises partway through, the checkpoint retains the tasks that already succeeded. After you fix the underlying issue, resume by invoking with None and the same thread_id; completed tasks are replayed from the checkpoint and execution continues from where it failed:

workflow.invoke(None, config)   # replays finished tasks, retries the rest

Functional API vs StateGraph

Both APIs share the same runtime — the same checkpointers, the same interrupt mechanism, the same stores. The choice is about how you want to express control flow.

A useful rule of thumb: if you can describe the workflow as "call this, then maybe loop, then call that," the Functional API will read more naturally. If you find yourself drawing arrows on a whiteboard with several possible next steps from each box, that diagram is a StateGraph.

You don't have to choose globally, either. Because both APIs compile to the same runtime, a common pattern is to use the Functional API for the high-level orchestration — the readable, mostly-sequential outer flow — and drop into a compiled StateGraph for a genuinely branchy sub-problem, invoking that graph from inside a @task. The checkpointer, interrupts, and streaming all continue to work across the boundary, so mixing the two is a supported design choice rather than a hack.

Wrapping up

The Functional API removes the ceremony from LangGraph without giving up what makes it valuable. You write plain functions, wrap the risky or expensive parts in @task, mark the top with @entrypoint, and attach a checkpointer — and in return you get durable execution, resumable human-in-the-loop pauses, cross-thread memory, and streaming. Reach for it when your logic is more sequence than graph, and keep StateGraph for the genuinely branchy, cyclic workflows it was designed for.