Money transfer workflow with saga pattern example
from temporalio import workflow
from temporalio.exceptions import ActivityError, ApplicationError
from datetime import timedelta
@workflow.defn
class MoneyTransferWorkflow:
@workflow.run
async def run(self, details):
# Withdraw money
try:
withdraw_result = await workflow.execute_activity(
withdraw,
details,
start_to_close_timeout=timedelta(seconds=10)
)
except ActivityError as e:
raise ApplicationError(
f"Withdrawal failed: {e.cause}",
type="WithdrawalError"
)
# Deposit money
try:
deposit_result = await workflow.execute_activity(
deposit,
details,
start_to_close_timeout=timedelta(seconds=10)
)
except ActivityError as e:
# Deposit failed - attempt refund
try:
await workflow.execute_activity(
refund,
withdraw_result,
start_to_close_timeout=timedelta(seconds=10)
)
raise ApplicationError(
f"Deposit failed but money refunded to source account",
type="DepositError"
)
except ActivityError as refund_err:
raise ApplicationError(
f"Deposit failed and refund also failed: {refund_err.cause}",
type="CriticalTransferError"
)
return f"Transfer complete: {withdraw_result}, {deposit_result}"
Handle Activity exceptions in Workflows with try/except
Use Python's try/except blocks to handle Activity failures in your Workflow. Catch ActivityError to handle Activity failures after exhausting retries, and use the cause field to access the original error. Common Temporal exceptions you can catch include: ActivityError, ChildWorkflowError, CancelledError, and TimeoutError.
Workflow Task retry on non-Temporal exceptions
Raising any Python exception other than a Temporal exception (like ValueError or TypeError) causes a Workflow Task failure, which retries automatically. Regular Python exceptions are treated as bugs that can be fixed with code deployment, not business logic failures. The Workflow Task retries indefinitely, letting you fix the bug and redeploy without losing Workflow state.
Workflow execution failure example in Python
from temporalio import workflow
from temporalio.exceptions import ApplicationError
@workflow.defn
class PizzaDeliveryWorkflow:
@workflow.run
async def run(self, order):
distance = await workflow.execute_activity(
calculate_distance,
order.address,
start_to_close_timeout=timedelta(seconds=10)
)
if order.is_delivery and distance.kilometers > 25:
workflow.logger.error("Customer outside service area")
raise ApplicationError(
"Customer lives outside the service area",
type="CustomerOutsideServiceArea"
)
# Continue with order...
Workflow Tasks run in a thread pool despite async definition
Workflow Tasks run in threads on the workflow_task_executor thread pool because they are CPU bound, need to be timed out for deadlock detection, and need to not block other Workflow Tasks. The async keyword in Workflow Definitions refers to the Workflow's own deterministic event loop, not the standard asyncio event loop. Each Workflow gets its own event loop that cycles through during a Workflow Task to make progress on futures until no more progress can be made.
Example: HelloWorldAgentWorkflow with TemporalModel
```python
@workflow.defn
class HelloWorldAgentWorkflow:
@workflow.run
async def run(self, prompt: str) -> str:
agent = Agent(
name="hello_world_agent",
model=TemporalModel("gemini-2.5-flash"),
instruction="You only respond in haikus.",
)
runner = InMemoryRunner(agent=agent, app_name="hello_world_app")
session = await runner.session_service.create_session(
app_name="hello_world_app", user_id="user"
)
final_text = ""
async with Aclosing(
runner.run_async(
user_id="user",
session_id=session.id,
new_message=types.Content(role="user", parts=[types.Part(text=prompt)]),
)
) as agen:
async for event in agen:
if event.content and event.content.parts:
for part in event.content.parts:
if part.text:
final_text = part.text
return final_text
```
This example shows a basic ADK agent running in a Workflow using TemporalModel for each model call.
Example: Starting a Workflow with GoogleAdkPlugin
```python
client = await Client.connect(
os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
plugins=[GoogleAdkPlugin()],
)
result = await client.execute_workflow(
HelloWorldAgentWorkflow.run,
"Tell me about recursion in programming.",
id="google-adk-agents-basic-workflow-id",
task_queue="google-adk-agents-basic",
)
print(f"Result: {result}")
```
This example shows how to start a Workflow with the GoogleAdkPlugin to link the client-side code to the Workflow execution.
Python example: get and set current workflow details
```python
@workflow.defn
class YourWorkflow:
@workflow.run
async def run(self, input: str) -> str:
# Get the current details
current_details = workflow.get_current_details()
print(f"Current details: {current_details}")
# Set/update the current details
workflow.set_current_details("Updated workflow details with new status")
return "Workflow completed"
```
This example demonstrates retrieving and updating dynamic Workflow details during execution.
Get and set current workflow details from within a workflow
Inside a Workflow, you can call workflow.get_current_details() to retrieve the current Workflow details and workflow.set_current_details() to update them. Unlike static summary/details set at Workflow start, the current Workflow details can be updated throughout the life of the Workflow. Current Workflow details support Markdown format (excluding images, HTML, and scripts) and can span multiple lines.
Python example: start workflow with static summary and details
```python
handle = await client.start_workflow(
YourWorkflow.run,
"workflow input",
id="your-workflow-id",
task_queue="your-task-queue",
static_summary="Order processing for customer #12345",
static_details="Processing premium order with expedited shipping"
)
```
This example shows how to start a Workflow with contextual metadata that will appear in the Temporal UI.
Add static summary and details when starting a workflow
When starting a Workflow using client.start_workflow() or client.execute_workflow(), you can provide static_summary and static_details parameters. The static_summary is a single-line description limited to 200 bytes that appears in the Workflow list view. The static_details can be multi-line and is limited to 20K bytes, appearing in the Workflow details view. Both support standard Markdown formatting excluding images, HTML, and scripts.
Example Workflow execution from client
```python
import asyncio
import uuid
from temporalio.client import Client
async def main():
client = await Client.connect("localhost:7233")
result = await client.execute_workflow(
"SayHelloWorkflow",
"Temporal",
id=f"say-hello-workflow-{uuid.uuid4()}",
task_queue="my-task-queue",
)
print("Workflow result:", result)
if __name__ == "__main__":
asyncio.run(main())
```
This example shows how to execute a Workflow named SayHelloWorkflow with the argument "Temporal" and a unique execution ID.
Example Workflow definition in Python
```python
from datetime import timedelta
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from activities import greet
@workflow.defn
class SayHelloWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return await workflow.execute_activity(
greet,
name,
schedule_to_close_timeout=timedelta(seconds=10),
)
```
This example shows a Workflow that executes a greet Activity with a 10-second timeout.
Use workflow.unsafe.imports_passed_through() for imports in Workflows
When importing Activities or other modules into a Workflow, wrap the import statements with `with workflow.unsafe.imports_passed_through():` to ensure proper Temporal SDK integration.
Execute Activity from Workflow using workflow.execute_activity
Call Activities from within a Workflow using `await workflow.execute_activity()`. Pass the Activity function, its arguments, and optionally a schedule_to_close_timeout parameter to specify how long the Activity can take.
Execute Workflow from client
Execute a Workflow using `await client.execute_workflow()` with the Workflow name, arguments, a unique execution ID, and the task_queue. The method returns the Workflow result.
Define Workflow with @workflow.defn decorator
Define a Workflow by creating a class and decorating it with `@workflow.defn`. The Workflow class must have a method decorated with `@workflow.run` that contains the workflow logic. Workflows orchestrate Activities and contain the application logic. Temporal Workflows are resilient and can run for years even if the underlying infrastructure fails.
Workflow interceptors and replay safety
Workflow inbound and outbound interceptor methods also execute during replay. Use replay-safe APIs for logging, randomness, and time in these interceptors. If you want to write generic code shared by all inbound Workflow call handlers but want to skip read-only operations, check workflow.unsafe.is_read_only(). Activity and Client interceptors are not affected by replay.
Define a Workflow with @workflow.defn decorator
In the Python SDK, Workflows are defined as classes. Use the @workflow.defn decorator on the Workflow class to identify it as a Workflow. The decorator accepts an optional 'name' parameter to customize the Workflow Type name. If the name parameter is not specified, the Workflow name defaults to the unqualified class name.
Workflow parameters must be serializable
Temporal Workflows may have any number of custom parameters. All Workflow Definition parameters must be serializable. Temporal strongly recommends using a single dataclass parameter containing all input fields rather than multiple parameters, so that individual fields may be altered without breaking the Workflow signature.
Workflow return values must be serializable
Workflow return values must also be serializable. Use the return statement to return an object from the Workflow. To retrieve the results of a Workflow Execution, use either start_workflow() or execute_workflow() asynchronous methods.
Workflow logic constraints for deterministic execution
Workflow code must be deterministic because the Temporal Server may replay the Workflow to reconstruct its state. Workflow logic is constrained by deterministic execution requirements and must not include: threading, randomness, external calls to processes, network I/O, global state mutation, or system date or time calls. All API safe for Workflows must run in the implicit asyncio event loop and be deterministic.
Use workflow.logger for logging in Workflows
Use workflow.logger instead of print() or the standard logging module for logging in Workflows. The SDK logger automatically suppresses log messages during replay to avoid duplicates.
Use workflow.random() for deterministic random numbers
Use workflow.random() to get a deterministic random.Random instance seeded per Workflow Execution. Never use random.random() or other random module functions directly in Workflow code.
Use workflow.uuid4() for generating UUIDs in Workflows
For UUIDs in Workflows, use workflow.uuid4() instead of uuid.uuid4() to ensure deterministic behavior.
Use workflow.now() for current time in Workflows
Use workflow.now() instead of datetime.now() or time.time() to get the current time in Workflows. The SDK returns the time of the last Workflow Task, which is consistent across replays.
Use workflow.unsafe.is_replaying_history_events() for detecting new events
Use workflow.unsafe.is_replaying_history_events() to detect when new events are occurring. This will be false during read-only operations like queries and update validators. This is what the SDK's built-in logger and tracing interceptors use internally.
Pass through Activity modules for deterministic calls
For performance and behavior reasons, pass through all modules using imports_passed_through(), including Activities, Nexus services, and third-party plugins whose calls will be deterministic. This can be done with workflow.unsafe.imports_passed_through() context manager or at Worker creation time by customizing the runner's restrictions with with_passthrough_modules().
Basic Workflow example with Activity execution
This example shows a basic Workflow that executes an Activity. The Workflow is defined with @workflow.defn(name="YourWorkflow"), has an async run method decorated with @workflow.run, and executes an activity using workflow.execute_activity() with a start_to_close_timeout parameter:
```python
from datetime import timedelta
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from your_activities_dacx import your_activity
from your_dataobject_dacx import YourParams
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return await workflow.execute_activity(
your_activity,
YourParams("Hello", name),
start_to_close_timeout=timedelta(seconds=10),
)
```
Workflow constructor and message handler example
This example shows how to use @workflow.init to initialize the Workflow with input parameters, and how to use wait_condition for message handling. The __init__ and run methods must have the same parameters with the same type annotations:
```python
from dataclasses import dataclass
from temporalio import workflow
@dataclass
class MyWorkflowInput:
name: str
@workflow.defn
class WorkflowRunSeesWorkflowInitWorkflow:
@workflow.init
def __init__(self, workflow_input: MyWorkflowInput) -> None:
self.name_with_title = f"Knight {workflow_input.name}"
self.title_has_been_checked = False
@workflow.run
async def get_greeting(self, workflow_input: MyWorkflowInput) -> str:
await workflow.wait_condition(lambda: self.title_has_been_checked)
return f"Hello, {self.name_with_title}"
```
Workflow parameter dataclass example
This example shows how to define a dataclass for Workflow parameters:
```python
from dataclasses import dataclass
@dataclass
class YourParams:
greeting: str
name: str
```
Logging example in Workflow
This example shows how to use workflow.logger for logging in a Workflow:
```python
from temporalio import workflow
@workflow.defn
class MyWorkflow:
@workflow.run
async def run(self, name: str) -> str:
workflow.logger.info("Starting workflow", name)
# ...
```
Use @workflow.init for Workflow constructor initialization
Workflow constructors are useful if you have message handlers that need access to Workflow input. Use the @workflow.init decorator on the __init__ method to receive the same Workflow parameters as the @workflow.run method. The __init__ method and run method must have the same parameters with the same type annotations. The Workflow input arguments are passed to both methods.
Mark Workflow entry point with @workflow.run
Use the @workflow.run decorator to mark the entry point method to be invoked. This must be set on one asynchronous method defined on the same class as @workflow.defn. The run method has positional parameters that define the Workflow parameters.
Cancel a Workflow Execution in Python
To cancel a Workflow Execution in Python, use the cancel() function on the Workflow handle obtained from the client. Example: await client.get_workflow_handle("your_workflow_id").cancel().
Terminate a Workflow Execution in Python
To terminate a Workflow Execution in Python, use the terminate() function on the Workflow handle obtained from the client. Example: await client.get_workflow_handle("your_workflow_id").terminate().
Cancellation vs Termination in Workflows
Canceling a Workflow provides a graceful way to stop Workflow Execution, similar to sending SIGTERM to a process, and allows the Workflow to handle cancellation and execute cleanup logic. Terminating a Workflow forcefully stops it immediately, similar to killing a process, with no cleanup opportunity. In most cases, canceling is preferable; terminate only if the Workflow is stuck and cannot be canceled normally.
Python SDK workflow documentation structure
The Python SDK documentation covers workflows with the following main sections: Workflow basics, Child Workflows, Continue-As-New, Cancellation, Timeouts, Message passing, Schedules, Timers, Versioning, and Workflow Streams.
Pause scheduled workflow example in Python
import asyncio
from temporalio.client import Client
async def main():
client = await Client.connect("localhost:7233")
handle = client.get_schedule_handle(
"workflow-schedule-id",
)
await handle.pause(note="Pausing the schedule for now")
if __name__ == "__main__":
asyncio.run(main())
List all scheduled workflows in Python
To list all schedules in Python, use the `list_schedules()` asynchronous method on the Client. If a schedule is added or deleted, it may not be available in the list immediately.
Describe scheduled workflow example in Python
import asyncio
from temporalio.client import Client
async def main():
client = await Client.connect("localhost:7233")
handle = client.get_schedule_handle(
"workflow-schedule-id",
)
desc = await handle.describe()
print(f"Returns the note: {desc.schedule.state.note}")
if __name__ == "__main__":
asyncio.run(main())
Describe a scheduled workflow in Python
To describe a scheduled workflow in Python, use the `describe()` asynchronous method on the Schedule Handle. This returns current Schedule configuration including information about past, current, and future workflow runs. Use the ScheduleDescription class to access the complete list of attributes.
Delete scheduled workflow example in Python
import asyncio
from temporalio.client import Client
async def main():
client = await Client.connect("localhost:7233")
handle = client.get_schedule_handle(
"workflow-schedule-id",
)
await handle.delete()
if __name__ == "__main__":
asyncio.run(main())
Delete a scheduled workflow in Python
To delete a scheduled workflow in Python, use the `delete()` asynchronous method on the Schedule Handle. Deleting a Schedule does not affect any workflows that were already started by the Schedule.
Backfill scheduled workflow example in Python
import asyncio
from datetime import datetime, timedelta
from temporalio.client import Client, ScheduleBackfill, ScheduleOverlapPolicy
async def main():
client = await Client.connect("localhost:7233")
handle = client.get_schedule_handle(
"workflow-schedule-id",
)
now = datetime.utcnow()
(
await handle.backfill(
ScheduleBackfill(
start_at=now - timedelta(minutes=10),
end_at=now - timedelta(minutes=9),
overlap=ScheduleOverlapPolicy.ALLOW_ALL,
),
),
)
print(f"Result: {handle}")
if __name__ == "__main__":
asyncio.run(main())
Backfill a scheduled workflow in Python
To backfill a scheduled workflow in Python, use the `backfill()` asynchronous method on the Schedule Handle. The backfill action executes actions ahead of their specified time range, useful for executing missed or delayed actions or testing the workflow before its scheduled time.
Create scheduled workflow example in Python
import asyncio
from datetime import timedelta
from temporalio.client import (
Client,
Schedule,
ScheduleActionStartWorkflow,
ScheduleIntervalSpec,
ScheduleSpec,
ScheduleState,
)
from your_workflows import YourSchedulesWorkflow
async def main():
client = await Client.connect("localhost:7233")
await client.create_schedule(
"workflow-schedule-id",
Schedule(
action=ScheduleActionStartWorkflow(
YourSchedulesWorkflow.run,
"my schedule arg",
id="schedules-workflow-id",
task_queue="schedules-task-queue",
),
spec=ScheduleSpec(
intervals=[ScheduleIntervalSpec(every=timedelta(minutes=2))]
),
state=ScheduleState(note="Here's a note on my Schedule."),
),
)
if __name__ == "__main__":
asyncio.run(main())
Update a scheduled workflow in Python
To update a scheduled workflow in Python, use the `update()` method on the Schedule Handle. Pass a callback function that takes `ScheduleUpdateInput` and returns `ScheduleUpdate`. The callback receives the current schedule description and can modify it to build the update.
Create a scheduled workflow in Python
To create a scheduled workflow in Python, use the `create_schedule()` asynchronous method on the Client. Pass the Schedule ID and a Schedule object. Set the `action` parameter to `ScheduleActionStartWorkflow` to start a workflow execution. Optionally set the `spec` parameter to `ScheduleSpec` to specify the schedule, or the `intervals` parameter to `ScheduleIntervalSpec` to specify the interval. Other options include `cron_expressions`, `skip`, `start_at`, and `jitter`.
Update scheduled workflow example in Python
import asyncio
from temporalio.client import (
Client,
ScheduleActionStartWorkflow,
ScheduleUpdate,
ScheduleUpdateInput,
)
async def main():
client = await Client.connect("localhost:7233")
handle = client.get_schedule_handle(
"workflow-schedule-id",
)
async def update_schedule_simple(input: ScheduleUpdateInput) -> ScheduleUpdate:
schedule_action = input.description.schedule.action
if isinstance(schedule_action, ScheduleActionStartWorkflow):
schedule_action.args = ["my new schedule arg"]
return ScheduleUpdate(schedule=input.description.schedule)
await handle.update(update_schedule_simple)
if __name__ == "__main__":
asyncio.run(main())
Trigger scheduled workflow example in Python
import asyncio
from temporalio.client import Client
async def main():
client = await Client.connect("localhost:7233")
handle = client.get_schedule_handle(
"workflow-schedule-id",
)
await handle.trigger()
if __name__ == "__main__":
asyncio.run(main())
Trigger a scheduled workflow in Python
To trigger a scheduled workflow in Python, use the `trigger()` asynchronous method on the Schedule Handle. This triggers an immediate action with a given Schedule. By default, this action is subject to the Overlap Policy of the Schedule, useful for executing a workflow outside of its scheduled time.
Pause a scheduled workflow in Python
To pause a scheduled workflow in Python, use the `pause()` asynchronous method on the Schedule Handle. When you pause a Schedule, all future workflow runs associated with the Schedule are temporarily stopped. You can pass a `note` parameter to provide a reason for pausing the schedule.
List scheduled workflows example in Python
import asyncio
from temporalio.client import Client
async def main() -> None:
client = await Client.connect("localhost:7233")
async for schedule in await client.list_schedules():
print(f"List Schedule Info: {schedule.info}.")
if __name__ == "__main__":
asyncio.run(main())
Workflow failure only via explicit ApplicationError
You will only fail a Workflow by manually raising an Temporalio::Error::ApplicationError from Workflow code. Any other exceptions raised in a Workflow, including typical Ruby RuntimeErrors, will only fail that particular Workflow Task and be retried, as they are treated as bugs rather than reasons for the Workflow Execution to return as failed.
Difference between Activity and Workflow exception handling
In Activities, any Ruby exceptions or custom exceptions are converted to Temporal ApplicationError. In Workflows, exceptions other than an explicit Temporal ApplicationError will only fail that particular Workflow Task and be retried. This difference reflects that Activity exceptions should signal true failures, while non-ApplicationError exceptions in Workflows are treated as code bugs that can be corrected with redeployment.
Ruby SDK documentation structure and topics
The Ruby SDK documentation covers Workflows, Activities, Workers, Temporal Client, Platform observability, Integrations, and Best practices. Key topics include Workflow basics, child workflows, continue-as-new, cancellation, timeouts, message passing, schedules, timers, futures, dynamic workflows, and versioning. Activity topics include basics, execution, standalone activities, timeouts, asynchronous activity completion, dynamic activities, and benign exceptions. Workers documentation covers worker processes and observability. Platform features include observability and enriching the UI. Integrations include Rails integration. Best practices cover error handling, testing, debugging, and converters/encryption.
static_details parameter for starting workflows in Ruby
When starting or executing a workflow in the Ruby SDK using client.start_workflow() or client.execute_workflow(), you can provide a static_details parameter. This can be multi-line and provides more comprehensive information that appears in the Workflow details view in the Temporal UI, with a limit of 20K bytes. The format supports standard Markdown excluding images, HTML, and scripts.
static_summary parameter for starting workflows in Ruby
When starting or executing a workflow in the Ruby SDK using client.start_workflow() or client.execute_workflow(), you can provide a static_summary parameter. This is a single-line description that appears in the Workflow list view in the Temporal UI and is limited to 200 bytes.
Example: Get and set current workflow details in Ruby
require 'temporalio'
class YourWorkflow < Temporalio::Workflow::Definition
def execute(input)
# Get the current details
current_details = Temporalio::Workflow.current_details
Temporalio::Workflow.logger.info("Current details: #{current_details}")
# Set/update the current details
Temporalio::Workflow.current_details = 'Updated workflow details with new status'
'Workflow completed'
end
end