integrations

245 notes in this subject, read out of this brain and free to use. This is page 1 of 5.

Python SDK integrations built on Plugin system

The Temporal Python SDK provides integrations with other tools and services. These integrations are built on the Temporal Python SDK's Plugin system, which developers can also use to build their own integrations.

LangSmithPlugin automatic run naming

When add_temporal_runs=True, the LangSmithPlugin creates runs with names like StartWorkflow:MyWorkflow, RunWorkflow:MyWorkflow, StartActivity:call_openai, and RunActivity:call_openai. Start* and Run* pairs appear as siblings: the Start* run is emitted by the side scheduling the operation (e.g., Client) and the Run* run is emitted by the side executing it (e.g., Worker).

wrap_openai integration with Activities

Use wrap_openai to patch an AsyncOpenAI client so that every API call creates a child run with the model name, prompt, completion, token counts, and latency. Set max_retries=0 on the client and use Temporal's Activity retry policy instead. Example: @traceable(name="Call OpenAI", run_type="llm") @activity.defn async def call_openai(request: OpenAIRequest) -> str: client = wrap_openai(AsyncOpenAI(max_retries=0)) response = await client.responses.create(model=request.model, input=request.input) return response.output_text

@traceable with run_type parameter

The @traceable decorator accepts a run_type parameter that can be set to 'chain', 'llm', 'tool', or 'retriever' to categorize the type of operation being traced.

@traceable on Workflow helper methods

Decorate private helper methods within Workflow classes with @traceable to create named runs for business logic. Do not put @traceable directly on @workflow.run, @workflow.signal, @workflow.update, or @workflow.query methods.

@traceable on Activities

Decorate Activity functions with @traceable to create LangSmith runs that appear nested under the Workflow that scheduled them. Example: @traceable(name="Fetch Weather", run_type="tool") @activity.defn async def fetch_weather(city: str) -> str: ...

LangSmithPlugin trace context propagation

The LangSmithPlugin propagates trace context across Temporal boundaries so that runs started on the Client nest correctly under Workflow and Activity runs on the Worker.

LangSmithPlugin project_name consistency

Use the same project_name on both the Worker and the Client so their traces land in the same LangSmith project.

LangSmithPlugin Worker environment variables

The Worker process must have access to the LANGSMITH_API_KEY environment variable and LANGCHAIN_TRACING_V2 must be set to true for tracing to work.

Avoiding duplicate runs with @traceable on Workflow entry methods

Do not put @traceable directly on @workflow.run, @workflow.signal, @workflow.update, or @workflow.query methods as this can produce duplicate or orphaned runs in LangSmith. Instead, move the logic into an inner function and decorate that inner function.

LangSmithPlugin Client configuration

Add the LangSmithPlugin to a Temporal Client to trace client-side operations like starting a Workflow or sending an Update: client = await Client.connect("localhost:7233", plugins=[LangSmithPlugin(project_name="my-project")]).

LangSmithPlugin add_temporal_runs parameter

By default, LangSmithPlugin(add_temporal_runs=False) only propagates LangSmith context so that @traceable calls nest correctly, without creating its own runs. Set add_temporal_runs=True to create runs for Temporal operations themselves: Workflow executions, Activity executions, Signals, Updates, Queries, and Child Workflows.

LangSmithPlugin Worker configuration

Add the LangSmithPlugin to a Worker by passing it in the plugins parameter with a project_name: worker = Worker(client, task_queue="my-task-queue", workflows=[MyWorkflow], activities=[my_activity], plugins=[LangSmithPlugin(project_name="my-project")]).

Client-side @traceable functions and project_name

Client-side @traceable functions run outside the plugin's interceptor scope, so they don't pick up project_name from the plugin. If you have a client-side @traceable that wraps a call into your Workflow, pass project_name to it explicitly so it lands in the same LangSmith project as the rest of the trace.

LangSmithPlugin installation

