Update LangGraph samples to current sdk-python main API

The plugin API has evolved past PR #1448:
- LangGraphPlugin(graphs=...) now takes dict[str, StateGraph], not list
- Same for entrypoints (dict[str, Pregel])
- Each node/task must declare execute_in: "activity" | "workflow" in metadata
- Workflows look up registered graphs/entrypoints via graph(name) /
  entrypoint(name) instead of importing them directly
- get_cache/set_cache replaced by cache() and graph(name, cache=...) /
  entrypoint(name, cache=...)

Other changes:
- State schemas converted from primitives (str, int) to TypedDict, since
  langgraph's StateT is bound to TypedDict / dataclass / pydantic models
- Module-level graph instances replaced with make_<name>_graph() factories
  because LangGraphPlugin mutates node metadata in-place at construction
  time, which prevented reuse across tests
- langgraph dep group now pulls langchain, langchain-anthropic, and
  temporalio[langgraph,langsmith] so the langsmith_tracing samples and
  human_in_the_loop's RunnableConfig usage actually resolve

Author: DABH

langgraph_plugin/functional_api/continue_as_new/run_worker.py

@@ -11,12 +11,14 @@
     PipelineFunctionalWorkflow,
     activity_options,
     all_tasks,
+    pipeline_entrypoint,
 )
 
 
 async def main() -> None:
     client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
     plugin = LangGraphPlugin(
+        entrypoints={"pipeline": pipeline_entrypoint},
         tasks=all_tasks,
         activity_options=activity_options,
     )

langgraph_plugin/functional_api/continue_as_new/workflow.py

@@ -10,7 +10,7 @@
 from langgraph.func import entrypoint as lg_entrypoint
 from langgraph.func import task
 from temporalio import workflow
-from temporalio.contrib.langgraph import get_cache, set_cache
+from temporalio.contrib.langgraph import cache, entrypoint
 
 
 @task
@@ -40,10 +40,13 @@ async def pipeline_entrypoint(data: int) -> dict:
     return {"result": loaded}
 
 
-all_tasks = [extract, transform, load]
+all_tasks: list[Any] = [extract, transform, load]
 
 activity_options = {
-    t.func.__name__: {"start_to_close_timeout": timedelta(seconds=30)}
+    t.func.__name__: {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=30),
+    }
     for t in all_tasks
 }
 
@@ -65,14 +68,14 @@ class PipelineFunctionalWorkflow:
 
     @workflow.run
     async def run(self, input_data: PipelineInput) -> dict[str, Any]:
-        set_cache(input_data.cache)
-        result = await pipeline_entrypoint.ainvoke(input_data.data)
+        app = entrypoint("pipeline", cache=input_data.cache)
+        result = await app.ainvoke(input_data.data)
 
         if input_data.phase < 3:
             workflow.continue_as_new(
                 PipelineInput(
                     data=input_data.data,
-                    cache=get_cache(),
+                    cache=cache(),
                     phase=input_data.phase + 1,
                 )
             )

langgraph_plugin/functional_api/control_flow/run_worker.py

@@ -11,12 +11,14 @@
     ControlFlowWorkflow,
     activity_options,
     all_tasks,
+    control_flow_pipeline,
 )
 
 
 async def main() -> None:
     client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
     plugin = LangGraphPlugin(
+        entrypoints={"control-flow": control_flow_pipeline},
         tasks=all_tasks,
         activity_options=activity_options,
     )

langgraph_plugin/functional_api/control_flow/workflow.py

@@ -7,10 +7,12 @@
 """
 
 from datetime import timedelta
+from typing import Any
 
 from langgraph.func import entrypoint as lg_entrypoint
 from langgraph.func import task
 from temporalio import workflow
+from temporalio.contrib.langgraph import entrypoint
 
 
 @task
@@ -74,7 +76,7 @@ async def control_flow_pipeline(items: list[str]) -> dict:
     return {"results": results, "summary": summary_text, "total": len(results)}
 
 
-all_tasks = [
+all_tasks: list[Any] = [
     validate_item,
     classify_item,
     process_urgent,
@@ -83,7 +85,10 @@ async def control_flow_pipeline(items: list[str]) -> dict:
 ]
 
 activity_options = {
-    t.func.__name__: {"start_to_close_timeout": timedelta(seconds=30)}
+    t.func.__name__: {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=30),
+    }
     for t in all_tasks
 }
 
@@ -92,4 +97,4 @@ async def control_flow_pipeline(items: list[str]) -> dict:
 class ControlFlowWorkflow:
     @workflow.run
     async def run(self, items: list[str]) -> dict:
-        return await control_flow_pipeline.ainvoke(items)
+        return await entrypoint("control-flow").ainvoke(items)

langgraph_plugin/functional_api/hello_world/run_worker.py

@@ -11,12 +11,14 @@
     HelloWorldFunctionalWorkflow,
     activity_options,
     all_tasks,
+    hello_entrypoint,
 )
 
 
 async def main() -> None:
     client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
     plugin = LangGraphPlugin(
+        entrypoints={"hello-world": hello_entrypoint},
         tasks=all_tasks,
         activity_options=activity_options,
     )

langgraph_plugin/functional_api/hello_world/workflow.py

@@ -4,10 +4,12 @@
 """
 
 from datetime import timedelta
