6. Functional API for Agents

Overview

In capsules 02-05 you built agents as graphs: nodes as capabilities, edges as transitions, conditional edges as decision points. StateGraph gives you visibility, serialization, and extensibility. But there's a cost: to express a simple loop (reason → tools → reason), you need to define nodes, edges, a routing function, and compile. For an agent with basic routing, that's a lot of scaffolding.

LangGraph offers a second API: the Functional API. Instead of modeling the agent as a declarative graph, you write it as normal Python code — while loops for agent loops, if/else for routing, try/except for error handling. Two decorators: @entrypoint marks the agent's main function, @task marks durable sub-operations. The result is an agent that reads like plain Python but has the same guarantees from the LangGraph runtime: durability, checkpointing, and streaming.

This capsule doesn't replace what you learned — it complements it. You'll reimplement patterns from the previous capsules with the Functional API and you'll see that some of them express more naturally. By the end, you'll have clear criteria for choosing between the two APIs.

Two APIs, One Framework

The same runtime, a different paradigm

LangGraph has a unified runtime that handles state, persistence, streaming, and fault tolerance. On top of that runtime, it offers two APIs:

Both APIs produce agents with the same guarantees. If you use a checkpointer, both persist state. If you use streaming, both emit events. The difference is how you express the logic.

Why two APIs

StateGraph is powerful for flows with multiple paths, subgraphs, and complex routing. But for the reason-act loop, the graph adds complexity with no benefit. Writing:

graph.add_edge("tools", "reason")
graph.add_conditional_edges("reason", should_continue, {"tools": "tools", "end": END})

...to express what in Python is:

while True:
    response = reason(messages)
    if not response.tool_calls:
        break
    messages = execute_tools(messages)

The Functional API exists for those cases: when Python's natural flow control is clearer than the graph's declarative structure.

What does NOT change

Regardless of the API: the LangGraph runtime handles execution, you can use checkpointers for persistence (M6), you can stream, tools (@tool) work the same, and models (init_chat_model, bind_tools) work the same. The StateGraph vs Functional API decision is about ergonomics, not capability.

@entrypoint and @task

The two decorators

from langgraph.func import entrypoint, task

@entrypoint() — Marks the agent's main function. It's the entry point, equivalent to the compiled graph.

@task — Marks durable sub-operations. Each @task is a unit of work that LangGraph can checkpoint and re-run if it fails.

Minimal example

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

@task
def call_model(messages: list):
    return model.invoke(messages)

@entrypoint()
def my_agent(messages: list) -> str:
    response = call_model(messages)
    return response.content

result = my_agent.invoke([HumanMessage(content="What is LangGraph?")])

It looks like normal Python, but @entrypoint() adds state handling, streaming, and checkpointing. And if call_model completes and something fails afterward, it won't be re-run on retry — its result is checkpointed thanks to @task.

@task vs a normal function

For prototypes, normal functions are fine. For production with durability, @task is the right way.

The Agent Loop with the Functional API

The reason-act loop as a while

