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

Temporal LangSmith Tracing Integration (Python)

Overview

temporalio.contrib.langsmith is a Temporal Plugin for the Python SDK that makes LangSmith traces work across Temporal Workflows and Activities. It propagates trace context across Worker boundaries so @traceable calls, LLM invocations, and Temporal operations show up as a single connected trace, and it suppresses duplicate traces during Workflow replays.

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 Python AI patterns (Pydantic data converter, disabling client-side LLM retries, generic LLM Activity shape) read references/python/ai-patterns.md. For conceptual LLM patterns shared across SDKs read references/core/ai-patterns.md. Python sandbox theory lives in references/python/determinism-protection.md — this integration handles the sandbox restrictions for @traceable for you, so do not restate sandbox rules here.

Install

uv add temporalio[langsmith]

Register the Plugin

Register LangSmithPlugin on both the Client (starter side) and every Worker. Strictly only the sides that produce traces need it, but registering everywhere avoids surprises with context propagation. The Client and Worker can use different configurations (e.g. different add_temporal_runs settings).

from temporalio.client import Client
from temporalio.contrib.langsmith import LangSmithPlugin

client = await Client.connect(
    "localhost:7233",
    plugins=[LangSmithPlugin(project_name="my-project")],
)

from temporalio.worker import Worker
from temporalio.contrib.langsmith import LangSmithPlugin

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

worker = Worker(
    client,
    task_queue="chatbot",
    workflows=[ChatbotWorkflow],
    activities=[call_openai],
)
await worker.run()

LangSmithPlugin parameters

Constructor is keyword-only.

Parameter	Type	Default	Purpose
client	langsmith.Client | None	None (auto-created)	LangSmith client; auto-created if not supplied.
project_name	str | None	None	LangSmith project name traces are written to.
add_temporal_runs	bool	False	When True, adds Temporal operation nodes (StartWorkflow, RunWorkflow, StartActivity, RunActivity) to the trace tree.
default_metadata	dict[str, Any] | None	None	Custom metadata attached to all LangSmith traces.
default_tags	list[str] | None	None	Custom tags attached to all LangSmith traces.

LangSmithInterceptor is also exported alongside LangSmithPlugin; the plugin is the registration entry point and is what user code should use.

Where @traceable works

Location	Works?	Notes
Inside Workflow methods	Yes	Traces called from inside @workflow.run, @workflow.signal, etc.; sync and async methods.
Inside Activity methods	Yes	Traces called from inside @activity.defn; sync and async methods.
On @activity.defn functions	Yes	Stack @traceable on top of @activity.defn (decorator order matters). Fires on every retry.
On Workflow methods	No	Do not wrap @traceable around @workflow.defn, workflow.run, workflow.signal; Use inside @workflow.run instead.

Decorator-order example for an Activity — @traceable on top:

from langsmith import traceable
from temporalio import activity

@traceable(name="Call OpenAI", run_type="llm")
@activity.defn
async def call_openai(...):
    ...

add_temporal_runs — Temporal operation visibility

By default (add_temporal_runs=False), only application @traceable runs appear in LangSmith. With add_temporal_runs=True, Temporal operation nodes are added so the orchestration layer is visible alongside application logic.

plugins=[LangSmithPlugin(project_name="my-project", add_temporal_runs=True)]

With add_temporal_runs=True, StartFoo and RunFoo appear as siblings: the start is the short-lived outbound RPC that enqueues work, the run is the actual execution.

Replay safety — handled by the plugin

The plugin makes @traceable replay-safe in the Workflow sandbox. You do not need to write extra code for this.

• Replay correctness and non-duplication is correctly handled by the plugin, no matter the cause of replay (happy paths, errors, crashes, etc.). Replayed Activities create no new trace data; new work after that produces fresh traces
• The plugin injects metadata using workflow.now() for timestamps and workflow.random() for UUIDs instead of datetime.now() and uuid4().
• LangSmith HTTP calls run on a background thread pool that does not interfere with deterministic Workflow execution.

