Google Cloud Workflows × Gemini API Production Orchestration Guide: Timeouts, Retries, and Cost Control

# Enable required APIs
gcloud services enable workflows.googleapis.com
gcloud services enable cloudscheduler.googleapis.com
gcloud services enable aiplatform.googleapis.com
 
# Create a service account for workflow execution
gcloud iam service-accounts create gemini-workflow-runner \
    --display-name="Gemini Workflow Runner"
 
SA_EMAIL="gemini-workflow-runner@YOUR_PROJECT_ID.iam.gserviceaccount.com"
 
# Grant necessary permissions
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/aiplatform.user"
 
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/workflows.invoker"
 
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/storage.objectViewer"

# workflow-gemini-summarize.yaml
main:
  params: [args]
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - location: "us-central1"
          - model: "gemini-2.5-pro"
          - input_text: ${args.text}
 
    - call_gemini_api:
        try:
          call: http.post
          args:
            url: ${"https://" + location + "-aiplatform.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model + ":generateContent"}
            auth:
              type: OAuth2
            headers:
              Content-Type: "application/json"
            body:
              contents:
                - role: "user"
                  parts:
                    - text: ${"Summarize the following text in 100 words or less:\n\n" + input_text}
              generationConfig:
                maxOutputTokens: 300
                temperature: 0.3
          result: gemini_response
        retry:
          predicate: ${http_retry_predicate}
          max_retries: 3
          backoff:
            initial_delay: 2
            max_delay: 60
            multiplier: 2
        except:
          as: e
          steps:
            - handle_api_error:
                raise: ${e}
 
    - extract_text:
        assign:
          - summary: ${gemini_response.body.candidates[0].content.parts[0].text}
 
    - return_result:
        return:
          summary: ${summary}
          input_tokens: ${gemini_response.body.usageMetadata.promptTokenCount}
          output_tokens: ${gemini_response.body.usageMetadata.candidatesTokenCount}
 
# Retry predicate: retry on rate limit (429) or temporary unavailability (503)
http_retry_predicate:
  params: [e]
  steps:
    - check_status:
        switch:
          - condition: ${e.code == 429 or e.code == 503}
            return: true
        return: false

# Deploy the workflow
gcloud workflows deploy gemini-summarize \
    --source=workflow-gemini-summarize.yaml \
    --location=us-central1 \
    --service-account="${SA_EMAIL}"
 
# Run a test execution
gcloud workflows run gemini-summarize \
    --location=us-central1 \
    --data='{"text": "Google Cloud Workflows is a serverless orchestration service that lets you define and execute multi-step processes involving HTTP-based services. Each step'\''s state is automatically preserved, enabling reliable execution of complex pipelines."}'

{
  "summary": "Google Cloud Workflows is a serverless service for orchestrating HTTP-based multi-step processes, automatically preserving each step's state for reliable execution.",
  "input_tokens": 68,
  "output_tokens": 29
}

# workflow-content-pipeline.yaml
main:
  params: [args]
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - location: "us-central1"
          - model: "gemini-2.5-pro"
          - topic: ${args.topic}
          - target_length: ${default(map.get(args, "target_length"), 2000)}
          - bucket: ${default(map.get(args, "output_bucket"), "gemini-workflow-outputs")}
          - run_id: ${sys.now()}
 
    # Step 1: Generate article outline
    - generate_outline:
        call: googleapis.aiplatform.v1.projects.locations.publishers.models.generateContent
        args:
          model: ${"projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model}
          body:
            contents:
              - role: "user"
                parts:
                  - text: |
                      Create an article outline for the topic: ${topic}
                      Target length: ~${target_length} words.
                      Return 4–6 H2 headings as a JSON array:
                      ["Heading 1", "Heading 2", ...]
            generationConfig:
              responseMimeType: "application/json"
              maxOutputTokens: 300
        result: outline_response
 
    - parse_outline:
        assign:
          - outline_json: ${outline_response.candidates[0].content.parts[0].text}
          - headings: ${json.decode(outline_json)}
 
    # Step 2: Generate section content in parallel
    - generate_sections:
        parallel:
          for:
            value: heading
            in: ${headings}
            steps:
              - generate_section_content:
                  call: http.post
                  args:
                    url: ${"https://" + location + "-aiplatform.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model + ":generateContent"}
                    auth:
                      type: OAuth2
                    timeout: 300
                    headers:
                      Content-Type: "application/json"
                    body:
                      contents:
                        - role: "user"
                          parts:
                            - text: |
                                Write the "${heading}" section of an article about ${topic}.
                                300–400 words, engaging and informative tone.
                      generationConfig:
                        maxOutputTokens: 700
                        temperature: 0.7
                  result: section_response
 
    # Step 3: Quality check — detect hallucinations and contradictions
    - quality_check:
        call: http.post
        args:
          url: ${"https://" + location + "-aiplatform.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model + ":generateContent"}
          auth:
            type: OAuth2
          timeout: 120
          headers:
            Content-Type: "application/json"
          body:
            contents:
              - role: "user"
                parts:
                  - text: |
                      Review the following content about "${topic}" for obvious factual errors
                      or internal contradictions. Reply "OK" if none found, or
                      "ISSUES: [description]" if problems are detected.
            generationConfig:
              maxOutputTokens: 200
              temperature: 0
          result: quality_response
 
    - check_quality_result:
        assign:
          - quality_text: ${quality_response.body.candidates[0].content.parts[0].text}
 
    - validate_quality:
        switch:
          - condition: ${text.substring(quality_text, 0, 6) == "ISSUES"}
            steps:
              - quality_failed:
                  raise:
                    code: 400
                    message: ${"Quality check failed: " + quality_text}
 
    # Step 4: Save output to Cloud Storage
    - save_to_storage:
        call: http.post
        args:
          url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + bucket + "/o?uploadType=media&name=articles/" + run_id + ".json"}
          auth:
            type: OAuth2
          headers:
            Content-Type: "application/json"
          body:
            topic: ${topic}
            headings: ${headings}
            quality_status: "OK"
            generated_at: ${run_id}
        result: storage_response
 
    - return_success:
        return:
          status: "SUCCESS"
          storage_path: ${"gs://" + bucket + "/articles/" + run_id + ".json"}
          headings_count: ${list.length(headings)}

# ❌ When using native googleapis connector:
# result.candidates[0].content.parts[0].text
 
# ✅ When using http.post:
# result.body.candidates[0].content.parts[0].text  ← note .body

- init_results:
    assign:
      - results: []
 
- parallel_generate:
    parallel:
      shared: [results]  # ← Declare shared variables explicitly
      for:
        value: item
        in: ${items}
        steps:
          - process:
              call: http.post
              # ...
              result: item_result
          - collect:
              assign:
                - results: ${list.concat(results, [item_result.body])}

# Write large content immediately to GCS
- save_large_content:
    call: http.post
    args:
      url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + bucket + "/o?uploadType=media&name=temp/" + item_id + ".txt"}
      auth:
        type: OAuth2
      body: ${large_content}
    result: gcs_result
 
# Pass only the URL to subsequent steps
- store_url_only:
    assign:
      - content_url: ${"gs://" + bucket + "/temp/" + item_id + ".txt"}

- call_gemini_with_timeout:
    call: http.post
    args:
      url: ${gemini_endpoint}
      timeout: 300  # 5 minutes — adjust based on your use case
      auth:
        type: OAuth2
      body:
        # ...

# Create a Cloud Scheduler job (runs daily at 6 AM JST = 9 PM UTC previous day)
gcloud scheduler jobs create http gemini-content-daily \
    --location=asia-northeast1 \
    --schedule="0 21 * * *" \
    --uri="https://workflowexecutions.googleapis.com/v1/projects/YOUR_PROJECT_ID/locations/us-central1/workflows/content-pipeline/executions" \
    --message-body='{"argument": "{\"topic\": \"Gemini API production best practices\", \"target_length\": 2000}"}' \
    --oauth-service-account-email="${SA_EMAIL}" \
    --oauth-token-scope="https://www.googleapis.com/auth/cloud-platform"

import json
import time
from google.cloud import workflows_v1
from google.cloud.workflows import executions_v1
 
def run_gemini_workflow(
    project_id: str,
    location: str,
    workflow_name: str,
    topic: str,
    target_length: int = 2000,
    poll_interval: int = 10,
    max_wait_seconds: int = 600
) -> dict:
    """
    Start a Cloud Workflow execution and wait for completion.
    
    Args:
        project_id: GCP project ID
        location: Workflow region (e.g., us-central1)
        workflow_name: Name of the deployed workflow
        topic: Content generation topic
        target_length: Target word count
        poll_interval: Seconds between status checks
        max_wait_seconds: Maximum time to wait before raising TimeoutError
    
    Returns:
        Workflow execution result as a dict
    
    Raises:
        TimeoutError: If the workflow doesn't complete within max_wait_seconds
        RuntimeError: If the workflow fails or is cancelled
    """
    exec_client = executions_v1.ExecutionsClient()
    parent = exec_client.workflow_path(project_id, location, workflow_name)
    
    # Start the workflow execution
    execution = exec_client.create_execution(
        request=executions_v1.CreateExecutionRequest(
            parent=parent,
            execution=executions_v1.Execution(
                argument=json.dumps({
                    "topic": topic,
                    "target_length": target_length,
                    "output_bucket": f"{project_id}-gemini-outputs"
                })
            )
        )
    )
    
    execution_name = execution.name
    print(f"Workflow started: {execution_name}")
    
    # Poll until completion
    elapsed = 0
    while elapsed < max_wait_seconds:
        execution = exec_client.get_execution(
            request=executions_v1.GetExecutionRequest(name=execution_name)
        )
        
        state = execution.state
        
        if state == executions_v1.Execution.State.SUCCEEDED:
            result = json.loads(execution.result)
            print(f"✅ Workflow succeeded: {result}")
            return result
            
        elif state == executions_v1.Execution.State.FAILED:
            error = execution.error
            raise RuntimeError(
                f"Workflow failed: {error.message} (HTTP status: {error.http_status_code})"
            )
            
        elif state == executions_v1.Execution.State.CANCELLED:
            raise RuntimeError("Workflow was cancelled")
        
        print(f"  Still running... ({elapsed}s elapsed)")
        time.sleep(poll_interval)
        elapsed += poll_interval
    
    raise TimeoutError(
        f"Workflow did not complete within {max_wait_seconds}s. "
        f"Execution: {execution_name}"
    )
 
if __name__ == "__main__":
    try:
        result = run_gemini_workflow(
            project_id="your-project-id",
            location="us-central1",
            workflow_name="content-pipeline",
            topic="Practical Gemini 2.5 Pro use cases for indie developers",
            target_length=2000
        )
        print(f"Output saved to: {result['storage_path']}")
    except RuntimeError as e:
        print(f"Error: {e}")
    except TimeoutError as e:
        print(f"Timeout: {e}")

from google.cloud import billing_budgets_v1
 
def create_gemini_budget_alert(
    billing_account_id: str,
    project_id: str,
    monthly_budget_usd: float,
    alert_thresholds: list[float] = [0.5, 0.8, 1.0]
) -> str:
    """
    Create a monthly budget alert for Gemini API (Vertex AI) spend.
    
    Args:
        billing_account_id: Billing account ID (e.g., 012345-ABCDEF-789012)
        project_id: GCP project to monitor
        monthly_budget_usd: Monthly budget cap in USD
        alert_thresholds: Alert firing thresholds (0.0 to 1.0)
    
    Returns:
        Resource name of the created budget
    """
    client = billing_budgets_v1.BudgetServiceClient()
    
    budget = billing_budgets_v1.Budget(
        display_name=f"Gemini API Budget - {project_id}",
        budget_filter=billing_budgets_v1.Filter(
            projects=[f"projects/{project_id}"],
            services=["services/F3B6-3D8E-295F"],  # Vertex AI / AI Platform
            credit_types_treatment=(
                billing_budgets_v1.Filter.CreditTypesTreatment.EXCLUDE_ALL_CREDITS
            )
        ),
        amount=billing_budgets_v1.BudgetAmount(
            specified_amount={
                "currency_code": "USD",
                "units": int(monthly_budget_usd),
                "nanos": int((monthly_budget_usd % 1) * 1e9)
            }
        ),
        threshold_rules=[
            billing_budgets_v1.ThresholdRule(
                threshold_percent=t,
                spend_basis=billing_budgets_v1.ThresholdRule.Basis.CURRENT_SPEND
            )
            for t in alert_thresholds
        ]
    )
    
    created = client.create_budget(
        parent=f"billingAccounts/{billing_account_id}",
        budget=budget
    )
    
    print(f"Budget alert created: {created.name}")
    print(f"Monthly cap: ${monthly_budget_usd} USD")
    print(f"Alert thresholds: {[f'{int(t*100)}%' for t in alert_thresholds]}")
    return created.name
 
# Example: $50/month budget with alerts at 50%, 80%, and 100%
create_gemini_budget_alert(
    billing_account_id="012345-ABCDEF-789012",
    project_id="my-gemini-project",
    monthly_budget_usd=50.0
)

gcloud alpha monitoring policies create \
    --policy-from-file=- << 'EOF'
{
  "displayName": "Cloud Workflow Failure Alert",
  "conditions": [
    {
      "displayName": "Workflow execution failure rate > 0",
      "conditionThreshold": {
        "filter": "resource.type=\"workflows.googleapis.com/Workflow\" AND metric.type=\"workflows.googleapis.com/finished_execution_count\" AND metric.labels.status=\"FAILED\"",
        "comparison": "COMPARISON_GT",
        "thresholdValue": 0,
        "duration": "60s",
        "aggregations": [
          {
            "alignmentPeriod": "300s",
            "perSeriesAligner": "ALIGN_SUM"
          }
        ]
      }
    }
  ]
}
EOF

# workflow-with-subworkflows.yaml
main:
  params: [args]
  steps:
    - generate_title:
        call: call_gemini
        args:
          prompt: ${"Write a compelling article title for the topic: " + args.topic}
          max_tokens: 100
          temperature: 0.8
        result: title_result
 
    - generate_intro:
        call: call_gemini
        args:
          prompt: ${"Write an engaging 150-word introduction for: " + title_result.text}
          max_tokens: 300
          temperature: 0.7
        result: intro_result
 
    - return_content:
        return:
          title: ${title_result.text}
          intro: ${intro_result.text}
          total_tokens: ${title_result.tokens + intro_result.tokens}
 
# Reusable subworkflow for Gemini API calls
call_gemini:
  params: [prompt, max_tokens, temperature]
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - location: "us-central1"
          - model: "gemini-2.5-pro"
 
    - api_call:
        try:
          call: http.post
          args:
            url: ${"https://" + location + "-aiplatform.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/publishers/google/models/" + model + ":generateContent"}
            auth:
              type: OAuth2
            timeout: 300
            headers:
              Content-Type: "application/json"
            body:
              contents:
                - role: "user"
                  parts:
                    - text: ${prompt}
              generationConfig:
                maxOutputTokens: ${max_tokens}
                temperature: ${temperature}
          result: response
        retry:
          predicate: ${retry_on_transient_error}
          max_retries: 3
          backoff:
            initial_delay: 2
            max_delay: 60
            multiplier: 2
        except:
          as: e
          steps:
            - log_and_raise:
                call: sys.log
                args:
                  text: ${"Gemini API call failed: " + string(e)}
                  severity: ERROR
            - re_raise:
                raise: ${e}
 
    - extract:
        return:
          text: ${response.body.candidates[0].content.parts[0].text}
          tokens: ${response.body.usageMetadata.totalTokenCount}
 
# Retry predicate subworkflow
retry_on_transient_error:
  params: [e]
  steps:
    - check:
        switch:
          - condition: ${e.code == 429 or e.code == 503 or e.code == 500}
            return: true
        return: false

from google.cloud import secretmanager
import json
 
def get_workflow_secrets(project_id: str, secret_name: str) -> dict:
    """
    Retrieve workflow configuration from Secret Manager.
    Secrets should be JSON-encoded key-value pairs.
    
    Args:
        project_id: GCP project ID
        secret_name: Name of the secret in Secret Manager
    
    Returns:
        dict of secret values
    """
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{project_id}/secrets/{secret_name}/versions/latest"
    
    response = client.access_secret_version(request={"name": name})
    payload = response.payload.data.decode("UTF-8")
    
    return json.loads(payload)
 
def run_workflow_with_secrets(project_id: str, workflow_name: str) -> dict:
    """
    Retrieve configuration from Secret Manager, then run the workflow.
    This prevents sensitive data from appearing in Cloud Scheduler payloads.
    """
    from google.cloud.workflows import executions_v1
    import time
    
    # Get configuration from Secret Manager
    config = get_workflow_secrets(project_id, "gemini-workflow-config")
    
    exec_client = executions_v1.ExecutionsClient()
    parent = exec_client.workflow_path(project_id, "us-central1", workflow_name)
    
    execution = exec_client.create_execution(
        request=executions_v1.CreateExecutionRequest(
            parent=parent,
            execution=executions_v1.Execution(
                argument=json.dumps({
                    "topic": config["daily_topic"],
                    "output_bucket": config["output_bucket"],
                    "notification_url": config["notification_webhook"]
                })
            )
        )
    )
    
    print(f"Workflow started: {execution.name}")
    return {"execution_name": execution.name}

# List all roles granted to the workflow service account
gcloud projects get-iam-policy YOUR_PROJECT_ID \
    --filter="bindings.members:gemini-workflow-runner@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
    --format="table(bindings.role)"
 
# Remove any roles that aren't actively required
gcloud projects remove-iam-policy-binding YOUR_PROJECT_ID \
    --member="serviceAccount:gemini-workflow-runner@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/ROLE_TO_REMOVE"

from google.cloud import bigquery
 
def analyze_workflow_performance(
    project_id: str,
    days: int = 30,
    workflow_name: str = "content-pipeline"
) -> None:
    """
    Summarize Cloud Workflow execution performance and estimated costs.
    
    Args:
        project_id: GCP project ID
        days: Number of days of history to analyze
        workflow_name: Name of the workflow to analyze
    """
    client = bigquery.Client()
    
    query = f"""
    SELECT
        DATE(timestamp) as date,
        COUNT(*) as total_executions,
        COUNTIF(JSON_VALUE(jsonPayload.state) = 'SUCCEEDED') as succeeded,
        COUNTIF(JSON_VALUE(jsonPayload.state) = 'FAILED') as failed,
        SAFE_DIVIDE(
            COUNTIF(JSON_VALUE(jsonPayload.state) = 'FAILED'),
            COUNT(*)
        ) * 100 as failure_rate_pct,
        -- Token usage aggregated from Gemini API response metadata
        SUM(
            CAST(JSON_VALUE(jsonPayload.metadata.totalTokenCount) AS INT64)
        ) as total_tokens,
        -- Estimated cost at Gemini 2.5 Pro pricing ($7/1M input, $21/1M output)
        SUM(
            CAST(JSON_VALUE(jsonPayload.metadata.promptTokenCount) AS INT64)
        ) / 1000000 * 7 +
        SUM(
            CAST(JSON_VALUE(jsonPayload.metadata.candidatesTokenCount) AS INT64)
        ) / 1000000 * 21 as estimated_cost_usd
    FROM `{project_id}.global._Default._AllLogs`
    WHERE
        resource.type = 'workflows.googleapis.com/Workflow'
        AND resource.labels.workflow_id = '{workflow_name}'
        AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {days} DAY)
    GROUP BY date
    ORDER BY date DESC
    """
    
    results = client.query(query)
    
    print(f"Workflow Performance: {workflow_name} (last {days} days)")
    print(f"{'Date':<12} {'Total':>7} {'OK':>5} {'Fail':>5} {'Fail%':>7} {'Tokens':>12} {'Est. Cost':>12}")
    print("-" * 70)
    
    total_cost = 0.0
    for row in results:
        tokens = row.total_tokens or 0
        cost = row.estimated_cost_usd or 0.0
        total_cost += cost
        fail_rate = row.failure_rate_pct or 0.0
        
        # Flag rows with >10% failure rate
        flag = " ⚠️" if fail_rate > 10 else ""
        
        print(
            f"{str(row.date):<12}"
            f"{row.total_executions:>7}"
            f"{row.succeeded:>5}"
            f"{row.failed:>5}"
            f"{fail_rate:>6.1f}%"
            f"{tokens:>12,}"
            f"  ${cost:>9.4f}{flag}"
        )
    
    print(f"\nTotal estimated cost over {days} days: ${total_cost:.4f} USD")

# Pattern: checkpoint outputs before risky steps
main:
  params: [args]
  steps:
    - init:
        assign:
          - checkpoint_bucket: "my-workflow-checkpoints"
          - run_id: ${sys.now()}
          - completed_steps: []
 
    - step_1_outline:
        # ... generate outline ...
        steps:
          - generate:
              # call Gemini
              result: outline_result
          - checkpoint:
              call: http.post
              args:
                url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + checkpoint_bucket + "/o?uploadType=media&name=checkpoints/" + run_id + "/outline.json"}
                auth:
                  type: OAuth2
                body: ${outline_result}
          - mark_complete:
              assign:
                - completed_steps: ${list.concat(completed_steps, ["outline"])}
 
    # If a later step fails, the checkpoint lets you resume from here
    - step_2_content:
        # ... with similar checkpoint pattern ...
        steps:
          - generate_content:
              # call Gemini for content
              result: content_result
          - checkpoint_content:
              call: http.post
              args:
                url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + checkpoint_bucket + "/o?uploadType=media&name=checkpoints/" + run_id + "/content.json"}
                auth:
                  type: OAuth2
                body: ${content_result}
 
    - final_return:
        return:
          run_id: ${run_id}
          completed_steps: ${completed_steps}
          checkpoint_path: ${"gs://" + checkpoint_bucket + "/checkpoints/" + run_id + "/"}

