@entrypoint: Defining an Agent as a Function

Capsule overview

@entrypoint is the functional equivalent of StateGraph + compile(). Where in the Graph API you defined a StateGraph with nodes, edges and typed state, compiled it, and got an executable graph — with @entrypoint you decorate a Python function and get the same thing: an executable object with invoke, stream, checkpointing, and interrupt support.

The fundamental difference: with StateGraph, the flow lives in the structure of the graph. With @entrypoint, the flow lives in the Python code inside the function. You don't define nodes or edges — you use variables, conditionals, loops, and function calls. LangGraph builds the graph implicitly at runtime.

In this capsule you'll learn to define a workflow with @entrypoint, understand what it does underneath, run it with invoke and stream, and compare it side by side with the Graph API so you can see exactly where the differences are.

Import

from langgraph.func import entrypoint

A single import. entrypoint comes from the langgraph.func module, which is the home of the Functional API.

Basic anatomy of @entrypoint

from langgraph.func import entrypoint

@entrypoint()
def my_agent(query: str) -> str:
    return f"Answer to: {query}"

result = my_agent.invoke("What is LangGraph?")
print(result)
# Expected output: Answer to: What is LangGraph?

Three things happened here:

1. You decorated the function with @entrypoint() — LangGraph wraps your function in a Pregel object (the same runtime that compiled graphs use)
2. The function takes a single positional argument — query: str. If you need to pass multiple pieces of data, use a dictionary
3. You ran it with .invoke() — the same interface as a graph compiled with graph.compile()

The function looks like plain Python. But by decorating it with @entrypoint(), LangGraph gives it:

