Metadata-Version: 2.4
Name: temporalio-strands-agents
Version: 0.1.0
Summary: Temporal integration for strands agents
Author: Temporal Technologies Inc
Author-email: Temporal Technologies Inc <sdk@temporal.io>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Dist: temporalio>=1.34.0,<2
Requires-Dist: strands-agents>=1.51.0
Requires-Dist: mcp>=1.9.4,<2
Requires-Dist: botocore>=1.34.0,<2
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/temporalio/ai-integrations/tree/main/python/strands_agents
Project-URL: Repository, https://github.com/temporalio/ai-integrations
Project-URL: Documentation, https://docs.temporal.io/develop/python/integrations/strands-agents
Description-Content-Type: text/markdown

# Strands Agents

> Release stage: [Public Preview](https://docs.temporal.io/develop/python/integrations/strands-agents).

This Temporal [Plugin](https://docs.temporal.io/develop/plugins-guide) allows you to run [Strands Agents](https://strandsagents.com/) inside Temporal Workflows, routing model invocations, tool calls, and MCP tool calls through Temporal Activities for durable execution, Temporal-managed retries, and timeouts.

## Installation

```sh
uv add temporalio-strands-agents
```

## Quickstart

`workflow.py` defines the workflow and runs the worker:

```python
import asyncio
from datetime import timedelta

from temporalio import workflow
from temporalio.client import Client
from temporalio.strands_agents import StrandsPlugin, TemporalAgent
from temporalio.worker import Worker


@workflow.defn
class MyWorkflow:
    def __init__(self) -> None:
        self.agent = TemporalAgent(start_to_close_timeout=timedelta(seconds=60))

    @workflow.run
    async def run(self, prompt: str) -> str:
        result = await self.agent.invoke_async(prompt)
        return str(result)


async def main() -> None:
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="strands",
        workflows=[MyWorkflow],
        plugins=[StrandsPlugin()],
    )
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())
```

`client.py` starts the workflow:

```python
import asyncio

from temporalio.client import Client

from workflow import MyWorkflow


async def main() -> None:
    client = await Client.connect("localhost:7233")
    result = await client.execute_workflow(
        MyWorkflow.run,
        "Hello",
        id="strands-quickstart",
        task_queue="strands",
    )
    print(result)


if __name__ == "__main__":
    asyncio.run(main())
```

Note: Use `agent.invoke_async(message)` instead of `agent(message)`. The synchronous form spawns a worker thread, which the workflow sandbox blocks.

## Models

`StrandsPlugin(models=...)` takes a mapping of `name → factory`. Each factory is called lazily on first use (on the worker, outside the workflow sandbox) and the constructed model is cached for the worker's lifetime. `TemporalAgent(model="name", ...)` selects which factory to invoke and carries the activity options for that agent's model calls. If `models` is omitted, the plugin registers a single `BedrockModel` factory under the name `"bedrock"` with Botocore retries disabled so Temporal owns retries.

```python
from botocore.config import Config as BotocoreConfig
from strands.models.anthropic import AnthropicModel
from strands.models.bedrock import BedrockModel

# workflow
@workflow.defn
class MultiModelWorkflow:
    def __init__(self) -> None:
        self.agent_a = TemporalAgent(
            model="claude",
            start_to_close_timeout=timedelta(seconds=60),
        )
        self.agent_b = TemporalAgent(
            model="bedrock",
            start_to_close_timeout=timedelta(seconds=60),
        )

# worker
Worker(..., plugins=[StrandsPlugin(models={
    "claude": lambda: AnthropicModel(client_args={"api_key": "..."}),
    "bedrock": lambda: BedrockModel(
        boto_client_config=BotocoreConfig(retries={"max_attempts": 0})
    ),
})])
```

Each `TemporalAgent` carries its own activity options (timeouts, retry policy, task queue, streaming topic) and dispatches to the shared model activity, which resolves the model name against the registered factories at runtime. A name not present in `models` raises `ValueError` inside the activity.

## Retries

`TemporalAgent` disables Strands' built-in `ModelRetryStrategy` so retries are handled exclusively by Temporal. The plugin's default Bedrock model also disables Botocore retries. When supplying your own model factory, disable that provider client's retries as shown in the Bedrock example above.

Configure retries via `retry_policy` on `TemporalAgent`, and on the activity options accepted by `workflow.activity_as_tool`, `workflow.activity_as_hook`, and `TemporalMCPClient`:

```python
from temporalio.common import RetryPolicy

TemporalAgent(
    start_to_close_timeout=timedelta(seconds=60),
    retry_policy=RetryPolicy(maximum_attempts=3),
)
```

Passing `retry_strategy=...` to `TemporalAgent(...)` raises `ValueError`; remove the argument (or pass `retry_strategy=None`) and put the retry config on the activity options instead.

## Snapshots

`TemporalAgent.take_snapshot()` and `TemporalAgent.load_snapshot()` raise `NotImplementedError`. Temporal's event history already persists workflow state durably at a finer granularity than Strands snapshots, so calling either inside a workflow is redundant.

## Structured Output

Like Strands `Agent`, `TemporalAgent` supports structured output with `structured_output_model`. The plugin defaults to [`pydantic_data_converter`](https://github.com/temporalio/sdk-python/tree/6adc0d84290a79952dee3ef02c36f6ed9334874a/temporalio/contrib/pydantic), so Pydantic types easily serialize across the activity and workflow boundary.

```python
from pydantic import BaseModel

class PersonInfo(BaseModel):
    name: str
    age: int

@workflow.defn
class MyWorkflow:
    def __init__(self) -> None:
        self.agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=60),
            structured_output_model=PersonInfo,
        )

    @workflow.run
    async def run(self, prompt: str) -> PersonInfo:
        result = await self.agent.invoke_async(prompt)
        return result.structured_output
```

## Streaming

To forward model chunks to external consumers, pass `streaming_topic="..."` to `TemporalAgent` and host a `WorkflowStream` on the workflow. Each `StreamEvent` is published on the named topic from inside the model activity; subscribers read via `WorkflowStreamClient`. Chunks are batched on `streaming_batch_interval` (default 100ms).

```python
# workflow
@workflow.defn
class MyWorkflow:
    def __init__(self) -> None:
        self.stream = WorkflowStream()
        self.agent = TemporalAgent(streaming_topic="events")

# client
async for item in WorkflowStreamClient.create(client, workflow_id).subscribe(
    ["events"], result_type=StreamEvent,
):
    print(item.data)
```

## Sandboxes

`TemporalSandbox` implements Strands' sandbox API by scheduling every command,
code, and filesystem operation as a Temporal Activity. Register the real
worker-side sandbox under a name, then select that name in workflow code:

```python
from strands.sandbox.docker import DockerSandbox
from temporalio.strands_agents import (
    SandboxWorkflowContext,
    StrandsPlugin,
    TemporalAgent,
    TemporalSandbox,
)

async def build_sandbox(context: SandboxWorkflowContext) -> DockerSandbox:
    # Application-specific and idempotent: return the existing container when
    # another activity worker has already provisioned this Workflow's sandbox.
    container = await get_or_create_build_container(
        context.chain.first_execution_run_id
    )
    return DockerSandbox(container.name)

# workflow
agent = TemporalAgent(
    sandbox=TemporalSandbox(
        "build",
        start_to_close_timeout=timedelta(minutes=5),
    ),
)

# worker
Worker(
    ...,
    plugins=[StrandsPlugin(sandboxes={
        "build": build_sandbox,
    })],
)
```

The plugin registers one shared set of sandbox activities regardless of how
many factories are configured. Each operation carries the selected sandbox
name in its activity input so the worker can dispatch it to the matching
factory.

The factory is called for every sandbox Activity with a `SandboxWorkflowContext`
containing the current `run_id` and a `chain` identity with the Workflow's
namespace, Workflow ID, and first execution Run ID. Use the chain identity to
find or recreate the same backing environment across Activity retries, Worker
restarts, Continue-As-New, Reset, and Cron runs. Unrelated Workflow chains
should not share an environment unless the application explicitly intends that.

The first execution Run ID is supplied by workflow code because Activity
metadata only identifies the current run. Treat sandbox activities as trusted
worker endpoints: workflow code with access to their task queue can call them
directly and choose this value. Do not share the task queue between mutually
untrusted workflows. If Workflow IDs can be reused, the backing environment
lookup must also prevent a new execution from reconnecting to an old execution's
environment, for example by expiring old environments before ID reuse.

Factories may be synchronous or asynchronous. Synchronous factories must only
construct a lightweight adapter and must not block the activity event loop;
use an asynchronous factory for remote lookup or provisioning. Calls may run
concurrently or on different workers, so lookup and provisioning must be
idempotent. Strands' `DockerSandbox` only connects to an already-running
container; it does not create one.

The plugin does not cache adapters. If constructing an adapter is expensive,
cache it in the factory along with any backend-specific liveness and recovery
logic. The cache remains an optimization: every worker process must still be
able to reconnect to the same Workflow-scoped backing environment. Protect
in-process cache misses when concurrent Activities could otherwise provision
the same environment twice.

Return an async context manager from the factory when an adapter owns clients,
sockets, subprocess handles, tunnels, or leases that should be scoped to one
Activity. Its exit method runs after that Activity finishes:

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async def build_managed_sandbox(context: SandboxWorkflowContext):
    adapter = await connect_to_sandbox(context)
    try:
        yield adapter
    finally:
        await adapter.aclose()
```

In either case, provisioning, teardown, and cleanup of orphaned backing
environments remain the application's responsibility; use a backend TTL or
reaper for workflows that are terminated before normal cleanup.

Successive sandbox Activities from one Workflow are routed independently across
the task queue. With more than one worker on the queue, a `write-file` can land
on one worker and the following `read-file` on another. The context factory must
therefore reconnect every worker to the same Workflow-scoped backing environment
rather than relying on per-process state. A single worker on the queue also
satisfies this.

Reset does not roll back commands or filesystem mutations already performed in
the external sandbox, just as it does not roll back other Activity side effects.
Account for that when resetting a Workflow that uses a sandbox.

`SandboxTimeoutError` is serialized as an Activity failure and retried under the
`retry_policy` you pass to `TemporalSandbox`. If retries are exhausted, it is
reconstructed inside the workflow with the sandbox's own message. Any
`FileNotFoundError` raised by a sandbox filesystem operation — including its
`SandboxPathNotFoundError` subclass — is non-retryable because the requested path
is absent, and is reconstructed in the same way. Factory failures and other
sandbox failures, including the `OSError` that Strands documents for a failed
`write_file`, are also retryable.

Like all Temporal Activities, sandbox operations have at-least-once execution
semantics. A worker can finish a command or filesystem mutation and fail before
recording its result, causing a retry to perform the operation again. Use a
bounded `retry_policy`, and make commands and mutations idempotent when repeated
execution would be unsafe.

By default, `TemporalSandbox.get_tools()` vends `sandbox_shell` and
`sandbox_file_editor`. A tool passed explicitly through `TemporalAgent(tools=...)`
with either name takes precedence, following Strands' normal sandbox-tool
override behavior.

Execution output is always buffered into the activity result so workflow replay
observes the same ordered `StreamChunk` and `ExecutionResult` values. For live,
observer-facing output, set `streaming_topic` and host a `WorkflowStream` on the
workflow. The activity publishes each `StreamChunk` as it arrives; the final
`ExecutionResult` is returned only through the buffered activity result:

```python
from datetime import timedelta

from temporalio.strands_agents import SandboxStreamEvent, TemporalSandbox
from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient

# workflow __init__
self.stream = WorkflowStream()
self.sandbox = TemporalSandbox(
    "build",
    start_to_close_timeout=timedelta(minutes=5),
    streaming_topic="sandbox-events",
)

# external client
async for item in WorkflowStreamClient.create(client, workflow_id).subscribe(
    ["sandbox-events"], result_type=SandboxStreamEvent
):
    event = item.data
    print(event.execution_id, event.sequence, event.chunk.data)
```

Each `SandboxStreamEvent` includes the sandbox name, an execution ID composed of
the Workflow Run ID and Activity ID, the Activity attempt, chunk sequence, and
`StreamChunk`. Events from concurrent executions may interleave on a shared
topic; group them by `execution_id` and `attempt`, then order them by `sequence`.
Workflow code still receives the correctly separated, complete buffered result
for each call. Because publications are observer-facing side effects of an
Activity attempt, a failed attempt may leave chunks in the topic before a retry
publishes its own output.

Streaming is disabled by default. When `streaming_topic=None`, sandbox
activities do not construct a `WorkflowStreamClient` and the workflow does not
need to host a `WorkflowStream`.

All sandbox Activity arguments and results are serialized into workflow history.
Keep command output and files within the server's configured payload size limits;
use external storage for large artifacts.

Do not put secret values directly in `env`, because those values are serialized
into workflow history. Use `temporal_worker_env_ref()` to serialize only the
name of a worker environment variable, then allow that name on every worker that
runs the sandbox activities:

```python
from temporalio.strands_agents import temporal_worker_env_ref

# workflow
await sandbox.execute(
    "build",
    env={"API_KEY": temporal_worker_env_ref("BUILD_API_KEY")},
)

# worker
plugin = StrandsPlugin(
    sandboxes={"build": build_sandbox},
    resolvable_worker_env_vars=["BUILD_API_KEY"],
)
```

Names are matched exactly. A reference to a name the worker does not allow is
passed to the sandbox unchanged; an allowed but unset variable resolves to an
empty string. `AllowAllWorkerEnvVars()` permits any name, but should only be used
when all workflow code on the task queue is trusted to read every environment
variable available to that worker.

## Tools

Decorate non-deterministic tools with `@activity.defn`, or if you're importing tools from `strands_tools`, wrap them in a thin async function. Then, register the activity on the worker via `Worker(activities=[...])` and pass it to the agent with `workflow.activity_as_tool(activity, **options)` along with any activity options (e.g. `start_to_close_timeout`):

```python
from strands_tools import shell
from temporalio.strands_agents import workflow as strands_workflow

@activity.defn
async def fetch_user(user_id: str) -> dict:
    ...

@activity.defn(name="shell")
async def shell_activity(command: str) -> dict:
    return shell.shell(command=command, non_interactive=True)

# workflow
agent = TemporalAgent(
    start_to_close_timeout=timedelta(seconds=60),
    tools=[
        strands_workflow.activity_as_tool(fetch_user, start_to_close_timeout=timedelta(seconds=30)),
        strands_workflow.activity_as_tool(shell_activity, start_to_close_timeout=timedelta(seconds=15)),
    ],
)

# worker
Worker(
    ...,
    activities=[fetch_user, shell_activity],
    plugins=[StrandsPlugin(models=MODELS)],
)
```

If a tool's activity fails, the agent receives a failed tool result containing
the activity's error message, and `AfterToolCallEvent.exception` contains the
underlying exception. This also applies when an MCP tool's call-tool activity
fails.

## Hooks

Strands' [hook system](https://strandsagents.com/) (`strands.hooks`) lets you subscribe callbacks to events in the agent lifecycle — invocation start/end, model call before/after, tool call before/after, message added. Pass `hooks=[MyHookProvider()]` to `TemporalAgent`: every single-agent hook event fires in workflow context, so deterministic callbacks just work.

```python
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import AfterToolCallEvent

class AuditHook(HookProvider):
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(AfterToolCallEvent, self._on_tool_call)

    def _on_tool_call(self, event: AfterToolCallEvent) -> None:
        # Pure local state - deterministic across replay.
        workflow.logger.info(f"tool {event.tool_use['name']} finished")

agent = TemporalAgent(start_to_close_timeout=..., hooks=[AuditHook()])
```

Callbacks run in workflow context, so they must be deterministic: no `time.time()`, `uuid.uuid4()`, or I/O — same rules as workflow code. For callbacks that need I/O (audit logging, metrics, alerting), use `workflow.activity_as_hook()` to dispatch the work as a Temporal activity:

```python
from temporalio.strands_agents.workflow import activity_as_hook

@activity.defn
async def persist_tool_call(tool_name: str) -> None:
    # I/O safely in an activity.
    ...

class AuditHook(HookProvider):
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(
            AfterToolCallEvent,
            activity_as_hook(
                persist_tool_call,
                activity_input=lambda event: event.tool_use["name"],
                start_to_close_timeout=timedelta(seconds=10),
            ),
        )
```

`activity_input` extracts serializable values from the event to pass as the activity's input. Use a dataclass or Pydantic model for multiple values. This is needed because events hold references to the `Agent`, `AgentTool` instances, etc., none of which cross the activity boundary.

## Human-in-the-loop interrupts

Strands offers two HITL surfaces; both work with the plugin. In each case, `agent.invoke_async()` returns `AgentResult(stop_reason="interrupt", interrupts=[...])` instead of raising. Pair this with a signal handler that supplies responses, then resume by calling `agent.invoke_async(responses)`.

### Hook-based interrupts

A hook on an interruptible event (e.g. `BeforeToolCallEvent`) can pause the agent by calling `event.interrupt(name, reason=...)`. The hook runs in workflow context, so it must be deterministic — no I/O.

```python
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import BeforeToolCallEvent

class ApprovalHook(HookProvider):
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeToolCallEvent, self._gate)

    def _gate(self, event: BeforeToolCallEvent) -> None:
        if event.interrupt("approval", reason="confirm delete") != "approve":
            event.cancel_tool = "denied"

@workflow.defn
class MyWorkflow:
    def __init__(self) -> None:
        self.agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=60),
            tools=[delete_thing],
            hooks=[ApprovalHook()],
        )
        self._approval: str | None = None

    @workflow.signal
    def approve(self, response: str) -> None:
        self._approval = response

    @workflow.run
    async def run(self, prompt: str) -> str:
        result = await self.agent.invoke_async(prompt)
        if result.stop_reason == "interrupt":
            await workflow.wait_condition(lambda: self._approval is not None)
            result = await self.agent.invoke_async([
                {"interruptResponse": {"interruptId": result.interrupts[0].id, "response": self._approval}}
            ])
        return str(result)
