integrations

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

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.

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).

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.

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.

Model not found raises ValueError in Activity

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

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.

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.

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.

Example: Worker with StrandsPlugin registration

The following example shows how to create a Worker that registers the Workflow and StrandsPlugin: ```python import asyncio import os from temporalio.client import Client from temporalio.contrib.strands import StrandsPlugin from temporalio.worker import Worker from strands_plugin.hello_world.workflow import HelloWorldWorkflow async def main() -> None: plugin = StrandsPlugin() client = await Client.connect( os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"), plugins=[plugin], ) worker = Worker( client, task_queue="strands-hello-world", workflows=[HelloWorldWorkflow], ) print("Worker started. Ctrl+C to exit.") await worker.run() if __name__ == "__main__": asyncio.run(main()) ``` The plugin is passed to the Client connection and automatically registers Activities for model calls.

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.

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.

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.

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]"`.

Temporal TypeScript SDK integrations available

The Temporal TypeScript SDK provides integrations with other tools and services. An integrations grid displays the available integrations for the TypeScript SDK.

LangSmith API key handling

The LangSmith API key is never accepted as a plugin option and never crosses a Temporal boundary. Supply a pre-constructed Client (which reads LANGSMITH_API_KEY from the process environment) or let the plugin build a default client from the environment.

Wrap client Workflow calls in traceable

Use langsmith/traceable to wrap client-side Workflow execution so the trace nests under your own run. Example: const pipeline = traceable(async () => { return client.workflow.execute(GreetingWorkflow, {...}); }, { name: 'user_pipeline' }); await pipeline();

Plugin ordering for composition

Register observability plugins first (outermost) so they observe everything beneath them, then governance, then agent-framework plugins. The LangSmithPlugin de-duplicates its own instrumentation, so a Worker built from a plugin-configured Client will not double-instrument.

addTemporalRuns false vs true trace structure

With addTemporalRuns: true, traces include first-class runs for Temporal operations as Start*/Run* pairs (e.g., StartWorkflow:, RunWorkflow:, StartActivity:, RunActivity:, HandleSignal:, HandleUpdate:). With addTemporalRuns: false (default), only traceable runs appear, but context still propagates so they nest correctly under client-side runs.

Configure Client with LangSmithPlugin

In TypeScript, add a LangSmithPlugin to your Client independently from the Worker. Create a Client instance with the plugin passed in the plugins array: new Client({ connection, plugins: [plugin] }). This links client-side operations like starting a Workflow to the Workflows they trigger.

Install LangSmith integration packages

To use LangSmith with Temporal TypeScript SDK, run: npm install @temporalio/langsmith langsmith

LangSmithPlugin options table

LangSmithPlugin configuration options: | Option | Default | Meaning | | --- | --- | --- | | client | new Client() | The LangSmith Client that runs are emitted to | | addTemporalRuns | false | Emit first-class runs for Temporal operations (StartWorkflow:, RunActivity:, HandleSignal:, …) in addition to traceable runs | | projectName | LangSmith default | Target LangSmith project for emitted runs | | tags | — | Tags attached to every run the plugin emits | | metadata | — | Metadata merged into every run the plugin emits. Credential-looking keys are scrubbed before emission |

Configure Worker with LangSmithPlugin

Import NativeConnection, Worker from '@temporalio/worker', Client from 'langsmith', and LangSmithPlugin from '@temporalio/langsmith'. Create a LangSmith Client instance (which reads LANGSMITH_API_KEY from environment), create a LangSmithPlugin instance passing the client and addTemporalRuns option, then pass the plugin in the plugins array when creating the Worker.

LangSmith integration overview

The LangSmithPlugin connects Temporal with LangSmith to trace AI agent Workflows. It propagates trace context across Temporal boundaries (Workflow → Activity → Child Workflow) so that runs started on the Client nest correctly under Workflow and Activity runs on the Worker. It can emit LangSmith runs for Temporal operations: Workflow executions, Activity executions, Signals, and Updates.

