activities/execution

235 notes in this subject, read out of this brain and free to use. This is page 3 of 4.

Activity finalizer callback in PHP

If you need to clean up resources after an activity completes, register a finalizer callback using $worker->registerActivityFinalizer(). This callback is called after each activity invocation. Example: $worker->registerActivityFinalizer(fn() => $kernel->shutdown());

execute_activity() example with timeout

Example: 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))

start_activity() vs execute_activity() for getting Activity results

Use start_activity() to start an Activity and return its ActivityHandle. Use execute_activity() to return the results. The call to spawn an Activity Execution generates the ScheduleActivityTask Command and provides the Workflow with an Awaitable. Workflow Executions can either block progress until the result is available through the Awaitable or continue progressing and make use of the result when it becomes available.

execute_activity() spawns Activity Execution from Workflow

The execute_activity() operation is used within a Workflow Definition to spawn an Activity Execution. It is a shortcut for start_activity() that waits on its result. In most cases, execute_activity() should be used unless advanced task capabilities are needed.

start_activity() returns ActivityHandle for separate waiting and cancellation

To get just the handle to wait and cancel separately from spawning an Activity, use start_activity() instead of execute_activity().

Required Activity Timeouts: Schedule-To-Close or Start-To-Close

Activity Execution semantics require that either a Schedule-To-Close Timeout or a Start-To-Close Timeout must be set. These values are set in the Activity Options as keyword arguments after the Activity arguments.

Available Activity timeout options

The available Activity timeout options that can be set as keyword arguments are: schedule_to_close_timeout, schedule_to_start_timeout, and start_to_close_timeout.

Count Standalone Activities with client.count_activities()

Use client.count_activities() to count Standalone Activity Executions matching a List Filter query. Returns the total count of executions (running, completed, failed, etc.) - not the number of queued tasks. Works the same way as counting Workflow Executions. Query parameter accepts List Filter syntax.

List Standalone Activities with client.list_activities()

Use client.list_activities() to list Standalone Activity Executions matching a List Filter query. Returns an async iterator yielding ActivityExecution entries. Only returns Standalone Activity Executions; Activities running inside Workflows are not included. Query parameter accepts List Filter syntax like "TaskQueue = 'my-task-queue'" or "ActivityType = 'MyActivity' AND Status = 'Running'".

Standalone Activities list activities example

Example of listing Standalone Activities: Use client.list_activities(query="TaskQueue = 'my-standalone-activity-task-queue'") then iterate with async for info in activities, accessing info.activity_id, info.activity_type, and info.status fields.

Get handle to existing Standalone Activity with client.get_activity_handle()

Use client.get_activity_handle() to create a handle to a previously started Standalone Activity by providing activity_id and run_id parameters. The handle can be used to wait for the result, describe, cancel, or terminate the Activity.

Start Standalone Activity without waiting for result with client.start_activity()

Use client.start_activity() to send a request to the Temporal Server to durably enqueue an Activity job without waiting for execution. Returns an activity handle that can be used later to get the result, describe, cancel, or terminate the Activity.

Execute a Standalone Activity with client.execute_activity()

Use client.execute_activity() to execute a Standalone Activity. Call this from application code, not from inside a Workflow Definition. This durably enqueues the Standalone Activity in the Temporal Server, waits for it to be executed on a Worker, and then fetches the result. Parameters include the activity function, args list, id, task_queue, and start_to_close_timeout.

Standalone Activities count activities example

Example of counting Standalone Activities: Use await client.count_activities(query="TaskQueue = 'my-standalone-activity-task-queue'") which returns resp with resp.count for total count and resp.groups list containing group objects with group_values and count.

Activity Heartbeat implementation in Python

To Heartbeat an Activity Execution in Python, use the activity.heartbeat() API. Heartbeats can accept detail data that persists on the server for retrieval during Activity retry. If an Activity calls heartbeat with detail arguments and then fails and is retried, heartbeat_details returns an iterable containing those details on the next run.

