# Terminal 3: Run a research query
uv run python -m start_workflow "What are the latest advances in quantum computing?"
```

---

## AI Integrations

Temporal Python SDK provides integrations with the following tools and services:

- [Braintrust](./braintrust.mdx)

---

## Interceptors - Python SDK

The behavior of the Python SDK can be customized in many useful ways by modifying inbound and outbound calls using Interceptors.
This is similar to the use of "middleware" in web frameworks such as [Django](https://docs.djangoproject.com/en/5.2/topics/http/middleware/), [Starlette](https://www.starlette.io/middleware/), and [Flask](https://flask.palletsprojects.com/en/stable/lifecycle/#middleware).

The methods you implement on your Interceptor classes can perform arbitrary side effects.
Interceptors can also perform arbitrary modifications to incoming and outgoing data before it is received by the SDK's "real" implementation.

There are five categories of inbound and outbound calls that you can modify in this way:

#### [Outbound Client calls](https://python.temporal.io/temporalio.client.OutboundInterceptor.html)

- `start_workflow()`
- `signal_workflow()`
- `list_workflows()`
- `update_schedule()`

This is not an exhaustive list; refer to the [Python SDK methods](https://python.temporal.io/temporalio.client.OutboundInterceptor.html) for more.

#### [Inbound Workflow calls](https://python.temporal.io/temporalio.worker.WorkflowInboundInterceptor.html)

- `execute_workflow()` (i.e. handle a Workflow Task that is starting a new Workflow Execution)
- `handle_query()`
- `handle_signal()`
- `handle_update_handler()`
- `handle_update_validator()`

#### [Outbound Workflow calls](https://python.temporal.io/temporalio.worker.WorkflowOutboundInterceptor.html)

- `start_activity()`
- `start_child_workflow()`
- `signal_child_workflow()`
- `signal_external_workflow()`
- `start_nexus_operation()`
- `start_local_activity()`

#### [Inbound Activity calls](https://python.temporal.io/temporalio.worker.ActivityInboundInterceptor.html)

- `execute_activity()`
  - i.e. handle a task to execute an Activity (this is the only Inbound Activity call)

#### [Outbound Activity calls](https://python.temporal.io/temporalio.worker.ActivityOutboundInterceptor.html)

- `info()`
- `heartbeat()`

The first of these categories is a Client call, and the remaining 4 are Worker calls.

## Client call Interceptors

To modify outbound Client calls, define a class inheriting from [`client.Interceptor`](https://python.temporal.io/temporalio.client.Interceptor.html), and implement the method `intercept_client()` to return an instance of [`OutboundInterceptor`](https://python.temporal.io/temporalio.client.OutboundInterceptor.html) that implements the subset of outbound Client calls that you wish to modify.

This example implements an Interceptor on outbound Client calls that sets a certain key in the outbound `headers` field.
A User ID is context-propagated by being sent in a header field with outbound requests:

```python
class ContextPropagationInterceptor(
    temporalio.client.Interceptor, temporalio.worker.Interceptor
):
    def __init__(
        self,
        payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter,
    ) -> None:
        self._payload_converter = payload_converter

    def intercept_client(
        self, next: temporalio.client.OutboundInterceptor
    ) -> temporalio.client.OutboundInterceptor:
        return _ContextPropagationClientOutboundInterceptor(
            next, self._payload_converter
        )

def set_header_from_context(
    input: _InputWithHeaders, payload_converter: temporalio.converter.PayloadConverter
) -> None:
    user_id_val = user_id.get()
    if user_id_val:
        input.headers = {
            **input.headers,
            HEADER_KEY: payload_converter.to_payload(user_id_val),
        }

class _ContextPropagationClientOutboundInterceptor(
    temporalio.client.OutboundInterceptor
):
    def __init__(
        self,
        next: temporalio.client.OutboundInterceptor,
        payload_converter: temporalio.converter.PayloadConverter,
    ) -> None:
        super().__init__(next)
        self._payload_converter = payload_converter

    async def start_workflow(
        self, input: temporalio.client.StartWorkflowInput
    ) -> temporalio.client.WorkflowHandle[Any, Any]:
        set_header_from_context(input, self._payload_converter)
        return await super().start_workflow(input)
