{"id":502,"library":"google-cloud-tasks","title":"Google Cloud Tasks Python Client","description":"Cloud Tasks is a fully managed service for dispatching and delivering a large number of distributed tasks. The `google-cloud-tasks` Python client library, currently at version 2.22.0, provides an idiomatic way to interact with the Cloud Tasks API for creating, managing, and scheduling tasks. Google Cloud client libraries typically receive updates on an as-needed basis, often following changes in the underlying Google Cloud APIs.","status":"active","version":"2.22.0","language":"python","source_language":"en","source_url":"https://github.com/googleapis/google-cloud-python/tree/main/packages/google-cloud-tasks","tags":["google-cloud","gcp","tasks","queue","asynchronous","serverless"],"install":[{"cmd":"pip install google-cloud-tasks","lang":"bash","label":"Install latest version"}],"dependencies":[{"reason":"Requires Python 3.9 or higher for current versions.","package":"Python","optional":false},{"reason":"Core Google API client library functionalities.","package":"google-api-core","optional":false},{"reason":"Used for serializing and deserializing API messages.","package":"protobuf","optional":false}],"imports":[{"note":"The primary client for interacting with Cloud Tasks. The `_v2` suffix indicates the API version.","symbol":"CloudTasksClient","correct":"from google.cloud import tasks_v2"},{"note":"Represents a unit of scheduled work.","symbol":"Task","correct":"from google.cloud.tasks_v2.types import Task"},{"note":"Used to configure an HTTP target for a task.","symbol":"HttpRequest","correct":"from google.cloud.tasks_v2.types import HttpRequest"},{"note":"Enum for specifying the HTTP method of a task's request.","symbol":"HttpMethod","correct":"from google.cloud.tasks_v2.types import HttpMethod"},{"note":"Used for authenticating tasks to Cloud Run or other OIDC-enabled services.","symbol":"OidcToken","correct":"from google.cloud.tasks_v2.types import OidcToken"},{"note":"Required for setting `schedule_time` for tasks using protobuf timestamps.","symbol":"timestamp_pb2","correct":"from google.protobuf import timestamp_pb2"}],"quickstart":{"code":"import os\nimport json\nimport datetime\nfrom google.cloud import tasks_v2\nfrom google.cloud.tasks_v2.types import Task, HttpRequest, HttpMethod\nfrom google.protobuf import timestamp_pb2\n\ndef create_http_task_with_auth(\n    project_id: str,\n    location_id: str,\n    queue_id: str,\n    url: str,\n    service_account_email: str,\n    payload: dict = None,\n    schedule_time_seconds: int = None,\n) -> Task:\n    \"\"\"Create an HTTP task with OIDC authentication.\n\n    Args:\n        project_id: The GCP project ID.\n        location_id: The location of the queue (e.g., 'us-central1').\n        queue_id: The ID of the Cloud Tasks queue.\n        url: The full URL of the HTTP target (e.g., a Cloud Run service).\n        service_account_email: The email of the service account for OIDC token.\n        payload: (Optional) Dictionary payload to send with the task.\n        schedule_time_seconds: (Optional) Delay in seconds before task execution.\n\n    Returns:\n        The created Task object.\n    \"\"\"\n    client = tasks_v2.CloudTasksClient()\n\n    # Construct the queue path.\n    parent = client.queue_path(project_id, location_id, queue_id)\n\n    # Construct the HTTP request for the task.\n    http_request = HttpRequest(\n        http_method=HttpMethod.POST,\n        url=url,\n        oidc_token={'service_account_email': service_account_email}\n    )\n\n    if payload:\n        http_request.headers['Content-Type'] = 'application/json'\n        http_request.body = json.dumps(payload).encode('utf-8')\n\n    # Construct the task.\n    task = Task(http_request=http_request)\n\n    if schedule_time_seconds:\n        # Schedule the task for a future time.\n        future_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=schedule_time_seconds)\n        timestamp = timestamp_pb2.Timestamp()\n        timestamp.FromDatetime(future_time)\n        task.schedule_time = timestamp\n\n    # Send the task creation request.\n    response = client.create_task(parent=parent, task=task)\n    print(f\"Created task: {response.name}\")\n    return response\n\n# --- Example Usage ---\nif __name__ == \"__main__\":\n    # Set environment variables for authentication (e.g., GOOGLE_APPLICATION_CREDENTIALS)\n    # and for the task details.\n    project = os.environ.get('GCP_PROJECT_ID', 'your-gcp-project-id')\n    location = os.environ.get('GCP_LOCATION_ID', 'us-central1')\n    queue = os.environ.get('GCP_TASK_QUEUE_ID', 'my-http-queue')\n    target_url = os.environ.get('TASK_TARGET_URL', 'https://your-cloud-run-service-url.run.app/task-handler')\n    sa_email = os.environ.get('TASK_SERVICE_ACCOUNT_EMAIL', 'your-service-account@your-gcp-project-id.iam.gserviceaccount.com')\n\n    if project == 'your-gcp-project-id':\n        print(\"Please set GCP_PROJECT_ID, GCP_LOCATION_ID, GCP_TASK_QUEUE_ID, TASK_TARGET_URL, and TASK_SERVICE_ACCOUNT_EMAIL environment variables.\")\n    else:\n        task_payload = {\"message\": \"Hello from Cloud Tasks!\", \"source\": \"python-client\"}\n        try:\n            created_task = create_http_task_with_auth(\n                project,\n                location,\n                queue,\n                target_url,\n                sa_email,\n                payload=task_payload,\n                schedule_time_seconds=60 # Schedule 1 minute in the future\n            )\n            print(f\"Successfully scheduled task: {created_task.name}\")\n        except Exception as e:\n            print(f\"Error creating task: {e}\")\n            print(\"Ensure Cloud Tasks API is enabled, queue exists, and authentication is set up.\")","lang":"python","description":"This quickstart demonstrates how to create an HTTP task with OIDC authentication, suitable for targeting services like Cloud Run or Cloud Functions that require identity verification. It includes setting a payload and an optional schedule time. Ensure your Google Cloud environment is configured for Application Default Credentials (ADC) or explicitly set `GOOGLE_APPLICATION_CREDENTIALS` for local development."},"warnings":[{"fix":"Set `GOOGLE_APPLICATION_CREDENTIALS` locally. Grant `roles/cloudtasks.enqueuer` to the caller for the queue, and appropriate invocation roles (e.g., `roles/run.invoker`) to the service account specified in `oidc_token.service_account_email` if using authenticated HTTP targets.","message":"Authentication is crucial. For local development, set the `GOOGLE_APPLICATION_CREDENTIALS` environment variable to point to a service account key file. For deployed applications, ensure the service account running your code has the `Cloud Tasks Enqueuer` role (or `cloudtasks.tasks.create` permission) for the queue, and if targeting authenticated endpoints (like Cloud Run), the task's OIDC token's service account must have permissions to invoke the target.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Design your task handlers to be idempotent. This means the outcome of processing a task multiple times is the same as processing it once. Techniques include using unique transaction IDs or checking state before applying changes.","message":"Cloud Tasks aims for exactly-once delivery but can occasionally execute tasks more than once due to system trade-offs. Your task handlers should be idempotent to safely handle duplicate executions without unintended side effects.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Configure appropriate rate limits and concurrency settings on your Cloud Tasks queues. Ensure your target services (e.g., Cloud Run, App Engine) are scaled sufficiently to handle the expected task dispatch rate. Implement client-side rate limiting or exponential backoff when enqueueing tasks if you frequently hit `ResourceExhausted` errors.","message":"Overwhelming your target service or improperly configuring queue rate limits can lead to backlogs and HTTP 429 (Too Many Requests) or 503 (Service Unavailable) errors. Cloud Tasks will implement backoff, but sustained issues require intervention.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Always encode the `body` field of an `HttpRequest` object to bytes using `.encode('utf-8')` (or similar) before sending.","message":"When sending a payload with an HTTP task, the body must be explicitly encoded to bytes (e.g., `json.dumps(payload).encode('utf-8')`). Failing to encode will result in type errors.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Upgrade your Python environment to 3.9 or higher. For older Python versions, you might need to use an older version of the `google-cloud-tasks` library, but this is not recommended.","message":"The client library requires Python 3.9 or newer. Older Python versions (3.8 and below) are unsupported.","severity":"breaking","affected_versions":"<= 2.21.0 (earlier versions supported Python >= 3.7)"},{"fix":"Add `import logging; logging.getLogger(\"google\").propagate = True` to your application's initialization code if you want Google library logs to be handled by the root logger.","message":"Default logging for `google-cloud-tasks` (and other `google` prefixed libraries) does not propagate to the root logger by default. If you expect to see logs via your application's root logger, you must explicitly enable propagation.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Set all required environment variables: `GCP_PROJECT_ID`, `GCP_LOCATION_ID`, `GCP_TASK_QUEUE_ID`, `TASK_TARGET_URL`, and `TASK_SERVICE_ACCOUNT_EMAIL` with appropriate values for your Cloud Tasks setup.","message":"Critical environment variables for GCP project, location, task queue, target URL, and service account email (`GCP_PROJECT_ID`, `GCP_LOCATION_ID`, `GCP_TASK_QUEUE_ID`, `TASK_TARGET_URL`, `TASK_SERVICE_ACCOUNT_EMAIL`) must be configured. The application cannot proceed without these settings.","severity":"breaking","affected_versions":"All versions"},{"fix":"Ensure that `GCP_PROJECT_ID`, `GCP_LOCATION_ID`, `GCP_TASK_QUEUE_ID`, `TASK_TARGET_URL`, and `TASK_SERVICE_ACCOUNT_EMAIL` environment variables are properly defined and accessible by your application.","message":"The Cloud Tasks client library, especially when used in examples or local development, requires specific environment variables to be set for project, location, queue, target URL, and service account email. Failing to set these will prevent basic operations or even application startup.","severity":"breaking","affected_versions":"All versions"}],"env_vars":null,"last_verified":"2026-05-12T14:25:55.926Z","next_check":"2026-06-26T00:00:00.000Z","problems":[{"fix":"Grant the `Cloud Tasks Enqueuer` role (`roles/cloudtasks.enqueuer`) to the service account or user. For HTTP target tasks, ensure the service account also has `appengine.applications.get` if it's an App Engine Flex or Standard environment.","cause":"The service account or user making the API call lacks the necessary IAM permissions to create tasks in the specified Cloud Tasks queue.","error":"google.api_core.exceptions.PermissionDenied: 403 The principal does not have cloudtasks.tasks.create permission for the resource queue."},{"fix":"Carefully review the task and `http_request` configuration. Ensure `schedule_time` is set for a future timestamp, the `body` for HTTP targets is correctly encoded (e.g., base64 for `bytes`), and all other parameters conform to the Cloud Tasks API's requirements for the chosen task type.","cause":"The task configuration provided to `create_task` is malformed or contains invalid values, such as a `schedule_time` in the past, an incorrectly formatted service account email in an OIDC token, or an invalid HTTP request body.","error":"google.api_core.exceptions.InvalidArgument: 400 Request contains an invalid argument."},{"fix":"Install the specific `google-cloud-tasks` package using `pip install google-cloud-tasks`. If the generic `google-cloud` package is installed, uninstall it first (`pip uninstall google-cloud`) as it is deprecated and can interfere with specific client libraries. Ensure you are installing into the correct virtual environment.","cause":"The `google-cloud-tasks` Python client library is not installed in the active Python environment, or there is an issue with the Python path. An older, deprecated `google-cloud` meta-package might also cause import conflicts.","error":"ModuleNotFoundError: No module named 'google.cloud.tasks'"},{"fix":"Implement client-side retry logic using `google.api_core.retry.Retry` with a suitable predicate (e.g., `if_transient_error`) and appropriate exponential backoff delays. Check the Google Cloud status dashboard for any ongoing service outages. Also, ensure your task's target endpoint is healthy, accessible, and can process requests within the configured timeout.","cause":"This error typically indicates a transient issue with the Cloud Tasks service itself, network connectivity problems, or the target endpoint for the task is unresponsive or overloaded, leading to a timeout or service unavailability.","error":"grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNAVAILABLE"}],"ecosystem":"pypi","meta_description":null,"install_score":100,"install_tag":"verified","quickstart_score":80,"quickstart_tag":"verified","pypi_latest":null,"install_checks":{"last_tested":"2026-05-12","tag":"verified","tag_description":"installs cleanly on critical runtimes, fast import, recently tested","results":[{"runtime":"python:3.10-alpine","python_version":"3.10","os_libc":"alpine (musl)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":1.66,"mem_mb":23.6,"disk_size":"70.4M"},{"runtime":"python:3.10-slim","python_version":"3.10","os_libc":"slim (glibc)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":1.01,"mem_mb":20.4,"disk_size":"68M"},{"runtime":"python:3.11-alpine","python_version":"3.11","os_libc":"alpine (musl)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":2.42,"mem_mb":25.6,"disk_size":"75.4M"},{"runtime":"python:3.11-slim","python_version":"3.11","os_libc":"slim (glibc)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":1.57,"mem_mb":22.6,"disk_size":"73M"},{"runtime":"python:3.12-alpine","python_version":"3.12","os_libc":"alpine (musl)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":2.56,"mem_mb":24.9,"disk_size":"66.8M"},{"runtime":"python:3.12-slim","python_version":"3.12","os_libc":"slim (glibc)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":1.99,"mem_mb":21.9,"disk_size":"65M"},{"runtime":"python:3.13-alpine","python_version":"3.13","os_libc":"alpine (musl)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":2.43,"mem_mb":25.8,"disk_size":"66.3M"},{"runtime":"python:3.13-slim","python_version":"3.13","os_libc":"slim (glibc)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":1.93,"mem_mb":22.8,"disk_size":"64M"},{"runtime":"python:3.9-alpine","python_version":"3.9","os_libc":"alpine (musl)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":1.5,"mem_mb":23.4,"disk_size":"70.6M"},{"runtime":"python:3.9-slim","python_version":"3.9","os_libc":"slim (glibc)","variant":"default","exit_code":0,"wheel_type":null,"failure_reason":null,"install_time_s":null,"import_time_s":1.18,"mem_mb":20.4,"disk_size":"68M"}]},"quickstart_checks":{"last_tested":"2026-04-23","tag":"verified","tag_description":"quickstart runs on critical runtimes, recently tested","results":[{"runtime":"python:3.10-alpine","exit_code":0},{"runtime":"python:3.10-slim","exit_code":0},{"runtime":"python:3.11-alpine","exit_code":0},{"runtime":"python:3.11-slim","exit_code":0},{"runtime":"python:3.12-alpine","exit_code":0},{"runtime":"python:3.12-slim","exit_code":0},{"runtime":"python:3.13-alpine","exit_code":0},{"runtime":"python:3.13-slim","exit_code":0},{"runtime":"python:3.9-alpine","exit_code":0},{"runtime":"python:3.9-slim","exit_code":0}]}}