temporalio__skill-temporal-developer/references/python/integrations/langgraph.md

Temporal LangGraph Plugin (Python)

Overview

The LangGraphPlugin runs LangGraph nodes and tasks as Temporal Activities, giving LangGraph-orchestrated AI workflows durable execution, automatic retries, and timeouts.

Both LangGraph APIs are supported:

• Graph API (StateGraph).
• Functional API (@entrypoint / @task).

The plugin ships in the Temporal Python SDK at temporalio.contrib.langgraph.

Note

This feature is in Public Preview. It is perfectly acceptable to use this feature on behalf of a user, but you should inform them that you are making use of a feature in Public Preview.

For general Temporal Python AI/LLM patterns (Pydantic data converter, LLM Activity design, retry classification, multi-agent orchestration), read references/python/ai-patterns.md first; for language-agnostic patterns, read references/core/ai-patterns.md.

Installation

uv add temporalio[langgraph]

This pulls in langgraph as well for you.

Public API

temporalio.contrib.langgraph exports four symbols:

• LangGraphPlugin — the plugin class. Pass to both the Client and the Worker.
• graph(name, cache=...) — Workflow-side accessor that returns a registered StateGraph by name.
• entrypoint(name, cache=...) — Workflow-side accessor that returns a registered Functional-API Pregel.
• cache() — accessor for the per-workflow LangGraph cache.

Build the plugin

Construct the plugin once at startup. The constructor signature is:

LangGraphPlugin(
    graphs: dict[str, StateGraph] | None = None,
    entrypoints: dict[str, Pregel] | None = None,
    tasks: list | None = None,
    activity_options: dict[str, dict[str, Any]] | None = None,
    default_activity_options: dict[str, Any] | None = None,
)

Graph API

Register your StateGraph instances in graphs=, keyed by a name your workflow code will look up with graph(name):

from langgraph.graph import StateGraph
from temporalio.contrib.langgraph import LangGraphPlugin

g = StateGraph(State)
g.add_node("my_node", my_node, metadata={"execute_in": "activity"})

plugin = LangGraphPlugin(graphs={"my-graph": g})

Functional API

Pass entrypoints in entrypoints=, tasks in tasks=, and per-task activity options in activity_options= (keyed by task function name):

from temporalio.contrib.langgraph import LangGraphPlugin

plugin = LangGraphPlugin(
    entrypoints={"my_entrypoint": my_entrypoint},
    tasks=[my_task],
    activity_options={"my_task": {"execute_in": "activity"}},
)

Register the plugin on Client and Worker

Pass the same LangGraphPlugin instance to both the Temporal Client and the Worker via their plugins= parameter, like any other Temporal Python SDK plugin:

from temporalio.client import Client
from temporalio.worker import Worker

client = await Client.connect("localhost:7233", plugins=[plugin])

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    plugins=[plugin],
)
await worker.run()

Execution location — required per node and per task

Every node (Graph API) and every task (Functional API) must declare execute_in, set to either "activity" or "workflow":

# Graph API — set via node metadata
graph.add_node("my_node", my_node, metadata={"execute_in": "activity"})
graph.add_node("tool_node", tool_node, metadata={"execute_in": "workflow"})

# Functional API — set via activity_options on the plugin
plugin = LangGraphPlugin(
    tasks=[my_task, tool_task],
    activity_options={
        "my_task": {"execute_in": "activity"},
        "tool_task": {"execute_in": "workflow"},
    },
)

• Use "activity" for I/O, LLM calls, or anything that needs Temporal retries and timeouts.
• Use "workflow" for logic that should run on the workflow thread: purely deterministic logic, or logic which orchestrates Temporal durable primitives within the workflow thread (e.g. multiple activity calls, wait conditions, etc.)
• execute_in cannot be defaulted in default_activity_options — it must be specified individually per node or task.

Activity options

For nodes or tasks with execute_in: "activity", you can pass parameters that flow through to workflow.execute_activity(): start_to_close_timeout, retry_policy, schedule_to_close_timeout, heartbeat_timeout.

Graph API — options in node metadata

from datetime import timedelta
from temporalio.common import RetryPolicy

g = StateGraph(State)
g.add_node("my_node", my_node, metadata={
    "execute_in": "activity",
    "start_to_close_timeout": timedelta(seconds=30),
    "retry_policy": RetryPolicy(maximum_attempts=3),
})

Functional API — options keyed by task name

from datetime import timedelta
from temporalio.common import RetryPolicy
from temporalio.contrib.langgraph import LangGraphPlugin

plugin = LangGraphPlugin(
    entrypoints={"my_entrypoint": my_entrypoint},
    tasks=[my_task],
    activity_options={
        "my_task": {
            "execute_in": "activity",
            "start_to_close_timeout": timedelta(seconds=30),
            "retry_policy": RetryPolicy(maximum_attempts=3),
        },
    },
)

Run a graph from a Workflow

Inside the Workflow, get the registered graph by name with graph(name), compile it, then ainvoke (or astream) as usual. If the graph requires a checkpointer (for example, when using interrupts), use InMemorySaver — Temporal supplies durability, so third-party checkpointers like PostgreSQL or Redis are not needed:

import typing
import langgraph.checkpoint.memory
from temporalio import workflow
from temporalio.contrib.langgraph import graph

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, input: str) -> typing.Any:
        g = graph("my-graph").compile(
            checkpointer=langgraph.checkpoint.memory.InMemorySaver(),
        )
        return await g.ainvoke({"input": input})

Use entrypoint(name) instead of graph(name) for Functional-API entrypoints.

Runtime context

LangGraph's run-scoped context (context_schema) is reconstructed on the Activity side, so a node can read and write runtime.context even when the node runs as an Activity:

from langgraph.runtime import Runtime
from typing_extensions import TypedDict
from temporalio.contrib.langgraph import graph

class Context(TypedDict):
    user_id: str

async def my_node(state: State, runtime: Runtime[Context]) -> dict:
    return {"user": runtime.context["user_id"]}

# In the Workflow:
g = graph("my-graph").compile()
await g.ainvoke({...}, context=Context(user_id="alice"))

The context object must be serializable by the configured Temporal payload converter, since it crosses the Activity boundary. If your context uses Pydantic models, configure pydantic_data_converter — see references/python/ai-patterns.md.

Tracing

For LangSmith tracing of LangGraph nodes and Temporal Activities together, use the Temporal LangSmith plugin (references/python/integrations/langsmith.md).

Hard constraints

• execute_in is mandatory on every node and task. Set it to "activity" or "workflow" per node/task — it cannot be set in default_activity_options.
• Use InMemorySaver as your checkpointer. Temporal handles durability; do not configure PostgreSQL, Redis, or other third-party checkpointers.
• LangGraph Store is not supported across the Activity boundary. If you pass a Store (e.g. InMemoryStore via graph.compile(store=...) or @entrypoint(store=...)), the plugin logs a warning on first use and runtime.store is None inside nodes. Use Workflow state for per-run memory, or an external database (Postgres/Redis/etc.) configured on each worker for shared memory across runs.
• Context objects must be serializable by the configured Temporal payload converter, since they cross the Activity boundary.

Resources

• references/python/ai-patterns.md — Python AI/LLM patterns (Pydantic data converter, LLM Activity design, retry/error classification).
• references/core/ai-patterns.md — language-agnostic AI/LLM patterns.
• references/python/integrations/langsmith.md - Companion LangSmith plugin.