```

It often happens that your Worker and Client interceptors will share code because they implement closely related logic.
In the Python SDK, you will typically want to create an interceptor class that inherits from _both_ `client.Interceptor` and `worker.Interceptor` as above, since their method sets do not overlap.

You can then pass this in the `interceptors` argument of `Client.connect()` in your client/starter code:

```python
client = await Client.connect(
    "localhost:7233",
    interceptors=[interceptor.ContextPropagationInterceptor()],
)
```

The `interceptors` list can contain multiple interceptors.
In this case they form a chain: a method implemented on an interceptor instance in the list can perform side effects, and modify the data, before passing it on to the corresponding method on the next interceptor in the list.

Your interceptor classes need not implement every method; the default implementation is always to pass the data on to the next method in the interceptor chain.
During execution, when the SDK encounters an Inbound Activity call, it will look to the first Interceptor instance, get hold of the appropriate intercepted method, and call it.
The intercepted method will perform its function then call the same method on the next Interceptor in the chain.
At the end of the chain the SDK will call the "real" SDK method.

## Worker call Interceptors

To modify inbound and outbound Workflow and Activity calls, define a class inheriting from `worker.Interceptor`.
This is an interface with two methods named `intercept_activity` and `workflow_interceptor_class`, which you can use to configure interceptions of Activity and Workflow calls, respectively.
`intercept_activity` returns an `ActivityInboundInterceptor`.

This example demonstrates using an interceptor to measure [Schedule-To-Start](/encyclopedia/detecting-activity-failures#schedule-to-start-timeout) latency:

```python
from temporalio.worker import (
    ActivityInboundInterceptor,
    ExecuteActivityInput,
    Interceptor,
    Worker,
)

class SimpleWorkerInterceptor(Interceptor):
    def intercept_activity(
        self, next: ActivityInboundInterceptor
    ) -> ActivityInboundInterceptor:
        return CustomScheduleToStartInterceptor(next)

class CustomScheduleToStartInterceptor(ActivityInboundInterceptor):
    async def execute_activity(self, input: ExecuteActivityInput):

        schedule_to_start = (
            activity.info().started_time
            - activity.info().current_attempt_scheduled_time
        )

        meter = activity.metric_meter()
        histogram = meter.create_histogram_timedelta(
            "custom_activity_schedule_to_start_latency",
            description="Time between activity scheduling and start",
            unit="duration",
        )
        histogram.record(
            schedule_to_start, {"workflow_type": activity.info().workflow_type}
        )
        return await self.next.execute_activity(input)

client = await Client.connect(
    "localhost:7233",
)
worker = Worker(
    client,
    interceptors=[SimpleWorkerInterceptor()],
    # ...
)
```

:::note

If you are inheriting methods from both `client.Interceptor` and `worker.Interceptor`, you should not pass your Interceptors directly to the `Worker()` constructor — instead, pass it to `Client.connect()`.
This will allow a Worker to use them from the underlying Client.
In other words, only pass the Interceptors to the `Worker()` if you are not using Client methods.

:::

The `workflow_interceptor_class` returns a `WorkflowInboundInterceptor` that works similarly to `ActivityInboundInterceptor`.

---

## Workflow message passing - Python SDK

A Workflow can act like a stateful web service that receives messages: Queries, Signals, and Updates.
The Workflow implementation defines these endpoints via handler methods that can react to incoming messages and return values.
Temporal Clients use messages to read Workflow state and control its execution.
See [Workflow message passing](/encyclopedia/workflow-message-passing) for a general overview of this topic.
This page introduces these features for the Temporal Python SDK.

## Write message handlers {#writing-message-handlers}

:::info
The code that follows is part of a working message passing [sample](https://github.com/temporalio/samples-python/tree/main/message_passing/introduction).
:::

Follow these guidelines when writing your message handlers:

- Message handlers are defined as methods on the Workflow class, using one of the three decorators: [`@workflow.query`](https://python.temporal.io/temporalio.workflow.html#query), [`@workflow.signal`](https://python.temporal.io/temporalio.workflow.html#signal), and [`@workflow.update`](https://python.temporal.io/temporalio.workflow.html#update).
- The parameters and return values of handlers and the main Workflow function must be [serializable](/dataconversion).
- Prefer [data classes](https://docs.python.org/3/library/dataclasses.html) to multiple input parameters.
  Data class parameters allow you to add fields without changing the calling signature.

### Query handlers {#queries}

A [Query](/sending-messages#sending-queries) is a synchronous operation that retrieves state from a Workflow Execution:

```python
class Language(IntEnum):
    Chinese = 1
    English = 2
    French = 3