Trace Signal and Update handlers

A traceable inside a Signal or Update handler nests under that handler's run, following the same Workflow-body semantics. Temporal-internal Queries (__temporal*, __stack_trace) are never traced.

Traceable nesting in Workflow bodies

Inside a Workflow body, sequential await inner(...) nesting is exact. Under Promise.all(...) fan-out, or for traceable calls made after an await in the same scope, parenting falls back to the Workflow run. This affects only the visual shape of the trace, never Workflow history or control flow. Activity-side and client-side traceable are unaffected.

LangSmith environment variables

Set LANGSMITH_TRACING=true (or LANGSMITH_TRACING_V2) and LANGSMITH_API_KEY with your API key to enable tracing. The plugin reads the same flags that the langsmith library uses, including LANGCHAIN_ aliases. Tracing is off by default.

Worker configuration with LangSmithPlugin example

import { NativeConnection, Worker } from '@temporalio/worker'; import { Client as LangSmithClient } from 'langsmith'; import { LangSmithPlugin } from '@temporalio/langsmith'; import * as activities from './activities'; const connection = await NativeConnection.connect({ address: 'localhost:7233' }); const langsmith = new LangSmithClient(); const plugin = new LangSmithPlugin({ client: langsmith, addTemporalRuns: true }); const worker = await Worker.create({ connection, taskQueue: 'langsmith', workflowsPath: require.resolve('./workflows'), activities, plugins: [plugin], }); await worker.run();

Trace multi-step agent example

import { executeChild, proxyActivities, workflowInfo } from '@temporalio/workflow'; import type * as activities from './activities'; const { gatherFacts, writeReport, reviewReport } = proxyActivities<typeof activities>({ startToCloseTimeout: '1 minute', }); export async function ReviewWorkflow(report: string): Promise<string> { return reviewReport(report); } export async function ResearchWorkflow(topic: string): Promise<string> { const facts = await gatherFacts(topic); const report = await writeReport(facts); return executeChild(ReviewWorkflow, { args: [report], workflowId: `${workflowInfo().workflowId}-review`, }); }

Trace Workflow bodies with traceable

A traceable works inside a Workflow body and the plugin keeps it replay-safe: each run is emitted exactly once and is never duplicated when the Workflow replays its history.

Trace Signal and Update handlers example

import { traceable } from 'langsmith/traceable'; import { allHandlersFinished, condition, defineSignal, defineUpdate, setHandler } from '@temporalio/workflow'; const classifyMessage = traceable(async (text: string): Promise<string> => `intent:${text}`, { name: 'classify_intent', }); const draftReply = traceable(async (text: string): Promise<string> => `reply:${text}`, { name: 'draft_reply', }); export const handleMessage = defineSignal<[string]>('handle_message'); export const composeReply = defineUpdate<string, [string]>('compose_reply'); export const complete = defineSignal('complete'); export async function ConversationWorkflow(): Promise<string[]> { const log: string[] = []; let done = false; setHandler(handleMessage, async (text: string) => { log.push(await classifyMessage(text)); }); setHandler(composeReply, async (text: string) => { const reply = await draftReply(text); log.push(reply); return reply; }); setHandler(complete, () => { done = true; }); await condition(() => done && allHandlersFinished()); return log; }

Trace Workflow with LangSmith example

import { traceable } from 'langsmith/traceable'; const extractKeyPoints = traceable(async (text: string): Promise<string> => `points:${text}`, { name: 'extract_key_points', }); const summarize = traceable(async (points: string): Promise<string> => `summary:${points}`, { name: 'summarize', }); export async function SummarizeWorkflow(text: string): Promise<string> { const points = await extractKeyPoints(text); return summarize(points); }

Trace Activity with LangSmith example

import { traceable } from 'langsmith/traceable'; const callModel = traceable( async (prompt: string): Promise<string> => { return `answer to: ${prompt}`; }, { name: 'inner_llm_call' }, ); export async function answer(prompt: string): Promise<string> { return callModel(prompt); }