In capsule 04 you built the loop as a cyclic graph: a reason node, a tools node, a return edge, a conditional edge. With the Functional API, that same loop is a while:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, ToolMessage, SystemMessage
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for information about a topic."""
    return f"Results for '{query}': relevant information found."

@tool
def analyze_text(text: str) -> str:
    """Analyze a text and extract its key points."""
    return f"Analysis ({len(text.split())} words): technical concepts identified."

tools = [search_web, analyze_text]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

@task
def reason(messages: list):
    return model_with_tools.invoke(messages)

@task
def execute_tool(tool_call: dict) -> str:
    return str(tools_by_name[tool_call["name"]].invoke(tool_call["args"]))

@entrypoint()
def agent(messages: list) -> str:
    max_iterations = 10
    iteration = 0

    while iteration < max_iterations:
        response = reason(messages)
        messages = messages + [response]
        iteration += 1

        if not response.tool_calls:
            return response.content

        for tc in response.tool_calls:
            result = execute_tool(tc)
            messages = messages + [
                ToolMessage(content=result, tool_call_id=tc["id"])
            ]

    return messages[-1].content

result = agent.invoke([
    SystemMessage(content="You are a research agent. Use tools to answer."),
    HumanMessage(content="What is LangGraph?")
])
print(result)

Side-by-side comparison

The same agent — in StateGraph you need add_node, add_edge, add_conditional_edges, and compile. With the Functional API, it's a while with an if/return. The behavior is identical; for the reason-act loop, the while is more natural and readable.

Stop conditions in the while

In StateGraph, the stop conditions live in should_continue. In the Functional API, they're conditions in the while:

@entrypoint()
def agent_with_budget(messages: list) -> str:
    budget = 1.0
    cost = 0.15

    for iteration in range(10):
        if budget < cost:
            break
        response = reason(messages)
        messages = messages + [response]
        budget -= cost
        if not response.tool_calls:
            return response.content
        for tc in response.tool_calls:
            result = execute_tool(tc)
            messages = messages + [ToolMessage(content=result, tool_call_id=tc["id"])]
    return messages[-1].content

Three stop conditions in plain Python code: max iterations (range(10)), budget (if budget < cost), and task complete (not tool_calls). In StateGraph all of this would go inside should_continue.

Routing with if/else

From conditional_edges to if/else

In capsule 05 the routing uses add_conditional_edges with a mapping of strings to nodes. With the Functional API, it's a direct if/else. There's no string mapping — it's native Python.

Multi-path routing

@task
def classify_query(query: str) -> str:
    return model.invoke([
        SystemMessage(content="Classify as: 'factual', 'analytical', or 'creative'."),
        HumanMessage(content=query)
    ]).content.strip().lower()

@entrypoint()
def routing_agent(query: str) -> str:
    query_type = classify_query(query)

    if "factual" in query_type:
        result = str(search_web.invoke({"query": query}))
    elif "analytical" in query_type:
        search_result = str(search_web.invoke({"query": query}))
        result = f"{search_result}\n\n{str(analyze_text.invoke({'text': search_result}))}"
    else:
        result = model.invoke([HumanMessage(content=query)]).content

    return model.invoke([
        SystemMessage(content="Summarize this information into a clear answer."),
        HumanMessage(content=f"Query: {query}\nResult: {result}")
    ]).content

Three paths, each with different logic, expressed as if/elif/else. No string mapping, no destination dictionaries.

Error Handling with try/except

From error nodes to try/except

In StateGraph, handling errors requires an error node or logic in the tools node with recovery edges. With the Functional API, you use try/except:

@task
def safe_execute_tool(tool_call: dict) -> str:
    if tool_call["name"] not in tools_by_name:
        return f"Error: tool '{tool_call['name']}' doesn't exist."
    try:
        return str(tools_by_name[tool_call["name"]].invoke(tool_call["args"]))
    except Exception as e:
        return f"Error executing {tool_call['name']}: {str(e)}"

@entrypoint()
def resilient_agent(messages: list) -> str:
    error_count = 0

    for iteration in range(10):
        response = reason(messages)
        messages = messages + [response]

        if not response.tool_calls:
            return response.content

        for tc in response.tool_calls:
            result = safe_execute_tool(tc)
            if result.startswith("Error"):
                error_count += 1
            messages = messages + [ToolMessage(content=result, tool_call_id=tc["id"])]

        if error_count >= 3:
            messages = messages + [
                HumanMessage(content="Multiple errors. Answer with what you have.")
            ]
            return reason(messages).content

    return messages[-1].content

Retry with backoff

A production pattern — in StateGraph you'd need a retry node, a counter in the state, and conditional edges. Here it's a for with a try/except:

import time

@task
def execute_with_retry(tool_call: dict, max_retries: int = 3) -> str:
    fn = tools_by_name[tool_call["name"]]
    last_error = None
    for attempt in range(max_retries):
        try:
            return str(fn.invoke(tool_call["args"]))
        except Exception as e:
            last_error = e
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
    return f"Error after {max_retries} attempts: {str(last_error)}"

Reimplementing the Research Agent

The complete agent

The same Research Agent from capsules 02-05 — planning, research in rounds, evaluation, synthesis — now with the Functional API:

from pydantic import BaseModel, Field

class ResearchPlan(BaseModel):
    sub_questions: list[str] = Field(description="Sub-questions to research")
    strategy: str = Field(description="Research strategy")

@task
def plan_research(query: str) -> ResearchPlan:
    return model.with_structured_output(ResearchPlan).invoke([
        SystemMessage(content="Generate 2-3 specific sub-questions and a strategy."),
        HumanMessage(content=f"Main question: {query}")
    ])

@task
def research_question(question: str) -> str:
    result = search_web.invoke({"query": question})
    return f"Q: {question}\nA: {result}"

@task
def evaluate_quality(query: str, results: list[str]) -> float:
    response = model.invoke([
        SystemMessage(content="Rate the quality from 0.0 to 1.0. Answer with ONLY the number."),
        HumanMessage(content=f"Question: {query}\n\nResults:\n" + "\n---\n".join(results))
    ])
    try:
        return float(response.content.strip())
    except ValueError:
        return 0.5

@task
def synthesize(query: str, results: list[str]) -> str:
    return model.invoke([
        SystemMessage(content="Synthesize the information into a complete, clear answer."),
        HumanMessage(content=f"Question: {query}\n\nData:\n" + "\n---\n".join(results))
    ]).content

@entrypoint()
def research_agent(query: str) -> str:
    plan = plan_research(query)
    all_results = []

    for round_num in range(3):
        for question in plan.sub_questions:
            all_results.append(research_question(question))

        quality = evaluate_quality(query, all_results)
        if quality >= 0.8:
            break

    return synthesize(query, all_results)

result = research_agent.invoke("What are the current trends in AI agents?")
print(result)

Comparison with the StateGraph version

The Research Agent with the Functional API reads like a Python script: plan, research in rounds, evaluate quality, synthesize. The entire flow is in one function.

When to Use StateGraph vs the Functional API

Detailed decision table

Practical rules

Use StateGraph when:

1. Your agent has 4+ nodes with complex routing between them
2. You need reusable subgraphs (capsule 07)
3. Graph visualization matters for debugging or communication
4. You're going to compose the agent into a multi-agent system (M8)
5. The flow has multiple independent cycles

Use the Functional API when:

1. The main pattern is a simple reason-act loop
2. Routing is if/else with 2-3 paths
3. You need error handling with try/except and retry
4. You're prototyping and want speed
5. The flow is fundamentally sequential with an inner loop

You can combine both — an @entrypoint can orchestrate internal StateGraphs:

@entrypoint()
def hybrid_agent(query: str) -> str:
    plan = plan_research(query)
    results = []
    for q in plan.sub_questions:
        result = research_state_graph.invoke({"messages": [HumanMessage(content=q)]})
        results.append(result["messages"][-1].content)
    return synthesize(query, results)

The key question

"Is my agent's main value in the structure of the flow or in the logic of the loop?"

• Structure matters (multiple paths, visualization, composition) → StateGraph
• The loop matters (iterate, decide, handle errors) → Functional API

Connection to the Project

In this module's project (capsule 08, Research Agent State Machine), you'll build the Research Agent with StateGraph because it requires visualization, subgraphs (capsule 07), and extensibility for future modules. But now you know you could implement the same logic with the Functional API.

In later modules:

• M5 (Planning): The reflection loop (plan → execute → evaluate → re-plan) is natural with the Functional API
• M6 (Memory): Both APIs support checkpointing. @entrypoint(checkpointer=memory) is equivalent to graph.compile(checkpointer=memory)
• M8 (Multi-Agent): StateGraph is more natural for supervisor patterns, but @entrypoint can orchestrate sub-agents
• M10 (Production): The decision may come down to your team's needs

Troubleshooting

Problem 1: TypeError when decorating with @entrypoint

Cause: @entrypoint needs parentheses even when you pass no arguments.

# Incorrect
@entrypoint
def agent(messages):
    ...

# Correct
@entrypoint()
def agent(messages):
    ...

@entrypoint is a factory. Without (), Python passes the function as an argument to entrypoint, causing a TypeError.

Problem 2: The state doesn't update between iterations

Cause: You're mutating a list instead of creating a new one.

# Problem — in-place mutation
messages.append(response)

# Solution — create a new list
messages = messages + [response]

With checkpointing, tracking is based on @task results. In-place mutations can cause inconsistencies.

Problem 3: The @task doesn't show up in streaming events

Cause: The function doesn't have the @task decorator.

# Invisible to streaming
def call_model(messages):
    return model.invoke(messages)

# Visible to streaming
@task
def call_model(messages):
    return model.invoke(messages)

Problem 4: I can't visualize my Functional API agent

Cause: The Functional API doesn't produce a visualizable graph.

Solution: (1) Explicit logging with print() in each @task, or (2) convert to StateGraph. If your debugging depends on seeing the graph, StateGraph is the better option.

Problem 5: Tools run sequentially, they should be parallel

Cause: A for loop runs the tools one by one.

Solution: Use asyncio.gather with an async @task for parallel execution:

import asyncio

@task
async def execute_tool_async(tool_call: dict) -> str:
    fn = tools_by_name[tool_call["name"]]
    return str(await asyncio.to_thread(fn.invoke, tool_call["args"]))

# Inside the async @entrypoint:
results = await asyncio.gather(*[
    execute_tool_async(tc) for tc in response.tool_calls
])

Exercises

Exercise 1: Basic agent loop with the Functional API (Easy)

Build an agent with @entrypoint and @task that uses search_web to answer questions. A while loop with a maximum of 5 iterations.

@task
def call_model(messages: list):
    return model_with_tools.invoke(messages)

@task
def run_tool(tool_call: dict) -> str:
    return str(tools_by_name[tool_call["name"]].invoke(tool_call["args"]))

@entrypoint()
def simple_agent(messages: list) -> str:
    for _ in range(5):
        response = call_model(messages)
        messages = messages + [response]
        if not response.tool_calls:
            return response.content
        for tc in response.tool_calls:
            result = run_tool(tc)
            messages = messages + [ToolMessage(content=result, tool_call_id=tc["id"])]
    return messages[-1].content

Exercise 2: Routing with if/elif/else (Easy)

Build an agent that classifies the query as "search", "calculation", or "conversation", and runs a different strategy for each type.

@task
def classify(query: str) -> str:
    return model.invoke([
        SystemMessage(content="Classify as 'search', 'calculation', or 'conversation'. ONE word."),
        HumanMessage(content=query)
    ]).content.strip().lower()

@task
def handle_search(query: str) -> str:
    return f"[Search results for: {query}]"

@task
def handle_calc(query: str) -> str:
    return model.invoke([
        SystemMessage(content="Solve this calculation step by step."),
        HumanMessage(content=query)
    ]).content

@entrypoint()
def router_agent(query: str) -> str:
    query_type = classify(query)
    if "search" in query_type:
        result = handle_search(query)
    elif "calculation" in query_type:
        result = handle_calc(query)
    else:
        result = model.invoke([HumanMessage(content=query)]).content
    return model.invoke([
        SystemMessage(content="Present this information in a friendly way."),
        HumanMessage(content=f"Query: {query}\nResult: {result}")
    ]).content

Exercise 3: Error handling with retry (Medium)

Create a @task called resilient_search that tries the search up to 3 times with exponential backoff (1s, 2s, 4s). If it fails, return a descriptive error. Integrate it into an agent.

import time

@task
def resilient_search(tool_call: dict) -> str:
    fn = tools_by_name.get(tool_call["name"])
    if not fn:
        return f"Error: tool '{tool_call['name']}' doesn't exist."
    for attempt in range(3):
        try:
            return str(fn.invoke(tool_call["args"]))
        except Exception as e:
            if attempt < 2:
                time.sleep(2 ** attempt)
            else:
                return f"Error after 3 attempts: {str(e)}"

@entrypoint()
def resilient_agent(messages: list) -> str:
    for _ in range(5):
        response = call_model(messages)
        messages = messages + [response]
        if not response.tool_calls:
            return response.content
        for tc in response.tool_calls:
            result = resilient_search(tc)
            messages = messages + [ToolMessage(content=result, tool_call_id=tc["id"])]
    return messages[-1].content

Exercise 4: Research Agent with quality rounds (Medium)

Implement a Research Agent with the Functional API that: (1) plans 3 sub-questions, (2) researches in rounds, (3) evaluates quality, (4) stops when quality >= 0.7 or after 3 rounds, (5) synthesizes. Every step should be a @task.

class Plan(BaseModel):
    questions: list[str] = Field(description="3 sub-questions")

@task
def plan(query: str) -> list[str]:
    return model.with_structured_output(Plan).invoke([
        SystemMessage(content="Generate exactly 3 sub-questions."),
        HumanMessage(content=query)
    ]).questions

@task
def investigate(question: str) -> str:
    return model.invoke([
        SystemMessage(content="Answer with technical detail."),
        HumanMessage(content=question)
    ]).content

@task
def evaluate(query: str, results: list[str]) -> float:
    r = model.invoke([
        SystemMessage(content="Rate the quality 0.0-1.0. ONLY the number."),
        HumanMessage(content=f"Question: {query}\n\n" + "\n---\n".join(results))
    ])
    try:
        return float(r.content.strip())
    except ValueError:
        return 0.5

@task
def synthesize_all(query: str, results: list[str]) -> str:
    return model.invoke([
        SystemMessage(content="Synthesize this into a complete answer."),
        HumanMessage(content=f"Question: {query}\n\n" + "\n---\n".join(results))
    ]).content

@entrypoint()
def quality_agent(query: str) -> str:
    questions = plan(query)
    all_results = []
    for round_num in range(3):
        for q in questions:
            all_results.append(investigate(q))
        score = evaluate(query, all_results)
        if score >= 0.7:
            break
    return synthesize_all(query, all_results)

Exercise 5: Convert StateGraph to Functional API with loop detection (Hard)

Convert the agent with logging from capsule 04 to the Functional API. Add loop detection: if the model calls the same tool with the same args twice in a row, force termination.

@task
def reason_logged(messages: list, iteration: int):
    print(f"\n{'='*50}")
    print(f"ITERATION {iteration} | Messages: {len(messages)}")
    response = model_with_tools.invoke(messages)
    if response.tool_calls:
        for tc in response.tool_calls:
            print(f"  → {tc['name']}({tc['args']})")
    else:
        print(f"  → Answer directly")
    return response

@task
def execute_logged(tool_call: dict) -> str:
    result = str(tools_by_name[tool_call["name"]].invoke(tool_call["args"]))
    print(f"  [TOOL] {tool_call['name']} → {result[:50]}")
    return result

@entrypoint()
def agent_with_loop_detection(messages: list) -> str:
    last_signature = None

    for iteration in range(1, 6):
        response = reason_logged(messages, iteration)
        messages = messages + [response]

        if not response.tool_calls:
            return response.content

        current_signature = tuple(
            (tc["name"], str(tc["args"])) for tc in response.tool_calls
        )
        if current_signature == last_signature:
            print(f"[STOP] Loop detected — the same tool call repeated")
            messages = messages + [
                HumanMessage(content="You're repeating yourself. Answer with what you have.")
            ]
            return reason_logged(messages, iteration + 1).content

        last_signature = current_signature
        for tc in response.tool_calls:
            result = execute_logged(tc)
            messages = messages + [ToolMessage(content=result, tool_call_id=tc["id"])]

    return messages[-1].content

Summary

In this capsule you learned:

• LangGraph has two APIs: StateGraph (declarative) and the Functional API (imperative). Both use the same runtime with the same guarantees.
• @entrypoint() marks the agent's main function. @task marks durable sub-operations that LangGraph can checkpoint.
• Agent loops with while are more natural than cyclic graphs for the simple reason-act pattern.
• Routing with if/else replaces add_conditional_edges for simple routing — more readable, less boilerplate.
• Error handling with try/except is Python's standard mechanism — retry, circuit breakers, and fallbacks are implemented with familiar patterns.
• StateGraph is better for complex flows, subgraphs, visualization, and multi-agent. The Functional API is better for simple loops, prototyping, and error handling.
• You can combine both: an @entrypoint can orchestrate internal StateGraphs.
• The key question: is the value in the structure of the flow or in the logic of the loop? Structure → StateGraph. Loop → Functional API.

Next capsule: Subgraphs as Agent Modules — encapsulating agent capabilities (research, analysis) as reusable subgraphs with clear interfaces.

Additional Resources

1. LangGraph Functional API — Conceptual Guide — Official documentation for @entrypoint and @task
2. LangGraph Functional API — How-to Guide — Step-by-step tutorial
3. LangGraph Durable Execution — How @task enables durability
4. StateGraph vs Functional API — The official comparison
5. LangGraph Streaming — Streaming events with both APIs