Install the Temporal Python SDK with the LangSmith extra using: uv add "temporalio[langsmith]>=1.26.0"

Using traceable() function inside Workflow methods

To trace the body of a @workflow.run, @workflow.signal, @workflow.update, or @workflow.query method, use the traceable() function to wrap an inner function instead of using the @traceable decorator directly on the entry method. Example: return await traceable(name=f"Update: {message[:60]}", run_type="chain")(self._handle_message)(message)

Graph API example with single activity node

Example showing how to define a single-node graph with StateGraph and execute it in a Workflow. The node process_query runs as an Activity with execute_in: "activity" and start_to_close_timeout of 10 seconds. The Workflow calls graph("hello-world").compile().ainvoke(query) to run the compiled graph. ```python from datetime import timedelta from langgraph.graph import START, StateGraph from temporalio import workflow from temporalio.contrib.langgraph import graph async def process_query(query: str) -> str: """Process a query and return a response.""" return f"Processed: {query}" def build_graph() -> StateGraph: """Construct a single-node graph.""" g = StateGraph(str) g.add_node( "process_query", process_query, metadata={ "execute_in": "activity", "start_to_close_timeout": timedelta(seconds=10), }, ) g.add_edge(START, "process_query") return g @workflow.defn class HelloWorldWorkflow: @workflow.run async def run(self, query: str) -> str: return await graph("hello-world").compile().ainvoke(query) ```

Worker configuration with LangGraphPlugin

Example configuring a Worker with LangGraphPlugin. Pass a dict mapping graph names to their StateGraph instances via the graphs parameter. Pass the plugin to Worker via the plugins parameter. ```python import asyncio from temporalio.client import Client from temporalio.contrib.langgraph import LangGraphPlugin from temporalio.worker import Worker async def main() -> None: client = await Client.connect("localhost:7233") plugin = LangGraphPlugin(graphs={"hello-world": build_graph()}) worker = Worker( client, task_queue="langgraph-hello-world", workflows=[HelloWorldWorkflow], plugins=[plugin], ) await worker.run() if __name__ == "__main__": asyncio.run(main()) ```

LangSmith tracing integration

For tracing LangGraph Workflows and Activities, use the Temporal LangSmith plugin. It composes with LangGraphPlugin—pass both plugins to your Worker.

Human-in-the-loop with LangGraph interrupt

LangGraph's interrupt() works with Temporal signals and queries to support human-in-the-loop patterns: (1) A graph node calls interrupt(draft), pausing execution. (2) The Workflow exposes the pending draft via a Temporal query. (3) An external process queries the draft and sends approval via a Temporal signal. (4) The graph resumes—interrupt() returns the signal value and the node completes.

Functional API ReAct agent example

Example of a ReAct agent loop using the Functional API with @task and @lg_entrypoint decorators. The agent_think task decides the next action, execute_tool task runs a tool, and the react_agent entrypoint orchestrates the loop with native Python control flow. ```python from datetime import timedelta from langgraph.func import entrypoint as lg_entrypoint from langgraph.func import task from temporalio import workflow from temporalio.contrib.langgraph import entrypoint @task def agent_think(query: str, history: list[str]) -> dict: """Decide the next action based on query and tool history.""" tool_results = [h for h in history if h.startswith("[Tool]")] if len(tool_results) < 2: return {"action": "tool", "tool_name": "search", "tool_input": query} return {"action": "final", "answer": f"Found: {'; '.join(tool_results)}"} @task def execute_tool(tool_name: str, tool_input: str) -> str: """Execute a tool by name.""" return f"[Tool] Result for {tool_name}({tool_input})" @lg_entrypoint() async def react_agent(query: str) -> dict: """ReAct agent loop: think -> act -> observe -> repeat.""" history: list[str] = [] while True: decision = await agent_think(query, history) if decision["action"] == "final": return {"answer": decision["answer"], "steps": len(history)} result = await execute_tool(decision["tool_name"], decision["tool_input"]) history.append(result) all_tasks = [agent_think, execute_tool] activity_options = { t.func.__name__: { "execute_in": "activity", "start_to_close_timeout": timedelta(seconds=30), } for t in all_tasks } @workflow.defn class ReactAgentWorkflow: @workflow.run async def run(self, query: str) -> dict: return await entrypoint("react-agent").ainvoke(query) ```