Trace Activities with traceable

A traceable from langsmith/traceable works unchanged inside an Activity body. The run shows up in LangSmith nested under RunActivity: for the Activity that scheduled it. Non-deterministic work in Workflows (LLM calls, tool executions, database queries, external API calls) must run inside Activities, making them an important place to add LangSmith runs.

Use TemporalMCPClient in Workflow

In your agent Workflow, use TemporalMCPClient to get tools from an MCP server by referencing it by name: ```ts import { TemporalMCPClient, temporalProvider } from '@temporalio/ai-sdk/workflow'; export async function mcpAgent(prompt: string): Promise<string> { const mcpClient = new TemporalMCPClient({ name: 'testServer' }); const tools = await mcpClient.tools(); const result = await generateText({ model: temporalProvider.languageModel('gpt-4o-mini'), prompt, tools, system: 'You are a helpful agent, You always use your tools when needed.', stopWhen: stepCountIs(5), }); return result.text; } ``` Both listing tools and calling them run as Activities behind the scenes, providing automatic retries, timeouts, and full observability.

Configure MCP client factories in Worker

Create a connection to MCP servers using the experimental_createMCPClient function from @ai-sdk/mcp package. Register multiple MCP servers by providing multiple factory functions in mcpClientFactories. Configure the Worker with mcpClientFactories in the AiSdkPlugin options: ```ts const mcpClientFactories = { testServer: () => createMCPClient({ transport: new StdioClientTransport({ command: 'node', args: ['lib/mcp-server.js'], }), }), }; const worker = await Worker.create({ plugins: [ new AiSdkPlugin({ modelProvider: openai, mcpClientFactories }), ], ... }); ```

MCP (Model Context Protocol) server integration

Model Context Protocol (MCP) is an open standard that lets AI applications connect to external tools and data sources. Calls to MCP servers are non-deterministic and would usually need to be implemented as Activities. The Temporal AI SDK integration handles this automatically and provides a built-in TemporalMCPClient for use inside Workflows.

Workflow tools example with Activity

Example of providing a weather tool to a Workflow agent: ```ts import { proxyActivities } from '@temporalio/workflow'; import { generateText, tool } from 'ai'; import { temporalProvider } from '@temporalio/ai-sdk/workflow'; import { z } from 'zod'; const { getWeather } = proxyActivities<typeof activities>({ startToCloseTimeout: '1 minute', }); export async function toolsAgent(question: string): Promise<string> { const result = await generateText({ model: temporalProvider.languageModel('gpt-4o-mini'), prompt: question, system: 'You are a helpful agent.', tools: { getWeather: tool({ description: 'Get the weather for a given city', inputSchema: z.object({ location: z.string().describe('The location to get the weather for'), }), execute: getWeather, }), }, stopWhen: stepCountIs(5), }); return result.text; } ```

AI agent tools must follow Workflow rules

When providing tools to AI agents via the Vercel AI SDK, tool functions run in Workflow context and must follow Workflow rules. This means they must call Activities or Child Workflows to perform non-deterministic operations like API calls. External API calls should be implemented as Activities and called from the tool function.

Simple haiku agent workflow

To implement a durable haiku agent using the AI SDK, create a Workflow that calls generateText() with temporalProvider.languageModel('gpt-4o-mini') as the model. The string you provide is passed to your configured modelProvider to create the model. Example: ```ts import { generateText } from 'ai'; import { temporalProvider } from '@temporalio/ai-sdk/workflow'; export async function haikuAgent(prompt: string): Promise<string> { const result = await generateText({ model: temporalProvider.languageModel('gpt-4o-mini'), prompt, system: 'You only respond in haikus.', }); return result.text; } ```

Worker configuration with AI SDK example

