Configure Functional API tasks

A task is a discrete unit of work whose result can be checkpointed on its own. When an entrypoint or a Graph API node runs several operations, wrap each operation in a task instead of putting all the work in one function body. With a checkpointer, completed task results are saved, so a resumed thread can skip finished work and continue from the next task. This page covers retries, timeouts, caching, and error recovery on tasks. For concepts, see the Functional API overview. For an ordered build path, see the Build checklist for the Functional API. For the shared retry and timeout lifecycle, see Fault tolerance.

Add a retry policy

Pass retry_policy on @task. The default RetryPolicy retries selected network and HTTP 5xx errors. It does not retry ValueError. This example retries ValueError.

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

# Used only to simulate a transient failure. Do not use a global like this in production.
attempts = 0

retry_policy = RetryPolicy(retry_on=ValueError)


@task(retry_policy=retry_policy)
def get_info():
    global attempts
    attempts += 1

    if attempts < 2:
        raise ValueError("Failure")
    return "OK"


@entrypoint()
def main(inputs):
    return get_info().result()


print(main.invoke({"any_input": "foobar"}))
# 'OK'

Set a timeout

Cap how long a single async attempt can run. If a retry policy retries the timeout error, the timer resets on the next attempt. Pass timeout on @task or @entrypoint in seconds or as a datetime.timedelta. Timeouts apply only to async tasks and entrypoints. Setting timeout on a sync function raises an error at declaration. The first attempt sleeps past the timeout and raises NodeTimeoutError, which subclasses TimeoutError. The retry policy retries that error. The second attempt returns immediately.

import asyncio

from langgraph.errors import NodeTimeoutError
from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy

attempts = 0


@task(
    timeout=1.0,
    retry_policy=RetryPolicy(retry_on=NodeTimeoutError),
)
async def call_api(url: str) -> str:
    global attempts
    attempts += 1
    if attempts < 2:
        await asyncio.sleep(2)
    return f"result from {url}"


@entrypoint(timeout=5.0)
async def workflow(inputs: dict) -> str:
    return await call_api(inputs["url"])


print(asyncio.run(workflow.ainvoke({"url": "https://example.com"})))
# 'result from https://example.com'

Cache a task

Cache a task result by input. Pass a cache on the entrypoint and a policy on the task. ttl is in seconds. The second call is a cache hit, so this run takes about 1 second, not 2.

import time

from langgraph.cache.memory import InMemoryCache
from langgraph.func import entrypoint, task
from langgraph.types import CachePolicy


@task(cache_policy=CachePolicy(ttl=120))
def slow_add(x: int) -> int:
    time.sleep(1)
    return x * 2


@entrypoint(cache=InMemoryCache())
def main(inputs: dict) -> dict[str, int]:
    result1 = slow_add(inputs["x"]).result()
    result2 = slow_add(inputs["x"]).result()
    return {"result1": result1, "result2": result2}


print(main.invoke({"x": 5}))
# {'result1': 10, 'result2': 10}

Resume after an error

A checkpointer is required so completed task results survive the failed run. slow_task is restored from the checkpoint. get_info runs again and returns "OK". The entrypoint still returns the slow_task result. Resume the same thread_id with None.

import time

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

# Used only to simulate a transient failure. Do not use a global like this in production.
attempts = 0


@task()
def get_info():
    global attempts
    attempts += 1

    if attempts < 2:
        raise ValueError("Failure")
    return "OK"


checkpointer = InMemorySaver()


@task
def slow_task():
    time.sleep(1)
    return "Ran slow task."


@entrypoint(checkpointer=checkpointer)
def main(inputs):
    slow_task_result = slow_task().result()
    get_info().result()  # raises ValueError on the first run
    return slow_task_result


config = {"configurable": {"thread_id": "1"}}

try:
    main.invoke({"any_input": "foobar"}, config=config)
except ValueError:
    pass

print(main.invoke(None, config=config))
# 'Ran slow task.'

Get started

Capabilities

Patterns

Build with the Graph API

Deployment

Frontend

Advanced

Configure Functional API tasks

Add a retry policy

Set a timeout

Cache a task

Resume after an error

See also

​Add a retry policy

​Set a timeout

​Cache a task

​Resume after an error

​See also

Add a retry policy

Set a timeout

Cache a task

Resume after an error

See also