# ...
        activity_result = await workflow.execute_activity(
            your_activity,
            YourParams(greeting, "Retry Policy options"),
            start_to_close_timeout=timedelta(seconds=10),
            # Retry Policy
            retry_policy=RetryPolicy(
                backoff_coefficient=2.0,
                maximum_attempts=5,
                initial_interval=timedelta(seconds=1),
                maximum_interval=timedelta(seconds=2),
                # non_retryable_error_types=["ValueError"],
            ),
        )
```

### Override the retry interval with `next_retry_delay` {#next-retry-delay}

To override the next retry interval set by the current policy, pass `next_retry_delay` when raising an [ApplicationError](/references/failures#application-failure) in an Activity.
This value replaces and overrides whatever the retry interval would normally be on the retry policy.

For example, you can set the delay interval based on an Activity's attempt count.
In the following example, the retry delay starts at 3 seconds after the first attempt.
It increases to 6 seconds for the second attempt, 9 seconds for the third attempt, and so forth.
This creates a steadily increasing backoff, versus the exponential approach used by [backoff coefficients](/encyclopedia/retry-policies#backoff-coefficient):

```python
from temporalio.exceptions import ApplicationError
from datetime import timedelta

@activity.defn
async def my_activity(input: MyActivityInput):
    try:
        # Your activity logic goes here
    except Exception as e:
        attempt = activity.info().attempt
        raise ApplicationError(
            f"Error encountered on attempt {attempt}",
            next_retry_delay=timedelta(seconds=3 * attempt),
        ) from e
```

## Heartbeat an Activity {#activity-heartbeats}

**How to Heartbeat an Activity using the Temporal Python SDK**

An [Activity Heartbeat](/encyclopedia/detecting-activity-failures#activity-heartbeat) is a ping from the [Worker Process](/workers#worker-process) that is executing the Activity to the [Temporal Service](/temporal-service).
Each Heartbeat informs the Temporal Service that the [Activity Execution](/activity-execution) is making progress and the Worker has not crashed.
If the Temporal Service does not receive a Heartbeat within a [Heartbeat Timeout](/encyclopedia/detecting-activity-failures#heartbeat-timeout) time period, the Activity will be considered failed and another [Activity Task Execution](/tasks#activity-task-execution) may be scheduled according to the Retry Policy.

Heartbeats may not always be sent to the Temporal Service—they may be [throttled](/encyclopedia/detecting-activity-failures#throttling) by the Worker.

Activity Cancellations are delivered to Activities from the Temporal Service when they Heartbeat. Activities that don't Heartbeat can't receive a Cancellation.
Heartbeat throttling may lead to Cancellation getting delivered later than expected.

Heartbeats can contain a `details` field describing the Activity's current progress.
If an Activity gets retried, the Activity can access the `details` from the last Heartbeat that was sent to the Temporal Service.

To Heartbeat an Activity Execution in Python, use the [`heartbeat()`](https://python.temporal.io/temporalio.activity.html#heartbeat) API.

```python
@activity.defn
async def your_activity_definition() -> str:
    activity.heartbeat("heartbeat details!")