Example Worker setup: ```ts import { openai } from '@ai-sdk/openai'; import { AiSdkPlugin } from '@temporalio/ai-sdk'; const worker = await Worker.create({ plugins: [ new AiSdkPlugin({ modelProvider: openai, }), ], connection, namespace: 'default', taskQueue: 'ai-sdk', workflowsPath: require.resolve('./workflows'), activities, }); ```

Provider credentials only needed on Worker

The Worker process must have access to your AI provider API credentials, typically read from environment variables. The client application that sends requests to the Temporal Service to start Workflow Executions does not need to know about these credentials.

AI SDK plugin import paths

Import the AiSdkPlugin from @temporalio/ai-sdk in your Worker setup. Import Workflow-side helpers such as temporalProvider and TemporalMCPClient from the @temporalio/ai-sdk/workflow subpath. The /workflow subpath is safe to use in the deterministic Workflow sandbox.

AI SDK integration overview

Temporal's integration with Vercel's AI SDK lets you use the AI SDK's API directly in Workflow code while Temporal handles Durable Execution. LLM API calls are non-deterministic and must run as Activities. The AI SDK plugin wraps calls to methods such as generateText(), streamText(), and streamObject() in Activities automatically, preserving the Vercel AI SDK's developer experience while Temporal handles Durable Execution.

Configure Worker to use AI SDK plugin

Install the @temporalio/ai-sdk package with: npm install @temporalio/ai-sdk. In your worker.ts file, configure the Worker with AiSdkPlugin in the plugins array, specifying a modelProvider such as openai. The modelProvider specifies which AI provider to use when creating models. Configure your Worker to poll a specific Task Queue and Namespace, and ensure your Client application uses the same Task Queue and Namespace.

Stream object agent example

Example Workflow that streams a structured object: ```ts export async function streamObjectAgent(prompt: string): Promise<string> { new WorkflowStream(); const result = streamObject({ model: objectStreamingProvider.languageModel('gpt-4o-mini'), schema: z.object({ recipe: z.object({ name: z.string(), ingredients: z.array(z.object({ name: z.string(), amount: z.string() })), steps: z.array(z.string()), }), }), prompt, }); // External subscribers see the object build up incrementally via the partial // JSON deltas published to the streaming topic. The workflow drains the // partial stream and durably resolves the final, validated object. for await (const _partial of result.partialObjectStream) { // Draining drives the stream; the consumer renders the live partials. } const object = await result.object; await sleep('500 milliseconds'); return object.recipe.name; } ```

Stream structured output with streamObject

streamObject() streams a structured object as its JSON is generated. It flows through the same model path as streamText(), so no extra wiring is needed beyond choosing a distinct streamingTopic so concurrent streams stay separable. External subscribers see the object build up incrementally, and the Workflow durably resolves the final, validated object.

Subscribe to Workflow stream topic as external consumer

External consumers subscribe to a Workflow's stream topic by Workflow ID and render each delta as it arrives. Each item's payload is the JSON-encoded AI SDK stream part: ```ts async function renderStream(client: Client, workflowId: string, topic: string): Promise<void> { const streamClient = WorkflowStreamClient.create(client, workflowId); for await (const item of streamClient.subscribe<Uint8Array>(topic, 0, { resultType: true })) { const part = JSON.parse(new TextDecoder().decode(item.data)); if (part.type === 'text-delta') process.stdout.write(part.delta); if (part.type === 'finish') break; } process.stdout.write('\n'); // Acknowledge receipt so the Workflow can complete without racing against // its in-memory stream log being discarded await client.workflow.getHandle(workflowId).signal(consumerDoneSignal); } ```

Streaming agent workflow example