@dataclass
class GetLanguagesInput:
    include_unsupported: bool

@workflow.defn
class GreetingWorkflow:
    def __init__(self) -> None:
        self.greetings = {
            Language.CHINESE: "你好，世界",
            Language.ENGLISH: "Hello, world",
        }

    @workflow.query
    def get_languages(self, input: GetLanguagesInput) -> list[Language]:
        # 👉 A Query handler returns a value: it can inspect but must not mutate the Workflow state.
        if input.include_unsupported:
            return list(Language)
        else:
            return list(self.greetings)
```

- The Query decorator can accept arguments.
  Refer to the API docs: [`@workflow.query`](https://python.temporal.io/temporalio.workflow.html#query).

- A Query handler uses `def`, not `async def`.
  You can't perform async operations like executing an Activity in a Query handler.

### Signal handlers {#signals}

A [Signal](/sending-messages#sending-signals) is an asynchronous message sent to a running Workflow Execution to change its state and control its flow:

```python
@dataclass
class ApproveInput:
    name: str

@workflow.defn
class GreetingWorkflow:
    ...
    @workflow.signal
    def approve(self, input: ApproveInput) -> None:
        # 👉 A Signal handler mutates the Workflow state but cannot return a value.
        self.approved_for_release = True
        self.approver_name = input.name
```

- The Signal decorator can accept arguments.
  Refer to the API docs: [`@workflow.signal`](https://python.temporal.io/temporalio.workflow.html#signal).

- The handler should not return a value.
  The response is sent immediately from the server, without waiting for the Workflow to process the Signal.

- Signal (and Update) handlers can be `async def`.
  This allows you to use Activities, Child Workflows, durable [`asyncio.sleep`](https://docs.python.org/3/library/asyncio-task.html#asyncio.sleep) Timers, [`workflow.wait_condition`](https://python.temporal.io/temporalio.workflow.html#wait_condition) conditions, and more.
  See [Async handlers](#async-handlers) and [Workflow message passing](/encyclopedia/workflow-message-passing) for guidelines on safely using async Signal and Update handlers.

### Update handlers and validators {#updates}

An [Update](/sending-messages#sending-updates) is a trackable synchronous request sent to a running Workflow Execution.
It can change the Workflow state, control its flow, and return a result.
The sender must wait until the Worker accepts or rejects the Update.
The sender may wait further to receive a returned value or an exception if something goes wrong:

```python
class Language(IntEnum):
    Chinese = 1
    English = 2
    French = 3

@workflow.defn
class GreetingWorkflow:
    ...
    @workflow.update
    def set_language(self, language: Language) -> Language:
        # 👉 An Update handler can mutate the Workflow state and return a value.
        previous_language, self.language = self.language, language
        return previous_language

    @set_language.validator
    def validate_language(self, language: Language) -> None:
        if language not in self.greetings:
            # 👉 In an Update validator you raise any exception to reject the Update.
            raise ValueError(f"{language.name} is not supported")