```

### Tool-body interrupts

A `@strands.tool` function can raise `InterruptException(Interrupt(...))` directly. The agent stops with the interrupt, the workflow handles the resume the same way as for hooks.

```python
from strands import tool
from strands.interrupt import Interrupt, InterruptException

@tool
def delete_thing(name: str) -> str:
    raise InterruptException(
        Interrupt(id=f"delete:{name}", name="approval", reason=f"delete {name}?")
    )
```

The same 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 just like the in-workflow case:

```python
from strands.interrupt import Interrupt, InterruptException
from temporalio.strands_agents.workflow import activity_as_tool

@activity.defn
async def delete_thing(name: str) -> str:
    if not await policy.is_authorized(name):
        raise InterruptException(
            Interrupt(id=f"delete:{name}", name="approval", reason=f"delete {name}?")
        )
    await storage.delete(name)
    return f"deleted {name}"

@workflow.defn
class MyWorkflow:
    def __init__(self) -> None:
        self.agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=60),
            tools=[activity_as_tool(delete_thing, start_to_close_timeout=timedelta(seconds=10))],
        )
```

This relies on the plugin's failure converter, which is installed via the client's data converter. **Attach `StrandsPlugin` to the client** (not just the worker) for activity-tool interrupts to work — workers built from that client pick up the plugin automatically.

```python
client = await Client.connect("localhost:7233", plugins=[StrandsPlugin(models=MODELS)])
Worker(client, task_queue="strands", workflows=[MyWorkflow], activities=[delete_thing])
```

## Continue-as-new

A chat-style workflow accumulates history with every turn and will eventually hit Temporal's per-workflow history limit. `workflow.info().is_continue_as_new_suggested()` flips true once the server decides history has grown large enough; check it after each turn and hand off to a fresh run, carrying `agent.messages` as input:

```python
from dataclasses import dataclass, field
from strands.types.content import Messages