LangGraphPlugin default_activity_options

Pass default_activity_options to LangGraphPlugin to apply the same Activity options across every node and task. Per-node metadata (Graph API) and per-task activity_options (Functional API) override these defaults key by key. Cannot set execute_in in default_activity_options; it must be set per node or task individually to prevent determinism bugs.

LangGraph Store not supported with Temporal

LangGraph's Store (for example, InMemoryStore passed via graph.compile(store=...) or @entrypoint(store=...)) is not accessible inside Activity-wrapped nodes. The Store holds live state that cannot cross the Activity boundary, and Activities may run on a different worker than the Workflow. If you pass a 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.) if you need shared memory across runs.

Graph API node metadata for Activity options

Every node in the Graph API must include "execute_in" set to either "activity" or "workflow" in its metadata dict. Pass Activity options as node metadata when calling add_node. Example: g.add_node("my_node", my_node, metadata={"execute_in": "activity", "start_to_close_timeout": timedelta(seconds=30), "retry_policy": RetryPolicy(maximum_attempts=3)}). Do not pass LangGraph's retry_policy parameter to add_node; use Temporal's RetryPolicy instead.

LangGraph streaming retry semantics

Streaming has at-least-once delivery per Activity attempt. When an Activity-wrapped node retries (transient failure, worker crash, etc.), the node function re-runs from scratch and re-publishes its writes—earlier publishes from the failed attempt are not rolled back. Subscribers should dedupe on a sequence ID included in each chunk or treat the stream as advisory and rely on the Workflow's final result for state.

Conditional edge functions must be async in LangGraph

Conditional edge functions like should_continue passed to add_conditional_edges must be async def, not plain def. Synchronous functions cause LangGraph to use run_in_executor, which is not supported inside Temporal's Workflow sandbox.

When to use execute_in activity for LangGraph nodes