Activity heartbeat example

Example showing how to use activity.heartbeat() in Python: ```python from temporalio import activity @activity.defn async def your_activity_definition() -> str: activity.heartbeat("heartbeat details!") ```

execute_activity versus start_activity in Python

execute_activity() is a shortcut for start_activity() that waits on its result. To get just a handle to wait and cancel separately, use start_activity(). execute_activity() should be used in most cases unless advanced task capabilities are needed.

Activity timeout options in Python

Activity timeouts are set as keyword arguments to execute_activity() or start_activity(). The available timeout parameters are: schedule_to_close_timeout, schedule_to_start_timeout, and start_to_close_timeout. All accept timedelta values.

Heartbeat Timeout configuration

heartbeat_timeout is a parameter for start_activity() and execute_activity() that sets the maximum time between Activity Heartbeats. It accepts a timedelta value. Activities that do not Heartbeat cannot receive Cancellations. Heartbeat throttling may lead to Cancellations being delivered later than expected.

Custom retry policy example in Python

from temporalio import workflow from temporalio.common import RetryPolicy from datetime import timedelta @workflow.defn class OrderWorkflow: @workflow.run async def run(self, order): # Custom retry for rate-limited service retry_policy = RetryPolicy( initial_interval=timedelta(seconds=10), backoff_coefficient=3.0, maximum_interval=timedelta(minutes=5), maximum_attempts=20, ) result = await workflow.execute_activity( call_external_service, order, start_to_close_timeout=timedelta(seconds=30), retry_policy=retry_policy, ) return result

Match retry policy to failure types

For transient failures (brief network issues): use defaults or a low initial_interval and backoff_coefficient. For intermittent failures (rate limiting): increase initial_interval and backoff_coefficient to space out retries and let the condition resolve. For cost-sensitive APIs: set maximum_attempts to limit retries, though it's usually better to use timeouts.

Multiple retry policies for same activity example

fast_retry = RetryPolicy( initial_interval=timedelta(seconds=1), backoff_coefficient=1.5, ) slow_retry = RetryPolicy( initial_interval=timedelta(seconds=30), backoff_coefficient=3.0, ) # Same Activity, different policies await workflow.execute_activity( process_order, order, start_to_close_timeout=timedelta(seconds=10), retry_policy=fast_retry, ) # Later, with different circumstances... await workflow.execute_activity( process_order, order, start_to_close_timeout=timedelta(seconds=10), retry_policy=slow_retry, )

Non-retryable error types in Retry Policy

List error types that shouldn't retry in your Retry Policy by setting non_retryable_error_types. When an Activity raises an ApplicationError, Temporal checks if its type is in non_retryable_error_types. If it matches, the Activity fails immediately without retries.

Non-retryable error types Retry Policy example

from temporalio import workflow from temporalio.common import RetryPolicy from datetime import timedelta @workflow.defn class CheckoutWorkflow: @workflow.run async def run(self, payment_details): retry_policy = RetryPolicy( non_retryable_error_types=[ "InvalidCardFormat", "InsufficientFunds", "AccountClosed", ] ) try: result = await workflow.execute_activity( process_payment, payment_details, start_to_close_timeout=timedelta(seconds=30), retry_policy=retry_policy, ) return result except ActivityError as e: workflow.logger.error(f"Payment failed: {e.cause}") # Handle the non-retryable error...

RetryPolicy attributes