• ✅ Managed execution (it's not a simple function call)
• ✅ Streaming support with .stream()
• ✅ Checkpointing if you pass a checkpointer
• ✅ Support for interrupt() (human-in-the-loop)

What @entrypoint does underneath

When you decorate a function with @entrypoint(), LangGraph doesn't run it directly. It creates a Pregel object — the same kind of object that graph_builder.compile() produces in the Graph API.

Your code:                         What LangGraph builds:
                                   
@entrypoint()                      ┌──────────────────────────┐
def my_agent(query):  ───────────▶ │  Pregel (runtime object)  │
    ...                            │  - invoke()               │
    return result                  │  - stream()               │
                                   │  - checkpointing          │
                                   │  - interrupt support       │
                                   └──────────────────────────┘

This means my_agent is no longer a regular Python function. It's a Pregel instance with methods like invoke, stream, ainvoke, and astream. You can't call it directly as my_agent("hello") — you need my_agent.invoke("hello").

from langgraph.func import entrypoint

@entrypoint()
def my_agent(query: str) -> str:
    return f"Processing: {query}"

print(type(my_agent))
# Expected output: <class 'langgraph.pregel.Pregel'>

result = my_agent.invoke("test")
print(result)
# Expected output: Processing: test

The input parameter

@entrypoint requires your function to accept a single positional argument as the workflow's input. If you need to pass multiple pieces of data, use a dictionary:

A single value

from langgraph.func import entrypoint

@entrypoint()
def simple_agent(query: str) -> str:
    return f"Answer to: {query}"

result = simple_agent.invoke("What is Python?")
print(result)
# Expected output: Answer to: What is Python?

Multiple values with a dictionary

from langgraph.func import entrypoint

@entrypoint()
def research_agent(inputs: dict) -> str:
    query = inputs["query"]
    max_results = inputs.get("max_results", 5)
    language = inputs.get("language", "en")
    return f"Searching '{query}' (max: {max_results}, language: {language})"

result = research_agent.invoke({
    "query": "prompt engineering",
    "max_results": 10,
    "language": "en"
})
print(result)
# Expected output: Searching 'prompt engineering' (max: 10, language: en)

Inputs and outputs must be JSON-serializable (dict, list, str, int, float, bool, None). You can't pass custom Python objects such as Pydantic models or your own classes directly as input.

The return: what the workflow produces

The value your @entrypoint function returns is the workflow's output. It's what you get back when you call invoke:

from langgraph.func import entrypoint

@entrypoint()
def agent_with_dict_output(query: str) -> dict:
    return {
        "answer": f"Answer to: {query}",
        "confidence": 0.95,
        "sources": ["source_1", "source_2"]
    }

result = agent_with_dict_output.invoke("What is RAG?")
print(result["answer"])
# Expected output: Answer to: What is RAG?
print(result["confidence"])
# Expected output: 0.95
print(result["sources"])
# Expected output: ['source_1', 'source_2']

The return value must be JSON-serializable too. If you need to return complex structured data, use nested dictionaries.

Running an @entrypoint

invoke — full execution

from langgraph.func import entrypoint

@entrypoint()
def my_agent(query: str) -> str:
    return f"Result for: {query}"

result = my_agent.invoke("How does LangGraph work?")
print(result)
# Expected output: Result for: How does LangGraph work?

invoke runs the whole workflow and returns the final result. Just like graph.invoke() in the Graph API.

stream — execution with streaming

from langgraph.func import entrypoint, task

@task
def step_one(query: str) -> str:
    return f"Step 1 completed for: {query}"

@task
def step_two(intermediate: str) -> str:
    return f"Step 2: processing '{intermediate}'"

@entrypoint()
def my_agent(query: str) -> str:
    result_1 = step_one(query).result()
    result_2 = step_two(result_1).result()
    return result_2

for chunk in my_agent.stream("my question"):
    print(chunk)
    print("---")
# Expected output:
# {'step_one': 'Step 1 completed for: my question'}
# ---
# {'step_two': "Step 2: processing 'Step 1 completed for: my question'"}
# ---
# {'my_agent': "Step 2: processing 'Step 1 completed for: my question'"}
# ---

stream emits one chunk per completed @task, plus a final chunk with the @entrypoint's result. Each chunk is a dictionary whose key is the name of the task/entrypoint.

This is useful for giving the user progressive feedback: "Step 1 done... Step 2 in progress... Final result."

Durable execution: a checkpointing preview

When you pass a checkpointer to @entrypoint, LangGraph saves the result of every completed @task. If the workflow crashes halfway through, you can resume it without re-running the tasks that already finished:

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

@task
def expensive_api_call(query: str) -> str:
    """Simulates a costly call to an external API."""
    return f"API result for: {query}"

@task
def process_result(raw_result: str) -> str:
    """Processes the API result."""
    return f"Processed: {raw_result}"

@entrypoint(checkpointer=InMemorySaver())
def durable_agent(query: str) -> str:
    raw = expensive_api_call(query).result()
    processed = process_result(raw).result()
    return processed

config = {"configurable": {"thread_id": "session-001"}}
result = durable_agent.invoke("LangGraph durability", config)
print(result)
# Expected output: Processed: API result for: LangGraph durability

The thread_id identifies the session. If execution is interrupted after expensive_api_call but before process_result, then when you invoke it again with the same thread_id, LangGraph recovers the saved result of expensive_api_call and picks up where it left off.

It doesn't redo work it already completed. That's durable execution — and you'll go deep on it in Module 8.

Injectable parameters

@entrypoint supports parameters that LangGraph injects automatically at runtime. You declare them as keyword-only arguments (after *):

from typing import Any
from langchain_core.runnables import RunnableConfig
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

@entrypoint(checkpointer=InMemorySaver())
def my_agent(
    query: str,
    *,
    previous: Any = None,
    config: RunnableConfig
) -> str:
    history = previous or "No history"
    thread = config["configurable"]["thread_id"]
    return f"[{thread}] {history} → {query}"

config = {"configurable": {"thread_id": "demo-001"}}

result_1 = my_agent.invoke("first question", config)
print(result_1)
# Expected output: [demo-001] No history → first question

result_2 = my_agent.invoke("second question", config)
print(result_2)
# Expected output: [demo-001] [demo-001] No history → first question → second question

previous gives you access to the state of the previous invocation without defining a TypedDict or reducers — it's how the Functional API handles short-term memory.

Side-by-side comparison: @entrypoint vs StateGraph

The same problem — a simple chatbot — solved with both APIs:

Graph API (StateGraph)

from typing import TypedDict, Annotated
import operator
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

def respond(state: ChatState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    last_message = state["messages"][-1]
    response = model.invoke(last_message)
    return {"messages": [response.content]}

graph_builder = StateGraph(ChatState)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": ["What is Python?"]})
print(result["messages"][-1])
# Expected output: Python is a programming language...

Lines of setup code: ~15 (TypedDict, StateGraph, add_node, add_edge × 2, compile)

Functional API (@entrypoint)

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint

@entrypoint()
def chat_agent(message: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(message)
    return response.content

result = chat_agent.invoke("What is Python?")
print(result)
# Expected output: Python is a programming language...

Lines of setup code: ~5 (import, decorator, function, invoke)

Reading the comparison

For a simple one-step chatbot, the Functional API is clearly more concise. But as the workflow grows (multiple nodes, conditional routing, visualization), the Graph API starts to justify itself.

There's no universal answer. There's context.

When @entrypoint alone is enough

For workflows that are a single function with internal logic, @entrypoint without @task works perfectly well:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint

@entrypoint()
def classifier_agent(message: str) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    classification = model.invoke(
        f"Classify this message: technical, creative, or informational. "
        f"Reply with the category ONLY.\n\nMessage: {message}"
    )
    category = classification.content.strip().lower()
    return {"category": category, "original_message": message}

result = classifier_agent.invoke("How do I build a REST API in Python?")
print(f"Category: {result['category']}")
# Expected output: Category: technical

It works without @task because all the logic lives in a single function. There are no independent steps that need individual checkpointing or parallelization.

When you need @task

@entrypoint alone isn't enough when your workflow has multiple steps that benefit from:

• Individual checkpointing: every step is saved, letting you resume from the last successful one
• Per-step streaming: the user sees progress as each task completes
• Parallelization: multiple tasks running at the same time
• Per-step retry: if a step fails, you only retry that step, not the whole workflow

from langgraph.func import entrypoint, task

@task
def fetch_data(source: str) -> dict:
    return {"source": source, "data": f"Data from {source}"}

@task
def analyze_data(data: dict) -> str:
    return f"Analysis of {data['source']}: data processed"

@task
def generate_report(analyses: list[str]) -> str:
    return f"Final report with {len(analyses)} analyses"

@entrypoint()
def research_pipeline(query: str) -> str:
    sources = ["web", "papers", "docs"]
    data_results = [fetch_data(s).result() for s in sources]
    analyses = [analyze_data(d).result() for d in data_results]
    report = generate_report(analyses).result()
    return report

for chunk in research_pipeline.stream("my research"):
    print(chunk)
# Expected output: one chunk per @task executed (3 fetch + 3 analyze + 1 report + 1 entrypoint = 8 chunks)

Every @task emits its own chunk in the stream. If fetch_data("papers") fails and you have checkpointing on, fetch_data("web") isn't re-run when you resume. That granularity isn't possible with @entrypoint alone.

The next capsule goes deep on @task, Futures, and .result().

Troubleshooting

Problem 1: "TypeError: 'Pregel' object is not callable"

Symptom: You try to call the function directly: my_agent("hello"). Cause: @entrypoint turns the function into a Pregel object. It's no longer a regular Python function. Fix:

# Wrong — the function is no longer directly callable
result = my_agent("hello")

# Right — use invoke
result = my_agent.invoke("hello")

Problem 2: "Tasks can only be called from within an entrypoint"

Symptom: An error when calling a @task outside an @entrypoint. Cause: @tasks need an @entrypoint's execution context to work (checkpointing, streaming). Fix:

from langgraph.func import entrypoint, task

@task
def my_task(x: int) -> int:
    return x * 2

# Wrong — @task outside an @entrypoint
# result = my_task(5).result()  # Error

# Right — @task inside an @entrypoint
@entrypoint()
def my_workflow(x: int) -> int:
    return my_task(x).result()

print(my_workflow.invoke(5))
# Expected output: 10

Problem 3: "SerializationError — inputs must be JSON-serializable"

Symptom: An error when passing custom Python objects as input to invoke. Cause: @entrypoint requires inputs and outputs to be JSON-serializable for checkpointing. Fix:

# Wrong — custom object as input
class Query:
    def __init__(self, text, lang):
        self.text = text
        self.lang = lang

# result = my_agent.invoke(Query("hello", "en"))  # Error

# Right — JSON-serializable dictionary
result = my_agent.invoke({"text": "hello", "lang": "en"})

Problem 4: "Streaming doesn't show intermediate steps"

Symptom: stream() only shows the final result, not intermediate steps. Cause: You aren't using @task for the intermediate steps. Without @task, all the code runs inside the @entrypoint as a single block. Fix: Wrap each step you want to see in the stream with @task. Only @tasks emit individual chunks in stream().

Problem 5: "The @entrypoint doesn't remember previous conversations"

Symptom: Every invocation starts from zero, with no memory of previous ones. Cause: You aren't using a checkpointer or the previous parameter. Fix: Pass checkpointer=InMemorySaver() to the decorator, declare previous: Any = None as a keyword-only argument, and use the same thread_id in the config across invocations.

Exercises

Exercise 1: Your first @entrypoint (Easy)

Create an @entrypoint that takes a name (string) and returns a personalized greeting in the format "Welcome to the Research Lab, {name}. Your session has started." Run it with invoke and check the result.

from langgraph.func import entrypoint

@entrypoint()
def welcome_agent(name: str) -> str:
    return f"Welcome to the Research Lab, {name}. Your session has started."

result = welcome_agent.invoke("Ana")
print(result)
# Expected output: Welcome to the Research Lab, Ana. Your session has started.

print(type(welcome_agent))
# Expected output: <class 'langgraph.pregel.Pregel'>

Exercise 2: @entrypoint with a dictionary input (Easy)

Create an @entrypoint that takes a dictionary with the keys topic, depth (str: "shallow" or "deep"), and language (str). The function should return a dictionary with query (formatted as "Research {topic} at {depth} level") and config (with language and a simulated timestamp).

from langgraph.func import entrypoint

@entrypoint()
def configure_research(inputs: dict) -> dict:
    topic = inputs["topic"]
    depth = inputs.get("depth", "shallow")
    language = inputs.get("language", "en")

    return {
        "query": f"Research {topic} at {depth} level",
        "config": {
            "language": language,
            "timestamp": "2026-03-08T10:00:00Z"
        }
    }

result = configure_research.invoke({
    "topic": "prompt engineering",
    "depth": "deep",
    "language": "en"
})
print(result["query"])
# Expected output: Research prompt engineering at deep level
print(result["config"])
# Expected output: {'language': 'en', 'timestamp': '2026-03-08T10:00:00Z'}

Exercise 3: Streaming with @task (Medium)

Create a workflow with an @entrypoint and three @tasks in sequence: parse_query (extracts keywords from the query), search_sources (simulates a search), and summarize (produces a summary). Use stream to watch the execution step by step and count how many chunks the stream produces.

from langgraph.func import entrypoint, task

@task
def parse_query(query: str) -> list[str]:
    words = query.lower().split()
    keywords = [w for w in words if len(w) > 3]
    return keywords

@task
def search_sources(keywords: list[str]) -> list[dict]:
    results = []
    for kw in keywords:
        results.append({"keyword": kw, "source": f"https://example.com/{kw}"})
    return results

@task
def summarize(results: list[dict]) -> str:
    sources_text = ", ".join(r["keyword"] for r in results)
    return f"Summary based on {len(results)} sources: {sources_text}"

@entrypoint()
def research_workflow(query: str) -> str:
    keywords = parse_query(query).result()
    results = search_sources(keywords).result()
    summary = summarize(results).result()
    return summary

chunk_count = 0
for chunk in research_workflow.stream("How does prompt engineering work in production?"):
    chunk_count += 1
    for key, value in chunk.items():
        print(f"Chunk #{chunk_count} [{key}]: {value}")
    print("---")

print(f"\nTotal chunks: {chunk_count}")
# Expected output:
# Chunk #1 [parse_query]: ['does', 'prompt', 'engineering', 'work', 'production?']
# ---
# Chunk #2 [search_sources]: [{'keyword': 'does', 'source': '...'}, ...]
# ---
# Chunk #3 [summarize]: 'Summary based on 5 sources: ...'
# ---
# Chunk #4 [research_workflow]: 'Summary based on 5 sources: ...'
# ---
# Total chunks: 4

Exercise 4: Side-by-side Graph API vs Functional API (Medium)

Implement an ETL (Extract, Transform, Load) pipeline two ways: with StateGraph (Graph API) and with @entrypoint + @task (Functional API). The pipeline takes a number, multiplies it by 2 (extract), adds 10 (transform), and formats it as a string (load). Compare the line count of each implementation.

# ====== GRAPH API ======
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class ETLState(TypedDict):
    messages: Annotated[list[str], operator.add]
    value: int

def extract(state: ETLState) -> dict:
    new_val = state["value"] * 2
    return {"messages": [f"Extract: {state['value']} → {new_val}"], "value": new_val}

def transform(state: ETLState) -> dict:
    new_val = state["value"] + 10
    return {"messages": [f"Transform: {state['value']} → {new_val}"], "value": new_val}

def load(state: ETLState) -> dict:
    return {"messages": [f"Load: result = {state['value']}"]}

graph_builder = StateGraph(ETLState)
graph_builder.add_node("extract", extract)
graph_builder.add_node("transform", transform)
graph_builder.add_node("load", load)
graph_builder.add_edge(START, "extract")
graph_builder.add_edge("extract", "transform")
graph_builder.add_edge("transform", "load")
graph_builder.add_edge("load", END)

graph = graph_builder.compile()
result_graph = graph.invoke({"messages": [], "value": 5})
print(f"Graph API: {result_graph['messages']}")
# Graph API: ['Extract: 5 → 10', 'Transform: 10 → 20', 'Load: result = 20']

# ====== FUNCTIONAL API ======
from langgraph.func import entrypoint, task

@task
def extract_fn(value: int) -> int:
    return value * 2

@task
def transform_fn(value: int) -> int:
    return value + 10

@task
def load_fn(value: int) -> str:
    return f"Result = {value}"

@entrypoint()
def etl_pipeline(value: int) -> str:
    extracted = extract_fn(value).result()
    transformed = transform_fn(extracted).result()
    loaded = load_fn(transformed).result()
    return loaded

result_func = etl_pipeline.invoke(5)
print(f"Functional API: {result_func}")
# Functional API: Result = 20

# Comparison
print(f"\nGraph API: ~25 lines of code")
print(f"Functional API: ~18 lines of code")
print(f"Same result, different expressiveness")

Exercise 5: Short-term memory with previous (Challenge)

Create an @entrypoint with InMemorySaver that keeps a conversation history using the previous parameter. The workflow takes a user message, appends it to the history (which comes from previous), and returns the full history as a list of strings. Check that after 3 invocations on the same thread, the history holds all 3 messages.

from typing import Any
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

@entrypoint(checkpointer=InMemorySaver())
def conversation_agent(message: str, *, previous: Any = None) -> list[str]:
    history = previous if previous is not None else []
    history.append(message)
    return history

config = {"configurable": {"thread_id": "convo-001"}}

result_1 = conversation_agent.invoke("Hi, how are you?", config)
print(f"Turn 1: {result_1}")
# Expected output: Turn 1: ['Hi, how are you?']

result_2 = conversation_agent.invoke("I want to research RAG", config)
print(f"Turn 2: {result_2}")
# Expected output: Turn 2: ['Hi, how are you?', 'I want to research RAG']

result_3 = conversation_agent.invoke("Specifically about chunking strategies", config)
print(f"Turn 3: {result_3}")
# Expected output: Turn 3: ['Hi, how are you?', 'I want to research RAG', 'Specifically about chunking strategies']

assert len(result_3) == 3, f"Expected 3 messages, got {len(result_3)}"
print("\n✅ The history correctly holds all 3 messages")

different_thread = {"configurable": {"thread_id": "convo-002"}}
result_new = conversation_agent.invoke("New thread", different_thread)
print(f"\nNew thread: {result_new}")
# Expected output: New thread: ['New thread']

assert len(result_new) == 1, "A different thread must start empty"
print("✅ Different threads keep independent histories")

Summary

In this capsule you learned:

• @entrypoint is the functional equivalent of StateGraph + compile() — it turns a Python function into an executable Pregel object with invoke and stream
• The decorated function takes a single positional argument (use a dictionary for multiple values). Inputs and outputs must be JSON-serializable
• @entrypoint doesn't run the function directly — it creates a runtime object. You call my_agent.invoke(), not my_agent()
• Streaming works at the @task level: each completed task emits a chunk. Without @task, the stream only shows the final result
• Checkpointing is enabled with checkpointer=InMemorySaver(). The results of each @task are saved, letting you resume from the last successful one
• Injectable parameters (previous, config, store, writer) are declared as keyword-only arguments. previous gives you access to what the previous invocation on the same thread returned
• Compared to StateGraph, @entrypoint needs less setup code but doesn't support visualization. For linear or sequential flows, the Functional API is more concise. For complex topologies with visual branching, the Graph API is clearer

Next capsule: @task: The Units That Make Up the Agent — you'll learn how @task defines independent units of work, what Futures are (like Promises in JavaScript), and when to use @task vs plain code inside the @entrypoint.

Additional resources

1. Functional API Overview — Official docs with a full explanation of @entrypoint and @task
2. @entrypoint Reference — Complete reference for the decorator and all its parameters
3. How to use the Functional API — Practical guide with streaming, memory, and human-in-the-loop examples
4. Choosing between Graph API and Functional API — Official criteria for choosing between the two APIs
5. Introducing the LangGraph Functional API — Launch blog post with the design motivation
6. Persistence in LangGraph — How the checkpointing that powers @entrypoint works
7. InMemorySaver Reference — Reference for the in-memory checkpointer used in the examples
8. Pregel — LangGraph Reference — Reference for the runtime object @entrypoint produces

Module 6 — LangChain & LangGraph: From Chains to Agents