Example streaming agent Workflow that hosts a WorkflowStream and consumes AI SDK stream output: ```ts export async function streamingAgent(prompt: string): Promise<string> { // Host the WorkflowStream as the first statement so its publish-signal handler // is registered before the streaming activity starts publishing deltas new WorkflowStream(); let consumerDone = false; setHandler(consumerDoneSignal, () => { consumerDone = true; }); const result = streamText({ model: streamingProvider.languageModel('gpt-4o-mini'), prompt, system: 'You only respond in haikus.', }); // Inside the workflow, deltas are replayed after the activity completes, // so this loop durably reassembles the full text let text = ''; for await (const delta of result.textStream) { text += delta; } // Wait for the subscriber to signal that it received the last delta, // with a 10 second timeout as a fallback await condition(() => consumerDone, '10 seconds'); return text; } ```

Configure streaming provider with topic

To stream text, configure a TemporalProvider with a streamingTopic to enable streaming and name the topic that deltas are published to: ```ts import { streamText } from 'ai'; import { TemporalProvider } from '@temporalio/ai-sdk/workflow'; export const STREAM_TOPIC = 'text-stream'; const streamingProvider = new TemporalProvider({ languageModel: { streamingTopic: STREAM_TOPIC }, }); ```

Install workflow streams package for streaming

Streaming uses the @temporalio/workflow-streams package. Install it alongside the AI SDK plugin with: npm install @temporalio/workflow-streams

Stream model output from Workflow

The AI SDK's streamText() and streamObject() functions stream a response incrementally instead of returning it all at once. The plugin runs the model call in an Activity and publishes each delta onto a Workflow Stream topic. External consumers can subscribe to that topic by Workflow ID to render tokens live as they arrive, while the Workflow durably reassembles the final result.

Streaming behavior inside Workflow vs external consumers

Inside the Workflow, the streamed result is reassembled after the Activity completes, so the Workflow's own textStream/partialObjectStream is not incremental. Live, token-by-token deltas are delivered to external consumers through a WorkflowStreamClient. This is inherent to Durable Execution: the Activity must run to completion before the Workflow can observe a deterministic, replayable result.

Workflow registration with StrandsPlugin

Create a Worker with Worker.create(), passing workflowsPath pointing to Workflow files, activities object for Activity implementations, and plugins array containing new StrandsPlugin(). The plugin automatically registers Activities for model calls. Omit models option to use default BedrockModel under name 'bedrock'.

OpenTelemetry integration with StrandsPlugin

Register OpenTelemetryPlugin on both client and Worker alongside StrandsPlugin. You get OpenTelemetry spans around model, tool, and MCP Activities plus spans Strands emits inside invoke.

Agent snapshots not supported in Temporal

TemporalAgent.takeSnapshot() and TemporalAgent.loadSnapshot() throw. Temporal's Event History persists Workflow state durably at finer granularity than Strands snapshots, making snapshots redundant inside a Workflow.

Stream agent output with streamingTopic

Pass streamingTopic: '...' to TemporalAgent and host a WorkflowStream on the Workflow via @temporalio/workflow-streams. Each model stream event is published on the named topic from inside the model Activity. Subscribers read events through WorkflowStreamClient. Chunks batch on streamingBatchInterval (default '100 milliseconds').

Strands Agents plugin currently experimental

The Temporal TypeScript SDK integration with Strands Agents is at experimental release stage. The API may change in future versions.

Configure retries with activityOptions.retry

TemporalAgent disables Strands' built-in ModelRetryStrategy so retries are handled by Temporal. Configure retries with activityOptions.retry on TemporalAgent for model calls, and on Activity options for workflow.activityAsTool, workflow.activityAsHook, and TemporalMCPClient. Passing retryStrategy to TemporalAgent throws.

Continue-as-New for long-running chat sessions

For chat-style Workflows with long message history, use Continue-as-New to start a fresh Workflow execution while carrying agent.messages forward as input. Check workflowInfo().continueAsNewSuggested after each turn to know when to hand off.

Hook callbacks must be deterministic

Hook callbacks run in Workflow context and must be deterministic. Do not use Date.now(), randomUUID(), or I/O inside hook callbacks. Use workflow.activityAsHook for anything requiring I/O.

Give your agent this brain