Great question. Quick clarification before we dive in: do you want to run your code as a long-lived HTTP “Service” on Cloud Run, or do you want Prefect to execute your code as one-off jobs on Cloud Run? Prefect’s recommended path for observability/monitoring is to run flows as Cloud Run Jobs via a Cloud Run work pool. You’ll get full visibility in Prefect Cloud (states, logs, retries, etc.). If you truly need a long-running Service, I can outline how to run a Prefect worker as a Cloud Run Service too. Below are the recommended steps for running Prefect 3.x flows on Cloud Run Jobs with full Prefect Cloud observability. Overview - You’ll create a Cloud Run work pool in Prefect Cloud. - You’ll start a Prefect worker (runs anywhere with GCP credentials) that provisions Cloud Run Jobs for each flow run. - You’ll deploy your flow using flow.from_source(...).deploy(...). The worker injects ephemeral Prefect credentials so runs automatically report states/logs to Prefect Cloud. Prereqs - Prefect 3.x and GCP integration installed:

pip install "prefect[gcp]"

- GCP setup: enable Cloud Run API, Artifact Registry (or Container Registry), and authenticate:

gcloud auth login
  gcloud config set project <PROJECT_ID>

- Docker configured to push to Artifact Registry (or push images another way). - Prefect Cloud login:

prefect cloud login

1) Create a Cloud Run work pool (v2 API recommended) - Create the pool:

prefect work-pool create cr-pool --type cloud-run-v2

- Inspect the default job template (handy to see supported variables):

prefect work-pool get-default-base-job-template --type cloud-run-v2

You’ll see variables like region, service_account_name, cpu, memory, env, vpc_connector_name, timeout, etc. Docs: - Cloud Run worker guide: https://docs-3.prefect.io/integrations/prefect-gcp/gcp-worker-guide - Work pools/workers concepts: https://docs-3.prefect.io/v3/concepts/workers - Customize job variables: https://docs-3.prefect.io/v3/how-to-guides/deployments/customize-job-variables 2) Start a Prefect worker for that pool Run the worker anywhere with GCP credentials that can create/run Cloud Run Jobs (e.g., a small GCE VM). The worker will poll Prefect Cloud and submit jobs to Cloud Run:

uv run --with 'prefect[gcp]' prefect worker start -p cr-pool

Required permissions for the worker’s GCP identity typically include: - roles/run.admin (create/run Jobs) - roles/iam.serviceAccountUser (to use the run-as service account) - roles/artifactregistry.reader (pull images) You can also run the worker in Cloud Run as a Service if you want it fully serverless; set min instances ≥ 1 so it stays warm. I can share that pattern if that’s your goal. 3) Write your flow

from prefect import flow, get_run_logger

@flow(log_prints=True)
def my_flow(name: str = "world"):
    print("This will appear in Prefect logs")
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Hello, {name} from Cloud Run!")

4) Deploy your flow to the Cloud Run work pool Example using a GitHub repo as source and building a Docker image for Artifact Registry: ``` from prefect import flow @flow def my_flow(): ... if name == "__main__": my_flow.from_source( source="https://github.com/your-org/your-repo", entrypoint="flows/my_flow.py:my_flow", ).deploy( name="prod", work_pool_name="cr-pool", # Will be built and pushed automatically if build=True, push=True image="us-central1-docker.pkg.dev/your-proj/prefect/my-flow:{{ sha }}", build=True, push=True, job_variables={ "region": "us-central1", "service_account_name": "prefect-cr-runner@your-proj.iam.gserviceaccount.com", "cpu": "1", # e.g. "1", "2" "memory": "2Gi", # e.g. "512Mi", "2Gi" "timeout": 3600, # seconds "max_retries": 0, "env": {

# Your app secrets/config; Prefect credentials are injected automatically "MY_SETTING": "value" }, # Optional: # "vpc_connector_name": "projects/your-proj/locations/us-central1/connectors/your-connector", }, )

Notes:
- The worker injects ephemeral Prefect auth into the Cloud Run Job, so your flow automatically reports to Prefect Cloud (you don’t need to bake a Prefect API key into the image).
- If you prefer building images yourself, set build=False/push=False and reference the pre-built image.

5) Trigger a run and monitor in Prefect Cloud
- Trigger:

prefect deployment run "my_flow/prod" --watch ``` - Then open the run in Prefect Cloud to see states, logs, retries, runtime, and parameters. What about a Cloud Run “Service” instead of Jobs? - Recommended: Use Cloud Run Jobs for flow execution (above). - If you need a Service, a common pattern is to deploy a Prefect worker as a Cloud Run Service (entrypoint runs

prefect worker start -p cr-pool

) with min instances ≥ 1 and IAM to create/run Jobs. That keeps everything serverless while still using Cloud Run Jobs for executions and Prefect Cloud for observability. If you want this, I can share a Dockerfile and service YAML. Would you like me to tailor the commands to your exact project/region and share a minimal Dockerfile for the flow image and/or the worker-as-a-service pattern?

Got it — you want a fast-starting, always-on Cloud Run Service that’s integrated with Prefect Cloud. The best pattern is to run a Prefect worker as a Cloud Run Service (min instances ≥ 1), then execute your flows with a Process work pool. This gives sub‑second startup (no Cloud Run Job cold starts) and full Prefect Cloud observability. Here’s the end-to-end approach. High-level architecture - Create a Prefect work pool of type

