# Serverless Workers on GCP Cloud Run - Python SDK

> For the complete documentation index, see [llms.txt](https://docs.temporal.io/llms.txt).
> Any documentation page is available as raw Markdown by appending `.md` to its URL.

> Run a Temporal Worker on a GCP Cloud Run worker pool using the Python SDK.

> **Public Preview**
> Cloud Run support is in Public Preview.

On a [GCP Cloud Run worker pool](https://cloud.google.com/run/docs/resource-model#worker-pools), you run a standard long-lived Temporal Worker.
Register Workflows and Activities the same way you would with any other Python Worker, and Temporal Cloud scales the pool up and down as work arrives and drains.

A Cloud Run Worker needs no Cloud Run-specific runtime or handler.
The one addition to a standard Worker is [Worker Versioning](/worker-versioning), which is required for Serverless Workers.

For the end-to-end deployment guide covering the Worker Pool, IAM, and compute configuration, see [Deploy a Serverless Worker on GCP Cloud Run](/production-deployment/worker-deployments/serverless-workers/cloud-run).

## Create a versioned Worker 

Build the Worker as you would any long-running Python Worker, then pass `deployment_config` to `Worker()` to declare the Worker Deployment Version and turn versioning on.

The following Worker reads its connection settings and Task Queue from the environment, so the same image can run against any Namespace:

```python
import asyncio
import os

from temporalio.client import Client
from temporalio.common import VersioningBehavior, WorkerDeploymentVersion
from temporalio.envconfig import ClientConfig
from temporalio.worker import Worker, WorkerDeploymentConfig

from my_activities import my_activity
from my_workflows import MyWorkflow

async def main() -> None:
    client = await Client.connect(**ClientConfig.load_client_connect_config())

    worker = Worker(
        client,
        task_queue=os.environ["TEMPORAL_TASK_QUEUE"],
        workflows=[MyWorkflow],
        activities=[my_activity],
        deployment_config=WorkerDeploymentConfig(
            version=WorkerDeploymentVersion(
                deployment_name="my-app",
                build_id="build-1",
            ),
            use_worker_versioning=True,
            default_versioning_behavior=VersioningBehavior.PINNED,
        ),
    )
    await worker.run()

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

`deployment_name` and `build_id` together identify the Worker Deployment Version. Both values must match the version you create with `temporal worker deployment create-version` in the deployment guide, or the Worker polls under a version the WCI does not manage.

Every Workflow needs a [versioning behavior](/worker-versioning#versioning-behaviors), either `PINNED` or `AUTO_UPGRADE`.
Setting `default_versioning_behavior` as shown above covers every Workflow on the Worker.
To set the behavior per Workflow instead, pass `versioning_behavior` to the `@workflow.defn` decorator:

```python
from temporalio import workflow
from temporalio.common import VersioningBehavior

@workflow.defn(versioning_behavior=VersioningBehavior.PINNED)
class MyWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        ...
```

For general Worker setup and options that are not specific to Cloud Run, see [Run a Worker](/develop/python/workers/run-worker-process).

## Configure the Temporal connection 

The `temporalio.envconfig` package loads [Temporal Client](/develop/python/client/temporal-client) configuration from environment variables and an optional TOML config file, so the Worker code carries no Namespace or credentials.
Set the non-secret values as environment variables on the Worker Pool, and mount the Temporal Cloud API key or TLS material from Secret Manager.
For the full list of supported variables, the config file format, and profiles, see [Environment configuration](/develop/environment-configuration).

`ClientConfig.load_client_connect_config()` returns the keyword arguments for `Client.connect`, which is why the Worker above unpacks it with `**`.
To inspect or change values before connecting, load the profile instead and convert it yourself:

```python
from temporalio.envconfig import ClientConfigProfile

profile = ClientConfigProfile.load()
connect_config = profile.to_client_connect_config()
client = await Client.connect(**connect_config)
```

## Keep Activities safe across scale-in 

The WCI decides when to remove an instance from Task Queue activity, not from what an individual instance is doing.
An instance running a long Activity can be stopped mid-execution.

Use [Activity Heartbeats](/develop/python/activities/timeouts#activity-heartbeats) so a retry resumes from the last recorded progress instead of starting over:

```python
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"
```

For how scale-in decisions are made, see [Serverless Workers on GCP Cloud Run](/serverless-workers/cloud-run#lifecycle).

## Configure OpenTelemetry 

Run an OpenTelemetry Collector as a sidecar in the Worker Pool. The Cloud Run OpenTelemetry plugin exports metrics and traces by OTLP gRPC to the Collector at `localhost:4317`. It configures the endpoint, service name, Core SDK metrics, and tracer provider from Cloud Run defaults.

Create the plugin and pass it to `Client.connect`. The sample also registers `CloudRunIdPlugin` on the same Client and
reads its connection settings from the environment:

<!--SNIPSTART python-cloud-run {"selectedLines": ["1-10"]}-->
[gcp/cloud_run/worker.py](https://github.com/temporalio/samples-python/blob/gcp-cloud-run/gcp/cloud_run/worker.py)
```py
# OpenTelemetryPlugin exports Core metrics and traces to the collector
# sidecar; CloudRunIdPlugin sets the client identity from instance metadata.
plugin = OpenTelemetryPlugin(add_temporal_spans=True)
client = await Client.connect(
    settings.address,
    namespace=settings.namespace,
    api_key=settings.api_key,
    tls=settings.tls,
    plugins=[plugin, CloudRunIdPlugin()],
)
```
<!--SNIPEND-->

Configure the Collector sidecar to receive OTLP gRPC on `localhost:4317`, export traces to Google Cloud, and export metrics to Google Managed Service for Prometheus:

<!--SNIPSTART python-cloud-run-collector-config-->
[gcp/cloud_run/collector-config.yaml](https://github.com/temporalio/samples-python/blob/gcp-cloud-run/gcp/cloud_run/collector-config.yaml)
```yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: localhost:4317

processors:
  batch/traces:
    send_batch_max_size: 200
    send_batch_size: 200
    timeout: 5s
  memory_limiter:
    check_interval: 1s
    limit_percentage: 65
    spike_limit_percentage: 20
  resource_detection:
    detectors: [gcp]
    timeout: 10s
  # Rename Temporal datapoint labels that collide with the target labels Managed Service for Prometheus injects (e.g. namespace).
  transform/collision:
    metric_statements:
      - context: datapoint
        statements:
          - set(attributes["exported_location"], attributes["location"])
          - delete_key(attributes, "location")
          - set(attributes["exported_cluster"], attributes["cluster"])
          - delete_key(attributes, "cluster")
          - set(attributes["exported_namespace"], attributes["namespace"])
          - delete_key(attributes, "namespace")
          - set(attributes["exported_job"], attributes["job"])
          - delete_key(attributes, "job")
          - set(attributes["exported_instance"], attributes["instance"])
          - delete_key(attributes, "instance")
          - set(attributes["exported_project_id"], attributes["project_id"])
          - delete_key(attributes, "project_id")
  # The Telemetry API expects the Google Cloud project in gcp.project_id.
  transform/set_project_id:
    error_mode: ignore
    trace_statements:
      - set(resource.attributes["gcp.project_id"], resource.attributes["gcp.project.id"]) where resource.attributes["gcp.project.id"] != nil
      - set(resource.attributes["gcp.project_id"], resource.attributes["cloud.account.id"]) where resource.attributes["gcp.project_id"] == nil and resource.attributes["cloud.account.id"] != nil

exporters:
  googlemanagedprometheus:
  otlp_grpc:
    endpoint: telemetry.googleapis.com:443
    compression: none
    balancer_name: pick_first
    auth:
      authenticator: googleclientauth

extensions:
  googleclientauth:
  health_check:
    endpoint: 0.0.0.0:13133

service:
  extensions: [googleclientauth, health_check]
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, resource_detection, transform/collision]
      exporters: [googlemanagedprometheus]
    traces:
      receivers: [otlp]
      processors:
        [memory_limiter, resource_detection, transform/set_project_id, batch/traces]
      exporters: [otlp_grpc]
```
<!--SNIPEND-->

Wait until the Collector accepts connections before starting the Worker. On shutdown, call `plugin.shutdown()` after the Worker stops so traces can flush. For a Collector configuration that exports traces to Google Cloud and metrics to Google Managed Service for Prometheus, see the [Python Cloud Run sample](https://github.com/temporalio/samples-python/tree/gcp-cloud-run/gcp/cloud_run).

## Set a Worker identity 

Use `CloudRunIdPlugin` to identify each Worker instance as `<instance-id>@<revision>`. It reads Cloud Run environment variables and instance metadata when
the Client connects, then Workers created from that Client inherit the identity.

Register the plugin when connecting the Client:

```python
from temporalio.client import Client
from temporalio.contrib.gcp.cloud_run.id import CloudRunIdPlugin

from gcp.cloud_run.settings import load_settings

# ...
    settings = load_settings()
    client = await Client.connect(
        settings.address,
        namespace=settings.namespace,
        api_key=settings.api_key,
        tls=settings.tls,
        plugins=[CloudRunIdPlugin()],
    )
    # ...
```

The [preceding example](#opentelemetry) registers `CloudRunIdPlugin` alongside the OpenTelemetry plugin.

The ID plugin requires the Cloud Run metadata server, so it fails when the Worker runs outside Cloud Run. For a complete
example, see the [Python Cloud Run sample](https://github.com/temporalio/samples-python/tree/gcp-cloud-run/gcp/cloud_run).