# Check for active executions before deploying an update
gcloud workflows executions list gemini-content-pipeline \
    --location=us-central1 \
    --filter="state=ACTIVE" \
    --format="table(name, startTime, state)"
 
# Deploy the updated workflow only when no active executions remain
gcloud workflows deploy gemini-content-pipeline \
    --source=workflow-content-pipeline-v2.yaml \
    --location=us-central1 \
    --service-account="${SA_EMAIL}"
 
# Verify the new version is active
gcloud workflows describe gemini-content-pipeline \
    --location=us-central1 \
    --format="table(name, revisionId, updateTime)"

# Cost estimation utility for planning workflow budgets
def estimate_monthly_cost(
    executions_per_day: int,
    steps_per_execution: int,
    avg_input_tokens: int,
    avg_output_tokens: int,
    model: str = "gemini-2.5-pro"
) -> dict:
    """
    Estimate monthly costs for a Cloud Workflows + Gemini API pipeline.
    
    Pricing as of April 2026:
    - Gemini 2.5 Pro: $7/1M input tokens, $21/1M output tokens
    - Gemini 2.5 Flash: $0.15/1M input tokens, $0.60/1M output tokens
    - Cloud Workflows: $0.01/1,000 steps (first 5,000/month free)
    """
    PRICING = {
        "gemini-2.5-pro": {"input": 7.0, "output": 21.0},
        "gemini-2.5-flash": {"input": 0.15, "output": 0.60},
        "gemini-2.5-flash-lite": {"input": 0.075, "output": 0.30},
    }
    
    if model not in PRICING:
        raise ValueError(f"Unknown model: {model}. Choose from {list(PRICING.keys())}")
    
    monthly_executions = executions_per_day * 30
    monthly_steps = monthly_executions * steps_per_execution
    
    # Cloud Workflows cost (first 5,000 steps free)
    billable_steps = max(0, monthly_steps - 5_000)
    workflows_cost = (billable_steps / 1_000) * 0.01
    
    # Gemini API cost
    monthly_input_tokens = monthly_executions * avg_input_tokens
    monthly_output_tokens = monthly_executions * avg_output_tokens
    
    input_cost = (monthly_input_tokens / 1_000_000) * PRICING[model]["input"]
    output_cost = (monthly_output_tokens / 1_000_000) * PRICING[model]["output"]
    gemini_cost = input_cost + output_cost
    
    total = workflows_cost + gemini_cost
    
    return {
        "monthly_executions": monthly_executions,
        "monthly_steps": monthly_steps,
        "workflows_cost_usd": round(workflows_cost, 4),
        "gemini_api_cost_usd": round(gemini_cost, 4),
        "total_monthly_cost_usd": round(total, 4),
        "cost_breakdown": {
            "workflows_pct": round(workflows_cost / total * 100, 1) if total > 0 else 0,
            "gemini_api_pct": round(gemini_cost / total * 100, 1) if total > 0 else 0,
        }
    }
 
# Example: daily content generation pipeline
estimate = estimate_monthly_cost(
    executions_per_day=3,       # Run 3 times per day
    steps_per_execution=15,     # ~15 workflow steps per run
    avg_input_tokens=8_000,     # ~8K input tokens per execution
    avg_output_tokens=2_000,    # ~2K output tokens per execution
    model="gemini-2.5-pro"
)
# Output: workflows ~$0.02, Gemini API ~$5.94, Total ~$5.96/month
print(f"Estimated monthly cost: ${estimate['total_monthly_cost_usd']}")
print(f"  Cloud Workflows: ${estimate['workflows_cost_usd']} ({estimate['cost_breakdown']['workflows_pct']}%)")
print(f"  Gemini API: ${estimate['gemini_api_cost_usd']} ({estimate['cost_breakdown']['gemini_api_pct']}%)")