@dataclass
class ChatInput:
    messages: Messages = field(default_factory=list)

@workflow.defn
class ChatWorkflow:
    def __init__(self) -> None:
        self._pending: list[str] = []
        self._done = False

    @workflow.signal
    def user_says(self, prompt: str) -> None:
        self._pending.append(prompt)

    @workflow.signal
    def end_chat(self) -> None:
        self._done = True

    @workflow.run
    async def run(self, input: ChatInput) -> None:
        agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=60),
            messages=list(input.messages),
        )
        while True:
            await workflow.wait_condition(lambda: self._pending or self._done)
            if self._done:
                return
            await agent.invoke_async(self._pending.pop(0))
            if workflow.info().is_continue_as_new_suggested():
                workflow.continue_as_new(ChatInput(messages=agent.messages))
```

## MCP

`StrandsPlugin(mcp_clients=...)` takes a mapping of `name → MCPClient factory`, mirroring the `models=` pattern. The plugin registers per-server `{name}-call-tool` and `{name}-list-tools` activities. Workflow-side, `TemporalMCPClient(server="name")` is a pure handle: it references the server by name, discovers tools by running `{name}-list-tools`, and carries the per-call activity options.

```python
from mcp import StdioServerParameters, stdio_client
from strands.tools.mcp.mcp_client import MCPClient
from temporalio.strands_agents import TemporalMCPClient