Use execute_in: "activity" when a node makes network calls (LLM calls, HTTP requests, database queries), has non-deterministic behavior (random numbers, current time, external data), is long-running or may fail (Activities get configurable timeouts, automatic retries, and heartbeating), or calls interrupt() (LangGraph's interrupt() is supported in Activity nodes).

When to use execute_in workflow for LangGraph nodes

Use execute_in: "workflow" when a node orchestrates other graphs (calls graph("child").compile().ainvoke(state)), performs pure state transformations (deterministic data reshaping, merging, or filtering with no I/O), or is a lightweight routing step (decides what happens next to avoid Activity round-trip overhead). Workflow code must be deterministic and must not make network calls, use random, read the system clock, or do file I/O.

LangGraph integration overview

Temporal's LangGraph integration provides durable execution, automatic retries, and timeouts for LangGraph AI agent workflows. The plugin supports both the Graph API (StateGraph with nodes and edges) and the Functional API (@entrypoint/@task decorators). Each graph node and task must specify whether it runs as a Temporal Activity or directly inside the Workflow.

LangGraph runtime context example

Example showing how to use LangGraph's runtime context in Activity nodes. The my_node function reads from runtime.context which is reconstructed on the Activity side. The context is passed when invoking the graph via ainvoke. ```python 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")) ```

Continue-as-new example with LangGraph cache

Example showing continue-as-new for long-running graphs. The cache() helper returns the current task-result cache as a serializable dict. Pass it to graph(name, cache=...) in the new run to skip re-executing nodes that already produced a result. ```python from temporalio import workflow from temporalio.contrib.langgraph import cache, graph @workflow.defn class LongRunningWorkflow: @workflow.run async def run(self, state: State, prior_cache: dict | None = None) -> State: g = graph("my-graph", cache=prior_cache).compile() # ... run some steps, then continue-as-new before history grows too large ... workflow.continue_as_new(args=[state, cache()]) ```

Checkpointer for LangGraph with Temporal

If LangGraph code requires a checkpointer (for interrupts, for example), use InMemorySaver. Temporal handles durability, so third-party checkpointers like PostgreSQL or Redis are not needed. Pass checkpointer when compiling the graph: g.compile(checkpointer=langgraph.checkpoint.memory.InMemorySaver()).

Streaming example with WorkflowStream

Example showing streaming from a LangGraph node via WorkflowStream. The token_node publishes tokens via get_stream_writer(). The Workflow constructs WorkflowStream() in __init__ and compiles the graph with the app. An external client subscribes to the topic to consume items. ```python from datetime import timedelta from langgraph.config import get_stream_writer from langgraph.graph import START, StateGraph from typing_extensions import TypedDict from temporalio import workflow from temporalio.contrib.langgraph import LangGraphPlugin, graph from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient class State(TypedDict): value: str async def token_node(state: State) -> dict[str, str]: writer = get_stream_writer() for token in ["hello", " ", "world"]: writer({"token": token}) writer({"done": True}) return {"value": "hello world"} @workflow.defn class StreamingWorkflow: def __init__(self) -> None: _ = WorkflowStream() self.app = graph("streaming").compile() @workflow.run async def run(self) -> str: result = await self.app.ainvoke({"value": ""}) return result["value"] g = StateGraph(State) g.add_node("token_node", token_node, metadata={"execute_in": "activity"}) g.add_edge(START, "token_node") plugin = LangGraphPlugin( graphs={"streaming": g}, default_activity_options={"start_to_close_timeout": timedelta(seconds=10)}, streaming_topic="tokens", ) # External client subscription: handle = await client.start_workflow( StreamingWorkflow.run, id="streaming-wf", task_queue="streaming-tq" ) ws_client = WorkflowStreamClient.create(client, handle.id) async for item in ws_client.topic("tokens", type=dict).subscribe(from_offset=0): print(item.data) if item.data.get("done"): break print(await handle.result()) ```

Subgraph orchestration example

Example showing a parent node that runs in the Workflow and dispatches to a child graph whose nodes run as Activities. The parent_node runs with execute_in: "workflow" and calls graph("child").compile().ainvoke(state) to run the subgraph. ```python async def parent_node(state: State) -> dict[str, str]: return await graph("child").compile().ainvoke(state) parent = StateGraph(State) parent.add_node("parent_node", parent_node, metadata={"execute_in": "workflow"}) parent.add_edge(START, "parent_node") plugin = LangGraphPlugin(graphs={"parent": parent, "child": child}) ```

Functional API task and entrypoint decorators

The Functional API uses @entrypoint and @task decorators to orchestrate tasks with native Python control flow (while, if/else, for) rather than declaring nodes and edges. Mark functions with @task, then call them from within an @lg_entrypoint decorated async function. Each task must specify execute_in ("activity" or "workflow") in activity_options passed to LangGraphPlugin.

LangGraph streaming with WorkflowStream

Set streaming_topic on LangGraphPlugin to stream intermediate values from a running graph. Calls to LangGraph's get_stream_writer() inside a node publish to the named topic on the Workflow's WorkflowStream. Activity nodes publish via a batched Temporal signal controlled by streaming_batch_interval (default 100ms). Workflow nodes publish synchronously to the in-Workflow stream with no signal. When streaming_topic is set, the Workflow must construct a WorkflowStream() in its @workflow.init; otherwise the plugin raises an error.

Continue-as-new for long-running LangGraph graphs

Long-running graphs can hit Temporal's per-Event history size limit. Use Temporal's continue-as-new to start a fresh execution while preserving the results of nodes and tasks that have already completed. The cache() helper returns the current task-result cache as a serializable dict. Pass it to graph(name, cache=...) or entrypoint(name, cache=...) in the new run to skip re-executing nodes that already produced a result.

LangGraph runtime context reconstruction

LangGraph's run-scoped context (context_schema) is reconstructed on the Activity side, so nodes and tasks can read from runtime.context. The context object must be serializable by the configured Temporal payload converter since it crosses the Activity boundary.

LangGraph primitives execution location

Node functions run in Activity or Workflow (controlled by execute_in). @task functions run in Activity or Workflow (controlled by execute_in in activity_options). Conditional edge functions (add_conditional_edges) always run in the Workflow and must be deterministic and async. interrupt() runs in Activity. Command(resume=...) runs in Workflow. InMemorySaver checkpointer runs in-Workflow (Temporal handles durability).

LangGraphPlugin Functional API configuration

Configure the Functional API with LangGraphPlugin by passing entrypoints dict (entrypoint name to compiled function), tasks list, and activity_options dict mapping task function names to their options. Example: plugin = LangGraphPlugin(entrypoints={"react-agent": react_agent}, tasks=all_tasks, activity_options=activity_options).

LangGraph streaming_topic coverage and limitations

streaming_topic wires up exactly one LangGraph stream mode: stream_mode="custom" — the values written through get_stream_writer(). The other modes ("messages", "values", "updates", and "debug") are not captured because they're emitted by LangGraph's orchestrator as it walks the graph. To stream those modes, bridge astream() in the Workflow and republish each yielded chunk to a WorkflowStream topic yourself.

Functional API Worker configuration

Example configuring a Worker with the Functional API. Pass entrypoints dict (entrypoint name to compiled function), tasks list, and activity_options dict mapping task function names to their configuration. ```python from temporalio.contrib.langgraph import LangGraphPlugin plugin = LangGraphPlugin( entrypoints={"react-agent": react_agent}, tasks=all_tasks, activity_options=activity_options, ) worker = Worker( client, task_queue="langgraph-react-agent", workflows=[ReactAgentWorkflow], plugins=[plugin], ) ```

Install LangGraph plugin for Python SDK

Install with: uv add "temporalio[langgraph]" or pip install "temporalio[langgraph]". Requires temporalio 1.27.0 or later. Python 3.11 or newer is required for the Functional API (@entrypoint/@task), interrupt(), and streaming from a node running in the Workflow.

Resume agent after interrupt with responses

When agent.invoke_async() returns stop_reason="interrupt", resume by calling agent.invoke_async(responses) with a list of InterruptResponseContent dictionaries. Each dictionary has the form `{"interruptResponse": {"interruptId": i.id, "response": response}}` where interruptId and response correspond to the interrupt that was received.

Interrupt agent from a tool with InterruptException

A @strands.tool function can raise `InterruptException(Interrupt(...))` directly to pause the agent. The agent stops with the interrupt, and the Workflow handles the resume the same way as for hooks. This approach also works from an activity_as_tool-wrapped Activity. The plugin's failure converter preserves the Interrupt payload across the Activity boundary, so AgentResult.interrupts is populated the same way.

Structured output from agent with Pydantic models

Pass a `structured_output_model` to TemporalAgent to have the agent return a typed object instead of free-form text. The plugin defaults to pydantic_data_converter, so Pydantic types serialize cleanly across the Activity and Workflow boundary. Access the structured output from result.structured_output.

Configuring MCP servers with StrandsPlugin

To connect agent to tools from MCP servers, configure the MCP clients on the Worker with `StrandsPlugin(mcp_clients=...)`, which takes a mapping of name to MCPClient factory, mirroring the models pattern. The plugin registers a per-server Activity and connects at Worker startup to enumerate available tools. In the Workflow, use `TemporalMCPClient(server="name")` as a handle that references the server by name and carries per-call Activity options.

activity_as_hook for I/O-safe hook callbacks

Use `activity_as_hook` to dispatch work as a Temporal Activity for callbacks that need I/O (audit logging, metrics, alerting). The `activity_input` parameter extracts serializable values from the event to pass as the Activity's input. Use a dataclass or Pydantic model for multiple values. This is necessary because hook events hold references to Agent, AgentTool instances, and other objects that cannot cross the Activity boundary.

MCP server enumeration happens at Worker startup

The plugin connects to each MCP server once at Worker startup to enumerate tools. The schema is frozen for the Worker's lifetime. Restart Workers to pick up MCP server changes. If a server is unavailable at startup, the Worker fails to start.

Stream agent output to clients with WorkflowStream

For long-running agent calls, pass `streaming_topic="..."` to TemporalAgent and host a WorkflowStream on the Workflow. Each StreamEvent is published from inside the model Activity. Subscribers read events through WorkflowStreamClient. Chunks are batched on `streaming_batch_interval` (default 100 ms).

TemporalAgent configuration with start_to_close_timeout

Create a TemporalAgent instance with `start_to_close_timeout` parameter set to a timedelta value (e.g., `timedelta(seconds=60)`). This sets the maximum time each model call Activity can run.

TemporalAgent must use invoke_async in Workflows

Inside a Workflow, always call `agent.invoke_async(message)`, not `agent(message)`. The synchronous form spawns a worker thread, which the Workflow sandbox blocks.

StrandsPlugin registers model Activities automatically

When creating a Worker with StrandsPlugin, pass the plugin to the Client connection. The plugin automatically registers the Activities that handle model calls, so they do not need to be manually registered.

Strands Agents plugin installation for Python

Install the Temporal Python SDK with Strands Agents support (requires temporalio 1.28.0 or later) using either `uv add "temporalio[strands-agents]"` or `pip install "temporalio[strands-agents]"`.

Strands plugin provides durable execution via Temporal Activities

The Strands Agents integration is an SDK Plugin that gives Strands agents durable execution via the Temporal platform. The plugin routes model invocations, tool calls, MCP tool calls, and hooks through Temporal Activities, so every step the agent takes is recorded in Workflow history and can survive crashes, restarts, and infrastructure failures.

Choosing and configuring Strands agent models

By default, StrandsPlugin uses Strands' own default model (BedrockModel). To use a different model, pass a `models` mapping to StrandsPlugin on the Worker. Each entry in the mapping pairs a name with a factory function that creates a model provider (such as AnthropicModel or BedrockModel). The provider is created on first use and reused for the Worker's lifetime. When providing a custom models mapping, each TemporalAgent must specify which model to use by name with the `model` parameter.

Wrapping tools as Temporal Activities with activity_as_tool

Strands tools that perform I/O, access external services, or produce non-deterministic results must run as Temporal Activities. Wrap each tool in an `@activity.defn` function, register the Activities on the Worker, and pass them to the agent using `activity_as_tool(activity_function, start_to_close_timeout=timedelta(...))`. Register the Activity functions on the Worker in the activities list.

Model not found raises ValueError in Activity

A model name not present in the models mapping raises ValueError inside the Activity at runtime.

Configure retry policy on TemporalAgent

TemporalAgent disables Strands' built-in ModelRetryStrategy so that retries are handled exclusively by Temporal. Configure retries with `retry_policy` parameter on TemporalAgent for model calls using RetryPolicy(maximum_attempts=...). Passing `retry_strategy=...` to TemporalAgent raises ValueError; remove the argument or pass `retry_strategy=None` and use `retry_policy` instead.

Retry policies on Activity-based tools and hooks

Configure retries on the Activity options accepted by `activity_as_tool`, `activity_as_hook`, and `TemporalMCPClient` for their respective calls by passing `retry_policy` with RetryPolicy configuration.

Give your agent this brain