AI-36: Add LangGraph plugin

pyproject.toml

@@ -30,6 +30,7 @@ opentelemetry = ["opentelemetry-api>=1.11.1,<2", "opentelemetry-sdk>=1.11.1,<2"]
 pydantic = ["pydantic>=2.0.0,<3"]
 openai-agents = ["openai-agents>=0.14.0", "mcp>=1.9.4, <2"]
 google-adk = ["google-adk>=1.27.0,<2"]
+langgraph = ["langgraph>=1.1.0"]
 langsmith = ["langsmith>=0.7.0,<0.8"]
 lambda-worker-otel = [
   "opentelemetry-api>=1.11.1,<2",
@@ -79,6 +80,7 @@ dev = [
   "pytest-rerunfailures>=16.1",
   "pytest-xdist>=3.6,<4",
   "moto[s3,server]>=5",
+  "langgraph>=1.1.0",
   "langsmith>=0.7.0,<0.8",
   "setuptools<82",
   "opentelemetry-exporter-otlp-proto-grpc>=1.11.1,<2",

temporalio/contrib/langgraph/README.md

@@ -0,0 +1,201 @@
+# LangGraph Plugin for Temporal Python SDK
+
+⚠️ **This package is currently at an experimental release stage.** ⚠️
+
+This Temporal [Plugin](https://docs.temporal.io/develop/plugins-guide) allows you to run [LangGraph](https://www.langchain.com/langgraph) nodes and tasks as Temporal Activities, giving your AI workflows durable execution, automatic retries, and timeouts. It supports both the LangGraph Graph API (``StateGraph``) and Functional API (``@entrypoint`` / ``@task``).
+
+## Installation
+
+```sh
+uv add temporalio[langgraph]
+```
+
+or with pip:
+
+```sh
+pip install temporalio[langgraph]
+```
+
+## Module layout
+
+Define your graphs, tasks, and entrypoints in a module **separate** from your `@workflow.defn` classes — the standard Temporal split. The plugin adds the graph/task modules to the workflow sandbox's passthrough list so its in-place rewrites are visible to the workflow. Workflow modules stay sandboxed.
+
+## Graph API
+
+```python
+# graphs.py
+from langgraph.graph import START, StateGraph
+
+my_graph = StateGraph(State)
+my_graph.add_node("my_node", my_node)
+my_graph.add_edge(START, "my_node")
+
+# workflow.py
+from temporalio import workflow
+from myapp.graphs import my_graph
+
+@workflow.defn
+class MyWorkflow:
+    @workflow.run
+    async def run(self, input):
+        return await my_graph.compile().ainvoke(input)
+
+# worker.py
+from temporalio.contrib.langgraph import LangGraphPlugin
+from myapp.graphs import my_graph
+
+plugin = LangGraphPlugin(graphs=[my_graph])
+```
+
+## Functional API
+
+```python
+# flows.py
+from langgraph.func import entrypoint, task
+
+@task
+async def my_task(x): ...
+
+@entrypoint()
+async def my_flow(inputs):
+    return await my_task(inputs)
+
+# workflow.py
+from temporalio import workflow
+from myapp.flows import my_flow
+
+@workflow.defn
+class MyWorkflow:
+    @workflow.run
+    async def run(self, input):
+        return await my_flow.ainvoke(input)
+
+# worker.py
+import datetime
+from temporalio.contrib.langgraph import LangGraphPlugin
+from myapp.flows import my_task
+
+plugin = LangGraphPlugin(
+    tasks=[my_task],
+    activity_options={
+        "my_task": {
+            "start_to_close_timeout": datetime.timedelta(seconds=30),
+        },
+    },
+)
+```
+
+## Checkpointer
+
+Use `InMemorySaver` as your checkpointer. Temporal handles durability, so third-party checkpointers (like PostgreSQL or Redis) are not needed.
+
+```python
+import langgraph.checkpoint.memory
+
+from temporalio import workflow
+from myapp.graphs import my_graph
+
+@workflow.defn
+class MyWorkflow:
+    @workflow.run
+    async def run(self, input):
+        app = my_graph.compile(
+            checkpointer=langgraph.checkpoint.memory.InMemorySaver(),
+        )
+        ...
+```
+
+## Activity Options
+
+Options are passed through to [`workflow.execute_activity()`](https://python.temporal.io/temporalio.workflow.html#execute_activity), which supports parameters like `start_to_close_timeout`, `retry_policy`, `schedule_to_close_timeout`, `heartbeat_timeout`, and more.
+
+### Graph API
+
+Pass per-node options as node `metadata`, or plugin-wide defaults via `default_activity_options`:
+
+```python
+import datetime
+from temporalio.common import RetryPolicy
+
+g = StateGraph(State)
+g.add_node("my_node", my_node, metadata={
+    "start_to_close_timeout": datetime.timedelta(seconds=30),
+    "retry_policy": RetryPolicy(maximum_attempts=3),
+})
+
+plugin = LangGraphPlugin(
+    graphs=[g],
+    default_activity_options={"start_to_close_timeout": datetime.timedelta(seconds=60)},
+)
+```
+
+### Functional API
+
+Pass activity options to the plugin, keyed by task function name:
+
+```python
+import datetime
+from temporalio.common import RetryPolicy
+from temporalio.contrib.langgraph import LangGraphPlugin
+
+plugin = LangGraphPlugin(
+    tasks=[my_task],
+    activity_options={
+        "my_task": {
+            "start_to_close_timeout": datetime.timedelta(seconds=30),
+            "retry_policy": RetryPolicy(maximum_attempts=3),
+        },
+    },
+)
+```
+
+### Running in the Workflow
+
+To skip the Activity wrapper and run a node or task directly in the Workflow, set `execute_in` to `"workflow"`:
+
+```python
+# Graph API
+g.add_node("my_node", my_node, metadata={"execute_in": "workflow"})
+
+# Functional API
+plugin = LangGraphPlugin(
+    tasks=[my_task],
+    activity_options={"my_task": {"execute_in": "workflow"}},
+)
+```
+
+## Continue-As-New
+
+To carry cached task results across a continue-as-new boundary, pass the cache to your next run and restore it with `set_cache`:
+
+```python
+from temporalio import workflow
+from temporalio.contrib.langgraph import get_cache, set_cache
+from myapp.graphs import my_graph
+
+@workflow.defn
+class MyWorkflow:
+    @workflow.run
+    async def run(self, input, prev_cache=None):
+        set_cache(prev_cache)
+        result = await my_graph.compile().ainvoke(input)
+        if should_continue(result):
+            workflow.continue_as_new(next_input, get_cache())
+        return result
+```
+
+## Running Tests
+
+Install dependencies:
+
+```sh
+uv sync
+```
+
+Run the test suite:
+
+```sh
+uv run pytest
+```
+
+Tests start a local Temporal dev server automatically — no external server needed.

temporalio/contrib/langgraph/__init__.py

@@ -0,0 +1,23 @@
+"""LangGraph plugin for Temporal SDK.
+
+.. warning::
+    This package is experimental and may change in future versions.
+    Use with caution in production environments.
+
+This plugin runs `LangGraph <https://github.com/langchain-ai/langgraph>`_ nodes
+and tasks as Temporal Activities, giving your AI agent workflows durable
+execution, automatic retries, and timeouts. It supports both the LangGraph Graph
+API (``StateGraph``) and Functional API (``@entrypoint`` / ``@task``).
+"""
+
+from temporalio.contrib.langgraph.langgraph_plugin import (
+    LangGraphPlugin,
+    get_cache,
+    set_cache,
+)
+
+__all__ = [
+    "LangGraphPlugin",
+    "get_cache",
+    "set_cache",
+]

temporalio/contrib/langgraph/activity.py

@@ -0,0 +1,99 @@
+"""Activity wrappers for executing LangGraph nodes and tasks."""
+
+from collections.abc import Awaitable
+from dataclasses import dataclass
+from inspect import iscoroutinefunction
+from typing import Any, Callable
+
+from langgraph.errors import GraphInterrupt
+from langgraph.types import Interrupt
+
+from temporalio import workflow
+from temporalio.contrib.langgraph.langgraph_config import (
+    get_langgraph_config,
+    set_langgraph_config,
+)
+from temporalio.contrib.langgraph.task_cache import (
+    cache_key,
+    cache_lookup,
+    cache_put,
+)
+
+
+@dataclass
+class ActivityInput:
+    """Input for a LangGraph activity, containing args, kwargs, and config."""
+
+    args: tuple[Any, ...]
+    kwargs: dict[str, Any]
+    langgraph_config: dict[str, Any]
+
+
+@dataclass
+class ActivityOutput:
+    """Output from a LangGraph activity, containing result or interrupts."""
+
+    result: Any = None
+    langgraph_interrupts: tuple[Interrupt] | None = None
+
+
+def wrap_activity(
+    func: Callable,
+) -> Callable[[ActivityInput], Awaitable[ActivityOutput]]:
+    """Wrap a function as a Temporal activity that handles LangGraph config and interrupts."""
+
+    async def wrapper(input: ActivityInput) -> ActivityOutput:
+        set_langgraph_config(input.langgraph_config)
+        try:
+            if iscoroutinefunction(func):
+                result = await func(*input.args, **input.kwargs)
+            else:
+                result = func(*input.args, **input.kwargs)
+            return ActivityOutput(result=result)
+        except GraphInterrupt as e:
+            return ActivityOutput(langgraph_interrupts=e.args[0])
+
+    return wrapper
+
+
+def wrap_execute_activity(
+    afunc: Callable[[ActivityInput], Awaitable[ActivityOutput]],
+    task_id: str = "",
+    **execute_activity_kwargs: Any,
+) -> Callable[..., Any]:
+    """Wrap an activity function to be called via workflow.execute_activity with caching."""
+
+    async def wrapper(*args: Any, **kwargs: Any) -> Any:
+        # LangGraph may inject a RunnableConfig as the 'config' kwarg. Strip it
+        # down to a serializable subset (metadata + tags) so it can cross the
+        # activity boundary; callbacks, stores, etc. aren't serializable.
+        if "config" in kwargs:
+            orig = kwargs["config"] or {}
+            kwargs["config"] = {
+                "metadata": dict(orig.get("metadata") or {}),
+                "tags": list(orig.get("tags") or []),
+            }
+
+        # Check task result cache (for continue-as-new deduplication).
+        key = cache_key(task_id, args, kwargs) if task_id else ""
+        if task_id:
+            found, cached = cache_lookup(key)
+            if found:
+                return cached
+
+        input = ActivityInput(
+            args=args, kwargs=kwargs, langgraph_config=get_langgraph_config()
+        )
+        output = await workflow.execute_activity(
+            afunc, input, **execute_activity_kwargs
+        )
+        if output.langgraph_interrupts is not None:
+            raise GraphInterrupt(output.langgraph_interrupts)
+
+        # Store in cache for future continue-as-new cycles.
+        if task_id:
+            cache_put(key, output.result)
+
+        return output.result
+
+    return wrapper