Retry Policy has these attributes: initial_interval (Delay before first retry, default: 1 second), backoff_coefficient (Multiplier for subsequent delays, default: 2.0), maximum_interval (Cap on retry delay, default: 100× initial interval), maximum_attempts (Maximum retry attempts, default: unlimited), non_retryable_error_types (Error types that shouldn't retry, default: empty).

Example: Exposing Activities as tools with activity_tool

```python weather_tool = temporalio.contrib.google_adk_agents.workflow.activity_tool( get_weather, start_to_close_timeout=timedelta(seconds=60) ) agent = Agent( name="weather_agent", model=TemporalModel("gemini-2.5-flash"), instruction="Use the get_weather tool to answer weather questions.", tools=[weather_tool], ) ``` This example shows wrapping a Temporal Activity as an ADK tool with a timeout.

activity_tool wraps Temporal Activities to expose them as ADK tools

The activity_tool function wraps a Temporal Activity so it can be used as a tool in an ADK Agent. When the model calls the tool, it runs as its own Activity with its own retries, timeouts, and appears in the Event history. It accepts the activity function and parameters like start_to_close_timeout.

Activity schedule_to_close_timeout parameter

The schedule_to_close_timeout parameter in workflow.execute_activity() specifies how long an Activity task has from when it is scheduled until it must complete. This is specified as a timedelta object.

Cloud Run Activity Heartbeat example for scale-in safety

from temporalio import activity @activity.defn async def my_activity(items: list[str]) -> str: for i, item in enumerate(items): activity.heartbeat(i) # ... process item return "done"

Cloud Run scale-in safety requires Activity Heartbeats

The WCI removes instances from Cloud Run based on Task Queue activity, not based on what individual instances are doing. An instance running a long Activity can be stopped mid-execution. Use Activity Heartbeats so that a retry resumes from the last recorded progress instead of starting over.

disable_eager_activity_execution always True in Lambda workers

disable_eager_activity_execution is always True in Lambda workers and cannot be overridden. Eager Activities require a persistent connection, which Lambda invocations do not maintain.

Tune timeouts for long-running activities in Lambda workers

If your Worker handles long-running Activities, increase graceful_shutdown_timeout, shutdown_deadline_buffer, and the Lambda invocation deadline (--timeout) together.

Activity execution in Ruby SDK

Activities are executed within a Workflow Definition using the `execute_activity` operation. The call generates a ScheduleActivityTask Command and produces three Activity Task related Events in the Workflow Execution Event History: ActivityTaskScheduled, ActivityTaskStarted, and ActivityTask[Closed].

Activity invocation parameters and return values impact history size

The values passed to Activities through invocation parameters or returned through a result value are recorded in the Execution history. The entire Execution history is transferred from the Temporal service to Workflow Workers when a Workflow state needs to recover. A large Execution history can adversely impact the performance of your Workflow. Be mindful of the amount of data you transfer through Activity invocation parameters or Return Values.

Ruby SDK execute_activity syntax and required parameters

To spawn an Activity Execution in Ruby, use the `execute_activity` operation from within your Workflow Definition. Activity Execution semantics rely on several parameters. The only required value that needs to be set is either a Schedule-To-Close Timeout or a Start-To-Close Timeout. These values are set as keyword parameters. The Activity result is returned from the `execute_activity` call.

Ruby SDK execute_activity example

class MyWorkflow < Temporalio::Workflow::Definition workflow_name :MyDifferentWorkflowName def execute(name) Temporalio::Workflow.execute_activity( MyActivity, { greeting: 'Hello', name: }, start_to_close_timeout: 100 ) end end This example shows how to execute an Activity from a Workflow Definition, passing parameters as a hash and specifying the start_to_close_timeout.

start_activity example for Standalone Activity

Example showing how to start a Standalone Activity without waiting for result: ```ruby handle = client.start_activity( StandaloneActivity::MyActivities::ComposeGreeting, 'Hello', 'World', id: 'standalone-activity-id', task_queue: 'standalone-activity-sample', start_to_close_timeout: 10 ) puts "Started Activity with id=#{handle.id} run_id=#{handle.run_id}" # Wait for the result later puts "Activity result: #{handle.result}" ```

Execute Standalone Activity with client.execute_activity

Use Temporalio::Client#execute_activity to execute a Standalone Activity and block until it completes. Call this from your application code, not from inside a Workflow Definition. This durably enqueues your Standalone Activity in the Temporal Server, waits for it to be executed on your Worker, and then returns the result. The first argument is the Activity to run (can be the Activity::Definition subclass, an instance of one, or a string/symbol name). Positional arguments after it are passed to the Activity's execute method. The call requires id, task_queue, and at least one of start_to_close_timeout or schedule_to_close_timeout.

Start Standalone Activity without waiting

Use Temporalio::Client#start_activity to start a Standalone Activity and get a handle without waiting for the result. This durably enqueues your Activity job in the Temporal Server without waiting for it to be executed by your Worker. The handle returned provides access to the Activity ID and run ID, and can be used to wait for the result later by calling handle.result.

Get handle to existing Standalone Activity

Use Temporalio::Client#activity_handle to create an ActivityHandle for a previously started Standalone Activity. Pass the activity ID as the first argument. Pass no run ID (the default) to target the latest run of the given Activity ID, or pass activity_run_id: to target a specific run. The handle can be used to wait for the result with handle.result, fetch metadata with handle.describe, request cancellation with handle.cancel, or force-close with handle.terminate.

Wait for Standalone Activity result

Call handle.result on an ActivityHandle to block until the Activity completes and return the result. Calling client.execute_activity is equivalent to calling client.start_activity to durably enqueue the Standalone Activity, then calling handle.result to wait for completion.

List Standalone Activities with client.list_activities

Use Temporalio::Client#list_activities to list Standalone Activity Executions that match a List Filter query. The result is an Enumerator of ActivityExecution values that fetches pages from the server on demand as the enumerator is consumed. These APIs return only Standalone Activity Executions; Activities running inside Workflows are not included. The query parameter accepts the same List Filter syntax used for Workflow Visibility, such as "TaskQueue = 'standalone-activity-sample'" or "ActivityType = 'ComposeGreeting' AND Status = 'Running'".

Count Standalone Activities with client.count_activities

Use Temporalio::Client#count_activities to count Standalone Activity Executions that match a List Filter query. This returns the total count of executions (running, completed, failed, etc.) — not the number of queued tasks. It works the same way as counting Workflow Executions. The result includes a count property and a groups property for grouped counts.

execute_activity example for Standalone Activity

Example showing how to execute a Standalone Activity and block for result: ```ruby result = client.execute_activity( StandaloneActivity::MyActivities::ComposeGreeting, 'Hello', 'World', id: 'standalone-activity-id', task_queue: 'standalone-activity-sample', start_to_close_timeout: 10 ) puts "Activity result: #{result}" ```

list_activities example for Standalone Activities

Example showing how to list Standalone Activity Executions: ```ruby client.list_activities("TaskQueue = 'standalone-activity-sample'").each do |execution| puts "#{execution.activity_id} #{execution.activity_type} #{execution.status}" end ```

count_activities example for Standalone Activities

Example showing how to count Standalone Activity Executions: ```ruby result = client.count_activities("TaskQueue = 'standalone-activity-sample'") puts "Total: #{result.count}" result.groups.each do |group| puts " #{group.group_values.join(',')} => #{group.count}" end ```

Heartbeat Timeout in Ruby SDK

The Heartbeat Timeout sets the maximum duration between Heartbeats before the Temporal Service considers the Activity failed. Set this using the heartbeat_timeout parameter when executing an activity, specified in seconds.

Send heartbeat in Ruby activity

Send a heartbeat in a Ruby activity by calling Temporalio::Activity::Context.current.heartbeat within the activity execution. This should be done periodically in loops or during long-running operations.

Activity heartbeat in Ruby SDK

A Heartbeat is a periodic signal from the Worker to the Temporal Service indicating the Activity is still alive and making progress. Heartbeats are used to detect Worker failure, deliver cancellations, and may contain custom progress details.

Activity timeout types in Ruby SDK

Each Activity Timeout controls a different aspect of how long an Activity Execution can take. The three types are: Schedule-To-Close Timeout, Start-To-Close Timeout, and Schedule-To-Start Timeout. At least one of start_to_close_timeout or schedule_to_close_timeout is required.

Execute activity with start_to_close_timeout in Ruby

In Ruby SDK, execute an activity with start_to_close_timeout using Temporalio::Workflow.execute_activity. Pass the activity class, arguments hash, and start_to_close_timeout parameter in seconds.

execute_activity options for schedule_to_close_timeout

The schedule_to_close_timeout option in Temporalio::Workflow.execute_activity specifies the maximum time allowed for an activity execution, specified in seconds. Example: Temporalio::Workflow.execute_activity(SayHelloActivity, name, schedule_to_close_timeout: 300) sets a 300 second timeout.

Activity timeout field names in Rust

The timeout configuration fields for Activity options in Rust are named: schedule_to_close_timeout, schedule_to_start_timeout, and start_to_close_timeout.

Configure Activity Start-To-Close Timeout in Rust

In Rust, set the Start-To-Close Timeout using ActivityOptions::start_to_close_timeout(Duration::from_secs(30)).

Activity heartbeat in Rust

An Activity Heartbeat is a signal from the Worker Process executing the Activity to the Temporal Service, indicating the Activity Execution is still making progress and the Worker has not crashed. If the service does not receive a heartbeat within the configured Heartbeat Timeout, the Activity can time out and be retried according to its Retry Policy. Heartbeats may be throttled by the Worker. Activity cancellation is delivered through heartbeat processing, so Activities that do not heartbeat cannot receive cancellation promptly. Heartbeats can include details describing current progress, which can be retrieved on retry.

Record heartbeat in Rust Activity

To heartbeat an Activity in Rust, call ctx.record_heartbeat() from inside the Activity method, passing details as a vector of values describing progress.

Heartbeat Timeout in Rust

A Heartbeat Timeout sets the maximum time allowed between heartbeats. Configure it as part of Activity options when scheduling the Activity using .heartbeat_timeout(Duration::from_secs(5)).

Activity timeouts available in Rust SDK

Three timeout types are available for Activity options in Rust: Schedule-To-Close Timeout (maximum time for overall Activity Execution), Start-To-Close Timeout (maximum time for a single Activity Task Execution), and Schedule-To-Start Timeout (maximum time from scheduling to Worker pickup, non-retryable by design). An Activity Execution must have either Start-To-Close Timeout or Schedule-To-Close Timeout set. Temporal strongly recommends setting Start-To-Close Timeout because the service relies on it to detect lost Activity Tasks and trigger retries.

Executing activities from workflows in Rust

Activities are executed using ctx.start_activity() within a workflow's #[run] method. The method takes the activity function reference, input parameters, and ActivityOptions. Example: ctx.start_activity(MyActivities::greet, name, ActivityOptions::start_to_close_timeout(Duration::from_secs(30))).await?

Cancellable Activity implementation example in Rust

Example of a cancellable Activity in Rust: ```rust #![allow(unreachable_pub)] use temporalio_macros::{activities}; use temporalio_sdk::{ activities::{ActivityContext, ActivityError}, }; pub struct CancellationActivities; #[activities] impl CancellationActivities { #[activity] pub async fn long_running_cancellable_activity( ctx: ActivityContext, _input: (), ) -> Result<String, ActivityError> { loop { if ctx.is_cancelled() { return Err(ActivityError::cancelled()); } ctx.record_heartbeat(vec![]); tokio::time::sleep(std::time::Duration::from_millis(200)).await; } } #[activity] pub async fn cleanup(_ctx: ActivityContext, _input: ()) -> Result<String, ActivityError> { Ok("cleanup done".to_string()) } } ``` This example shows a long-running Activity that periodically checks if it has been cancelled using ctx.is_cancelled(), records heartbeats, and sleeps. It returns a cancellation error when cancelled. A separate cleanup Activity can be executed after cancellation.

Give your agent this brain