Context propagation

Trace context flows automatically across Client → Workflow → Activity → Child Workflow → Nexus via Temporal headers. Do not pass context manually.

Worked example — chatbot Workflow with @traceable

A Workflow stays alive waiting for user messages and dispatches each message to an Activity that calls the LLM. @traceable can be used both inside @workflow.run and stacked on top of @activity.defn.

Activity (wraps the LLM call):

from langsmith import traceable
from langsmith.wrappers import wrap_openai
from openai import AsyncOpenAI
from temporalio import activity

@traceable(name="Call OpenAI", run_type="chain")
@activity.defn
async def call_openai(request: OpenAIRequest) -> Response:
    client = wrap_openai(AsyncOpenAI())  # traced LangSmith wrapper
    return await client.responses.create(
        model=request.model,
        input=request.input,
        instructions=request.instructions,
    )

Workflow (orchestrates the conversation; @traceable used inside @workflow.run, not on the class):

from datetime import timedelta
from langsmith import traceable
from temporalio import workflow

@workflow.defn
class ChatbotWorkflow:
    @workflow.run
    async def run(self) -> str:
        # @traceable works inside Workflows — fully replay-safe
        now = workflow.now().strftime("%b %d %H:%M")
        return await traceable(
            name=f"Session {now}", run_type="chain",
        )(self._run_with_trace)()

    async def _run_with_trace(self) -> str:
        while not self._done:
            await workflow.wait_condition(
                lambda: self._pending_message is not None or self._done
            )
            if self._done:
                break

            message = self._pending_message
            self._pending_message = None

            @traceable(name=f"Query: {message[:60]}", run_type="chain")
            async def _query(msg: str) -> str:
                response = await workflow.execute_activity(
                    call_openai,
                    OpenAIRequest(model="gpt-4o-mini", input=msg),
                    start_to_close_timeout=timedelta(seconds=60),
                )
                return response.output_text

            self._last_response = await _query(message)

        return "Session ended."

With add_temporal_runs=False, the trace contains only application logic:

Session Apr 03 14:30
  Query: "What's the weather in NYC?"
    Call OpenAI
      openai.responses.create  (auto-traced by wrap_openai)

With add_temporal_runs=True and the caller wrapping start_workflow in @traceable:

Ask Chatbot                      # @traceable wrapper around client.start_workflow
  StartWorkflow:ChatbotWorkflow
  RunWorkflow:ChatbotWorkflow
    Session Apr 03 14:30
      Query: "What's the weather in NYC?"
        StartActivity:call_openai
        RunActivity:call_openai
          Call OpenAI
            openai.responses.create

Grouping Activity retries under one trace

Because Temporal retries failed Activities and @traceable on @activity.defn fires per attempt, wrap the Activity call in an outer @traceable to group the attempts together:

@traceable(name="Call OpenAI", run_type="llm")
@activity.defn
async def call_openai(...):
    ...

@traceable(name="my_step", run_type="chain")
async def my_step(message: str) -> str:
    return await workflow.execute_activity(
        call_openai,
        ...
    )

Result:

my_step
  Call OpenAI           # first attempt
    openai.responses.create
  Call OpenAI           # retry
    openai.responses.create

Common mistakes

• @traceable on a @workflow.defn class. Not supported — use @traceable inside @workflow.run instead.
• @activity.defn on top of @traceable. Wrong order — @traceable must be the outer decorator on Activities.
• Registering the plugin only on the Client. Register on both Client and every Worker.
• Positional argument to LangSmithPlugin. The constructor is keyword-only — use LangSmithPlugin(project_name="...").
• Combining with temporalio.contrib.opentelemetry and expecting unified traces. They are independent integrations; this reference covers LangSmith only.

Additional Resources

• references/python/integrations/langgraph.md - LangGraph + Temporal plugin - enables running LangGraph agents as durable Temporal workflows.