Cloud Run Handler - Starter Kit

Google Cloud Run is a serverless compute platform that lets you run containers without managing servers. It automatically scales up or down based on traffic, integrates with other GCP services, and supports deployments in multiple regions. Cloud Run abstracts away infrastructure management, allowing you to focus on your application logic.

Example

To create the GCPCloudRunHandler, pass Google service account credentials (either JSON file path or dict).

import os
from superagentx.handler.gcp.cloud_run import GCPCloudRunHandler

cloud_run_handler = GCPCloudRunHandler(
    creds=os.getenv("GOOGLE_APPLICATION_CREDENTIALS")  # Path to service account JSON
)

get_attr:
Safely fetches an attribute, dict key, or list index with a default fallback.

from superagentx.handler.gcp.cloud_run import get_attr
data = {"name": "cloudrun", "type": "service"}
print(get_attr(data, "name"))         # "cloudrun"
print(get_attr(data, "invalid", "N/A"))  # "N/A"

get_nested_attr:
Fetches nested attributes safely using a path list.

from superagentx.handler.gcp.cloud_run import get_nested_attr
data = {"spec": {"template": {"spec": {"service_account": "default"}}}}
print(get_nested_attr(data, ["spec", "template", "spec", "service_account"]))
# Output: "default"

client (property):
Lazy initialization of the async Cloud Run client.

client = cloud_run_handler.client
print(type(client))  # google.cloud.run_v2.ServicesAsyncClient

close:
Closes the async client to free resources.

await cloud_run_handler.close()

_process_container_info:
Extracts metadata for a Cloud Run container (image, ports, env vars, resources).

container_info = await cloud_run_handler._process_container_info(container_obj)
print(container_info["image"])

_process_service_info:
Processes a Cloud Run service object into a structured dictionary.

service_info = await cloud_run_handler._process_service_info(service_obj, "us-central1")
print(service_info["name"], service_info["url"])

_fetch_services_for_location:
Fetches all Cloud Run services in a specific region.

services = await cloud_run_handler._fetch_services_for_location("us-central1")
print([s["name"] for s in services])

fetch_location_with_semaphore:
Helper method to fetch services from a location with concurrency limits.

# Example inside async workflow
services = await cloud_run_handler._fetch_services_for_location("us-west1")

get_gcp_cloud_run_details:
Lists all Cloud Run services across all supported GCP regions with summary.

cloud_run_data = await cloud_run_handler.get_gcp_cloud_run_details()
print(cloud_run_data["summary"])

​Example

Example