```

In addition to obtaining cancellation information, Heartbeats also support detail data that persists on the server for retrieval during Activity retry.
If an Activity calls `heartbeat(123, 456)` and then fails and is retried, `heartbeat_details` returns an iterable containing `123` and `456` on the next Run.

#### Set a Heartbeat Timeout {#heartbeat-timeout}

**How to set a Heartbeat Timeout using the Temporal Python SDK**

A [Heartbeat Timeout](/encyclopedia/detecting-activity-failures#heartbeat-timeout) works in conjunction with [Activity Heartbeats](/encyclopedia/detecting-activity-failures#activity-heartbeat).

[`heartbeat_timeout`](https://python.temporal.io/temporalio.worker.StartActivityInput.html#heartbeat_timeout) is a class variable for the [`start_activity()`](https://python.temporal.io/temporalio.workflow.html#start_activity) function used to set the maximum time between Activity Heartbeats.

```python
workflow.start_activity(
    activity="your-activity",
    schedule_to_close_timeout=timedelta(seconds=5),
    heartbeat_timeout=timedelta(seconds=1),
)
```

`execute_activity()` is a shortcut for [`start_activity()`](https://python.temporal.io/temporalio.workflow.html#start_activity) that waits on its result.

To get just the handle to wait and cancel separately, use `start_activity()`. `execute_activity()` should be used in most cases unless advanced task capabilities are needed.

```python
workflow.execute_activity(
    activity="your-activity",
    name,
    schedule_to_close_timeout=timedelta(seconds=5),
    heartbeat_timeout=timedelta(seconds=1),
)
```

---

## Python SDK developer guide

![Python SDK Banner](/img/assets/banner-python-temporal.png)

:::info PYTHON SPECIFIC RESOURCES
Build Temporal Applications with the Python SDK.

**Temporal Python Technical Resources:**

- [Python SDK Quickstart - Setup Guide](https://docs.temporal.io/develop/python/set-up-your-local-python)
- [Python API Documentation](https://python.temporal.io)
- [Python SDK Code Samples](https://github.com/temporalio/samples-python)
- [Python SDK Github](https://github.com/temporalio/sdk-python)
- [Temporal 101 in Python Free Course](https://learn.temporal.io/courses/temporal_101/python/)

**Get Connected with the Temporal Python Community:**

- [Temporal Python Community Slack](https://app.slack.com/client/TNWA8QCGZ)
- [Python SDK Forum](https://community.temporal.io/tag/python-sdk)
  :::

## [Core Application](/develop/python/core-application)

Use the essential components of a Temporal Application (Workflows, Activities, and Workers) to build and run a Temporal application.

- [Develop a Basic Workflow](/develop/python/core-application#develop-workflows)
- [Develop a Basic Activity](/develop/python/core-application#develop-activities)
- [Start an Activity Execution](/develop/python/core-application#activity-execution)
- [Run Worker Processes](/develop/python/core-application#run-a-dev-worker)

## [Temporal Client](/develop/python/temporal-client)

Connect to a Temporal Service and start a Workflow Execution.

- [Connect to Development Temporal Service](/develop/python/temporal-client#connect-to-development-service)
- [Connect to Temporal Cloud](/develop/python/temporal-client#connect-to-temporal-cloud)
- [Start a Workflow Execution](/develop/python/temporal-client#start-workflow-execution)

## [Python SDK Sandbox](/develop/python/python-sdk-sandbox)

Use third-party Python modules without non-deterministic behavior.

## [Python SDK sync vs. async implementations](/develop/python/python-sdk-sync-vs-async)

Implement synchronous or asynchronous Activities.

## [Testing](/develop/python/testing-suite)

Set up the testing suite and test Workflows and Activities.

- [Test Frameworks](/develop/python/testing-suite#test-frameworks)
- [Testing Activities](/develop/python/testing-suite#test-activities)
- [Testing Workflows](/develop/python/testing-suite#test-workflows)
- [How to Replay a Workflow Execution](/develop/python/testing-suite#replay)

## [Failure detection](/develop/python/failure-detection)

Explore how your application can detect failures using timeouts and automatically attempt to mitigate them with retries.

- [Workflow Timeouts](/develop/python/failure-detection#workflow-timeouts)
- [Set Activity Timeouts](/develop/python/failure-detection#activity-timeouts)
- [Heartbeat an Activity](/develop/python/failure-detection#activity-heartbeats)

## [Workflow message passing](/develop/python/message-passing)

Send messages to and read the state of Workflow Executions.

- [Develop with Signals](/develop/python/message-passing#signals)
- [Develop with Queries](/develop/python/message-passing#queries)
- [Develop with Updates](/develop/python/message-passing#updates)
- [What is a Dynamic Handler](/develop/python/message-passing#dynamic-handler)

## [Interrupt a Workflow feature guide](/develop/python/cancellation)

Interrupt a Workflow Execution with a Cancel or Terminate action.

- [Cancel a Workflow](/develop/python/cancellation#cancellation)
- [Terminate a Workflow](/develop/python/cancellation#termination)
- [Reset a Workflow](/develop/python/cancellation#reset): Resume a Workflow Execution from an earlier point in its Event History.
- [Cancel an Activity from a Workflow](/develop/python/cancellation#cancel-activity)

## [Asynchronous Activity completion](/develop/python/asynchronous-activity-completion)

Complete Activities asynchronously.

- [Asynchronously Complete an Activity](/develop/python/asynchronous-activity-completion)

## [Versioning](/develop/python/versioning)

Change Workflow Definitions without causing non-deterministic behavior in running Workflows.

- [Introduction to Versioning](/develop/python/versioning)
- [How to Use the Patching API](/develop/python/versioning#patching)

## [Observability](/develop/python/observability)

Configure and use the Temporal Observability APIs.

- [Emit Metrics](/develop/python/observability#metrics)
- [Set up tracing](/develop/python/observability#tracing)
- [Log from a Workflow](/develop/python/observability#logging)
- [Use Visibility APIs](/develop/python/observability#visibility)

## [Debugging](/develop/python/debugging)

Explore various ways to debug your application.

- [Debugging](/develop/python/debugging)

## [Schedules](/develop/python/schedules)

Run Workflows on a schedule and delay the start of a Workflow.

- [Schedule a Workflow](/develop/python/schedules#schedule-a-workflow)
- [Temporal Cron Jobs](/develop/python/schedules#temporal-cron-jobs)
- [Start Delay](/develop/python/schedules#start-delay)

## [Data encryption](/develop/python/converters-and-encryption)

Use compression, encryption, and other data handling by implementing custom converters and codecs.

- [Custom Payload Codec](/develop/python/converters-and-encryption#custom-payload-codec)
- [Payload Conversion](/develop/python/converters-and-encryption#payload-conversion)

## Temporal Nexus

The [Temporal Nexus](/develop/python/nexus) feature guide shows how to use Temporal Nexus to connect durable executions within and across Namespaces using a Nexus Endpoint, a Nexus Service contract, and Nexus Operations.

- [Create a Nexus Endpoint to route requests from caller to handler](/develop/python/nexus#create-nexus-endpoint)
- [Define the Nexus Service contract](/develop/python/nexus#define-nexus-service-contract)
- [Develop a Nexus Service and Operation handlers](/develop/python/nexus#develop-nexus-service-operation-handlers)
- [Develop a caller Workflow that uses a Nexus Service](/develop/python/nexus#develop-caller-workflow-nexus-service)
- [Make Nexus calls across Namespaces with a dev Server](/develop/python/nexus#register-the-caller-workflow-in-a-worker-and-start-the-caller-workflow)
- [Make Nexus calls across Namespaces in Temporal Cloud](/develop/python/nexus#nexus-calls-across-namespaces-temporal-cloud)

## [Durable Timers](/develop/python/timers)

Use Timers to make a Workflow Execution pause or "sleep" for seconds, minutes, days, months, or years.

- [Sleep](/develop/python/timers)

## [Child Workflows](/develop/python/child-workflows)

Explore how to spawn a Child Workflow Execution and handle Child Workflow Events.

- [Start a Child Workflow Execution](/develop/python/child-workflows)

## [Continue-As-New](/develop/python/continue-as-new)

Continue the Workflow Execution with a new Workflow Execution using the same Workflow ID.

- [Continue-As-New](/develop/python/continue-as-new)

## [Interceptors](/develop/python/interceptors)

Manage inbound and outbound SDK calls, enhance tracing, and add authorization to your Workflows and Activities.

- [Interceptors](/develop/python/interceptors)

## [Enriching the User Interface](/develop/python/enriching-ui)

Add descriptive information to workflows and events for better visibility and context in the UI.

- [Adding Summary and Details to Workflows](/develop/python/enriching-ui#adding-summary-and-details-to-workflows)

---

## Braintrust Integration

Temporal's integration with [Braintrust](https://braintrust.dev) gives you full observability into your AI agent
Workflows—tracing every LLM call, managing prompts without code deploys, and tracking costs across models.

When building AI agents with Temporal, you get durable execution: automatic retries, state persistence, and the ability
to recover from failures mid-workflow. Braintrust adds the observability layer: see exactly what your agents are doing,
iterate on prompts in a UI, and measure whether changes improve outcomes.

The integration connects these capabilities with minimal code changes. Every Workflow and Activity becomes a span in
Braintrust, and every LLM call is traced with inputs, outputs, tokens, and latency.

:::info

The Temporal Python SDK integration with Braintrust is currently in
[Public Preview](/evaluate/development-production-features/release-stages#public-preview). Refer to the
[Temporal product release stages guide](/evaluate/development-production-features/release-stages) for more information.

:::

All code snippets in this guide are taken from the
[deep research sample](https://github.com/braintrustdata/braintrust-cookbook/blob/main/examples/TemporalDeepResearch/TemporalDeepResearch.mdx). Refer to the sample for the
complete code and run it locally.

## Prerequisites

- This guide assumes you are already familiar with Braintrust. If you aren't, refer to the
  [Braintrust documentation](https://www.braintrust.dev/docs) for more details.
- If you are new to Temporal, we recommend reading [Understanding Temporal](/evaluate/understanding-temporal) or taking
  the [Temporal 101](https://learn.temporal.io/courses/temporal_101/) course.
- Ensure you have set up your local development environment by following the
  [Set up your local development environment](/develop/python/core-application) guide. When you're done, leave the
  Temporal Development Server running if you want to test your code locally.

## Configure Workers to use Braintrust

Workers execute the code that defines your Workflows and Activities. To trace Workflow and Activity execution in
Braintrust, add the `BraintrustPlugin` to your Worker.

Follow the steps below to configure your Worker.

1. Install the Braintrust SDK with Temporal support.

   ```bash
   uv pip install "braintrust[temporal]"
   ```

2. Initialize the Braintrust logger before creating your Worker. The logger must be initialized first so that spans are
   properly connected.

   ```python

   from braintrust import init_logger

   # Initialize BEFORE creating the Temporal client or worker
   init_logger(project=os.environ.get("BRAINTRUST_PROJECT", "my-project"))
   ```

3. Add the `BraintrustPlugin` to your Worker.

   ```python
   from braintrust.contrib.temporal import BraintrustPlugin
   from temporalio.worker import Worker

   worker = Worker(
       client,
       task_queue="my-task-queue",
       workflows=[MyWorkflow],
       activities=[my_activity],
       plugins=[BraintrustPlugin()],  # Add this line
   )
   ```

4. Add the plugin to your Temporal Client as well. This enables span context propagation, linking client code to the
   Workflows it starts.

   ```python
   from temporalio.client import Client
   from braintrust.contrib.temporal import BraintrustPlugin

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

5. Run the Worker. Ensure the Worker process has access to your Braintrust API key via the `BRAINTRUST_API_KEY`
   environment variable.

   ```bash
   export BRAINTRUST_API_KEY="your-api-key"
   python worker.py
   ```

   :::tip

   You only need to provide API credentials to the Worker process. The client application that starts Workflow
   Executions doesn't need the Braintrust API key.

   :::

## Trace LLM calls with wrap_openai

The simplest way to trace LLM calls is to wrap your OpenAI client. Every call through the wrapped client automatically
creates a span in Braintrust with inputs, outputs, token counts, and latency.

```python
from braintrust import wrap_openai
from openai import AsyncOpenAI