+from typing import Any
 
 from langgraph.func import entrypoint as lg_entrypoint
 from langgraph.func import task
 from temporalio import workflow
+from temporalio.contrib.langgraph import entrypoint
 
 
 @task
@@ -23,10 +25,13 @@ async def hello_entrypoint(query: str) -> dict:
     return {"result": result}
 
 
-all_tasks = [process_query]
+all_tasks: list[Any] = [process_query]
 
 activity_options = {
-    t.func.__name__: {"start_to_close_timeout": timedelta(seconds=10)}
+    t.func.__name__: {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=10),
+    }
     for t in all_tasks
 }
 
@@ -35,4 +40,4 @@ async def hello_entrypoint(query: str) -> dict:
 class HelloWorldFunctionalWorkflow:
     @workflow.run
     async def run(self, query: str) -> dict:
-        return await hello_entrypoint.ainvoke(query)
+        return await entrypoint("hello-world").ainvoke(query)

langgraph_plugin/functional_api/human_in_the_loop/run_worker.py

@@ -11,12 +11,14 @@
     ChatbotFunctionalWorkflow,
     activity_options,
     all_tasks,
+    chatbot_entrypoint,
 )
 
 
 async def main() -> None:
     client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
     plugin = LangGraphPlugin(
+        entrypoints={"chatbot": chatbot_entrypoint},
         tasks=all_tasks,
         activity_options=activity_options,
     )

langgraph_plugin/functional_api/human_in_the_loop/workflow.py

@@ -12,6 +12,7 @@
 from langgraph.func import task
 from langgraph.types import Command, interrupt
 from temporalio import workflow
+from temporalio.contrib.langgraph import entrypoint
 
 
 @task
@@ -40,10 +41,13 @@ async def chatbot_entrypoint(user_message: str) -> dict:
     return {"response": final_response}
 
 
-all_tasks = [generate_draft, request_human_review]
+all_tasks: list[Any] = [generate_draft, request_human_review]
 
 activity_options = {
-    t.func.__name__: {"start_to_close_timeout": timedelta(seconds=30)}
+    t.func.__name__: {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=30),
+    }
     for t in all_tasks
 }
 
@@ -66,21 +70,22 @@ def get_draft(self) -> str | None:
 
     @workflow.run
     async def run(self, user_message: str) -> dict[str, Any]:
-        chatbot_entrypoint.checkpointer = InMemorySaver()
+        app = entrypoint("chatbot")
+        app.checkpointer = InMemorySaver()
         config = RunnableConfig(
             {"configurable": {"thread_id": workflow.info().workflow_id}}
         )
 
         # First invocation: runs until interrupt() pauses for human review
-        result = await chatbot_entrypoint.ainvoke(user_message, config, version="v2")
+        result = await app.ainvoke(user_message, config, version="v2")
 
         self._draft = result.interrupts[0].value
 
         # Wait for human feedback via Temporal signal
         await workflow.wait_condition(lambda: self._human_input is not None)
 
         # Resume with the human's feedback
-        resumed = await chatbot_entrypoint.ainvoke(
+        resumed = await app.ainvoke(
             Command(resume=self._human_input), config, version="v2"
         )
         return resumed.value

langgraph_plugin/functional_api/langsmith_tracing/run_worker.py

@@ -15,6 +15,7 @@
     ChatFunctionalWorkflow,
     activity_options,
     all_tasks,
+    chat_entrypoint,
 )
 
 
@@ -30,6 +31,7 @@ async def main() -> None:
         workflows=[ChatFunctionalWorkflow],
         plugins=[
             LangGraphPlugin(
+                entrypoints={"chat": chat_entrypoint},
                 tasks=all_tasks,
                 activity_options=activity_options,
             )

langgraph_plugin/functional_api/langsmith_tracing/workflow.py

@@ -8,21 +8,22 @@
 """
 
 from datetime import timedelta
+from typing import Any
 
+from langchain.chat_models import init_chat_model
 from langgraph.func import entrypoint as lg_entrypoint
 from langgraph.func import task
 from langsmith import traceable
 from temporalio import workflow
-
-from langchain.chat_models import init_chat_model
+from temporalio.contrib.langgraph import entrypoint
 
 
 @task
 @traceable(name="chat_task", run_type="chain")
 def chat(message: str) -> str:
     """Call an LLM to respond to the message. Traced by LangSmith."""
     response = init_chat_model("claude-sonnet-4-6").invoke(message)
-    return response.content
+    return str(response.content)
 
 
 @lg_entrypoint()
@@ -32,10 +33,13 @@ async def chat_entrypoint(message: str) -> dict:
     return {"response": response}
 
 
-all_tasks = [chat]
+all_tasks: list[Any] = [chat]
 
 activity_options = {
-    t.func.__name__: {"start_to_close_timeout": timedelta(seconds=30)}
+    t.func.__name__: {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=30),
+    }
     for t in all_tasks
 }
 
@@ -44,4 +48,4 @@ async def chat_entrypoint(message: str) -> dict:
 class ChatFunctionalWorkflow:
     @workflow.run
     async def run(self, message: str) -> dict:
-        return await chat_entrypoint.ainvoke(message)
+        return await entrypoint("chat").ainvoke(message)