```

- The Update decorator can take arguments (like, `name`, `dynamic` and `unfinished_policy`) as described in the API reference docs for [`workflow.update`](https://python.temporal.io/temporalio.workflow.html#update).

- About validators:
  - Use validators to reject an Update before it is written to History.
    Validators are always optional.
    If you don't need to reject Updates, you can skip them.
  - The SDK automatically provides a validator decorator named `@<update-handler-name>.validator`.
    The validator must accept the same argument types as the handler and return `None`.

- Accepting and rejecting Updates with validators:
  - To reject an Update, raise an exception of any type in the validator.
  - Without a validator, Updates are always accepted.
- Validators and Event History:
  - The `WorkflowExecutionUpdateAccepted` event is written into the History whether the acceptance was automatic or programmatic.
  - When a Validator raises an error, the Update is rejected and `WorkflowExecutionUpdateAccepted` _won't_ be added to the Event History.
    The caller receives an "Update failed" error.

- Use [`workflow.current_update_info`](https://python.temporal.io/temporalio.workflow.html#current_update_info) to obtain information about the current Update.
  This includes the Update ID, which can be useful for deduplication when using Continue-As-New: see [Ensuring your messages are processed exactly once](/handling-messages#exactly-once-message-processing).
- Update (and Signal) handlers can be `async def`, letting them use Activities, Child Workflows, durable [`asyncio.sleep`](https://docs.python.org/3/library/asyncio-task.html#asyncio.sleep) Timers, [`workflow.wait_condition`](https://python.temporal.io/temporalio.workflow.html#wait_condition) conditions, and more.
  See [Async handlers](#async-handlers) and [Workflow message passing](/encyclopedia/workflow-message-passing) for safe usage guidelines.

## Send messages {#send-messages}

To send Queries, Signals, or Updates, you call methods on a [WorkflowHandle](https://python.temporal.io/temporalio.client.WorkflowHandle.html) object:

- Use [start_workflow](https://python.temporal.io/temporalio.client.Client.html#start_workflow) to start a Workflow and return its handle.

- Use [get_workflow_handle_for](https://python.temporal.io/temporalio.client.Client.html#get_workflow_handle_for) to retrieve a typed Workflow handle by its Workflow Id.

For example:

```python
client = await Client.connect("localhost:7233")
workflow_handle = await client.start_workflow(
    GreetingWorkflow.run, id="greeting-workflow-1234", task_queue="my-task-queue"
)
```

To check the argument types required when sending messages -- and the return type for Queries and Updates -- refer to the corresponding handler method in the Workflow Definition.

:::warning Using Continue-as-New and Updates

- Temporal _does not_ support Continue-as-New functionality within Update handlers.
- Complete all handlers _before_ using Continue-as-New.
- Use Continue-as-New from your main Workflow Definition method, just as you would complete or fail a Workflow Execution.

:::

### Send a Query {#send-query}

Use [`WorkflowHandle.query`](https://python.temporal.io/temporalio.client.WorkflowHandle.html#query) to send a Query to a Workflow Execution:

```python
supported_languages = await workflow_handle.query(
    GreetingWorkflow.get_languages, GetLanguagesInput(supported_only=True)
)
```

- Sending a Query doesn’t add events to a Workflow's Event History.

- You can send Queries to closed Workflow Executions within a Namespace's Workflow retention period.
  This includes Workflows that have completed, failed, or timed out.
  Querying terminated Workflows is not safe and, therefore, not supported.

- A Worker must be online and polling the Task Queue to process a Query.

### Send a Signal {#send-signal}

You can send a Signal to a Workflow Execution from a Temporal Client or from another Workflow Execution.
However, you can only send Signals to Workflow Executions that haven’t closed.

#### Send a Signal from a Client {#send-signal-from-client}

Use [`WorkflowHandle.signal`](https://python.temporal.io/temporalio.client.WorkflowHandle.html#signal) to send a Signal:

```python
await workflow_handle.signal(GreetingWorkflow.approve, ApproveInput(name="me"))
```

- The call returns when the server accepts the Signal; it does _not_ wait for the Signal to be delivered to the Workflow Execution.

- The [WorkflowExecutionSignaled](/references/events#workflowexecutionsignaled) Event appears in the Workflow's Event History.

#### Send a Signal from a Workflow {#send-signal-from-workflow}

A Workflow can send a Signal to another Workflow, known as an _External Signal_.
You'll need a Workflow handle for the external Workflow.
Use [`get_external_workflow_handle_for`](https://python.temporal.io/temporalio.workflow.html#get_external_workflow_handle_for):

<tt>See full sample</tt>

```python