process

. - Package and deploy a Cloud Run Service whose container starts: - a small HTTP server listening on $PORT (required by Cloud Run) - the Prefect worker that polls your process work pool - Deploy flows to that work pool using

flow.from_source(...).deploy(...)

(or bake code into the image). - The worker executes flow runs in-place inside the running container; logs/states stream to Prefect Cloud immediately. 1) Create a Prefect work pool (process)

prefect work-pool create cr-service-pool --type process

Helpful docs: - Work pools/workers: https://docs-3.prefect.io/v3/concepts/workers - Customize job variables (mostly N/A for process pools): https://docs-3.prefect.io/v3/how-to-guides/deployments/customize-job-variables 2) Build a worker container for Cloud Run Use a tiny HTTP server to satisfy Cloud Run’s $PORT contract and start the worker in the foreground. Dockerfile:

FROM python:3.11-slim

ENV PYTHONUNBUFFERED=1 \
    PORT=8080

WORKDIR /app

# Install Prefect + any deps your flows need (add git if pulling from_source)
RUN pip install --no-cache-dir "prefect[gcp]" requests

# Optionally: install your project dependencies
# COPY requirements.txt .
# RUN pip install --no-cache-dir -r requirements.txt

# A tiny HTTP server and worker launcher
COPY run.sh /app/run.sh
RUN chmod +x /app/run.sh

# If you want to keep code in the image instead of pulling from_source:
# COPY . /app

CMD ["/app/run.sh"]

run.sh:

#!/bin/sh
set -e

# Minimal HTTP server on $PORT for Cloud Run
python -c "import http.server, socketserver, os; port=int(os.getenv('PORT','8080')); http.server.ThreadingHTTPServer(('',port), http.server.SimpleHTTPRequestHandler).serve_forever()" &

# Start Prefect worker and point it at your process work pool
exec prefect worker start -p ${PREFECT_WORK_POOL:-cr-service-pool}

3) Deploy the Cloud Run Service - Build and push your image (replace region/project/repo as needed):

gcloud auth configure-docker us-central1-docker.pkg.dev
docker build -t us-central1-docker.pkg.dev/<PROJECT>/prefect/worker:latest .
docker push us-central1-docker.pkg.dev/<PROJECT>/prefect/worker:latest

- Deploy with min instances ≥ 1 and “CPU always allocated” enabled (to keep the worker running without traffic). In gcloud or console, set: - Min instances: 1 (or higher for redundancy) - CPU always allocated: enabled - Service account: one that can reach the Internet (Prefect Cloud + your code host) and access your resources Example deploy (fill in values; you can set these in the console as well):

gcloud run deploy prefect-worker \
  --image us-central1-docker.pkg.dev/<PROJECT>/prefect/worker:latest \
  --region us-central1 \
  --service-account prefect-worker@<PROJECT>.<http://iam.gserviceaccount.com|iam.gserviceaccount.com> \
  --min-instances 1 \
  --set-env-vars PREFECT_WORK_POOL=cr-service-pool \
  --set-env-vars PREFECT_API_URL=<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>> \
  --set-env-vars PREFECT_API_KEY=<YOUR_PREFECT_CLOUD_API_KEY> \
  --port 8080 \
  --no-allow-unauthenticated

Notes: - Store

PREFECT_API_KEY

in Secret Manager and mount it as an env var for security. - Ensure outbound Internet is allowed so the worker can reach Prefect Cloud and (if using from_source) your Git host. 4) Write and deploy your flow Using from_source so the worker pulls code at runtime (no rebuilds required): ``` from prefect import flow, get_run_logger @flow(log_prints=True) def my_flow(name: str = "world"): print("print -> shows up in Prefect")

get_run_logger().info(f"Hello, {name} from Cloud Run Service!") if name == "__main__": my_flow.from_source( source="https://github.com/your-org/your-repo", entrypoint="flows/my_flow.py:my_flow", ).deploy( name="prod", work_pool_name="cr-service-pool", build=False, push=False, # parameters={}, schedule=..., etc. as needed )

Trigger a run and monitor:

prefect deployment run "my_flow/prod" --watch ``` What you’ll get - Near-instant execution: no Cloud Run Job cold start; the worker is already warm. - Full observability in Prefect Cloud: states, logs, retries, durations, parameters. - Centralized ops: scale by increasing min instances and/or running multiple workers pointing to the same pool. Operational tips - Scale throughput: increase Cloud Run min instances or run multiple workers per instance (e.g., separate process with a different work queue). Also consider Prefect concurrency limits at the work pool or deployment level. - Networking: if using

from_source

, allow egress to your Git provider. If your code is private, set up git credentials via env vars/SSH keys. - Health checks: Cloud Run’s default health checks will hit your tiny HTTP server. If you want a richer health endpoint, swap the

http.server

snippet with a small FastAPI/Flask app. - IAM: The Cloud Run Service account only needs to reach external services (Prefect Cloud + your data infra). Unlike Job mode, it doesn’t need Run Admin because it’s not creating jobs. If you share your project/region and whether you want from_source or code baked into the image, I can tailor the Dockerfile and deploy command exactly to your setup.