# workflow
@workflow.defn
class MyWorkflow:
    def __init__(self) -> None:
        echo = TemporalMCPClient(server="echo", start_to_close_timeout=timedelta(seconds=30))
        self.agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=60),
            tools=[echo],
        )

# worker
Worker(
    ...,
    plugins=[StrandsPlugin(
        mcp_clients={
            "echo": lambda: MCPClient(
                lambda: stdio_client(
                    StdioServerParameters(command="...", args=[...]),
                ),
            ),
        },
    )],
)
```

Each factory returns a fully configured `MCPClient`, so you can pass options like `tool_filters`, `prefix`, `elicitation_callback`, or `tasks_config` to it.

By default, `TemporalMCPClient` re-lists the server's tools (via `{name}-list-tools`) on every agent turn, so an MCP server that is restarted mid-workflow — with tools added, removed, or renamed — is picked up. To list the tools just once at the beginning of the workflow and reuse that schema for the workflow's lifetime (one fewer activity per turn), set `cache_tools=True`:

```python
echo = TemporalMCPClient(server="echo", cache_tools=True, start_to_close_timeout=timedelta(seconds=30))
```

To amortize connection setup, the `{name}-call-tool` and `{name}-list-tools` activities share a worker-process MCP connection that is opened lazily and reused across calls. The connection is disconnected after it sits idle for `mcp_connection_idle_timeout` (default 5 minutes); the timer resets on every reuse:

```python
StrandsPlugin(
    mcp_clients={"echo": lambda: MCPClient(...)},
    mcp_connection_idle_timeout=timedelta(seconds=30),
)
```

## Observability

`StrandsPlugin` composes cleanly with [`OpenTelemetryPlugin`](https://github.com/temporalio/sdk-python/tree/6adc0d84290a79952dee3ef02c36f6ed9334874a/temporalio/contrib/opentelemetry). Register `OpenTelemetryPlugin` on the client (workers built from that client pick it up automatically) and `StrandsPlugin` on the worker. You'll get OTel spans around the model, tool, and MCP activities the plugin schedules, plus any spans Strands itself emits inside `invoke_async`:

```python
import opentelemetry.trace
from temporalio.contrib.opentelemetry import OpenTelemetryPlugin, create_tracer_provider

opentelemetry.trace.set_tracer_provider(create_tracer_provider())

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

Worker(
    client,
    task_queue="strands",
    workflows=[MyWorkflow],
    plugins=[StrandsPlugin(models=MODELS)],
)
```

Set the tracer provider before connecting the client. See the [OpenTelemetry plugin README](https://github.com/temporalio/sdk-python/tree/6adc0d84290a79952dee3ef02c36f6ed9334874a/temporalio/contrib/opentelemetry) for exporter setup.
