Google Cloud Run Function - AgentFlow Documentation

Overview

Execute Google Cloud Run functions from AgentFlow for serverless AI processing and automation. Build custom logic that scales automatically and integrates with Google Cloud services.

Prerequisites

Google Cloud Account

Active GCP account with billing enabled

Cloud Run Access

Permissions to deploy Cloud Run services

gcloud CLI

Install and configure gcloud CLI

AgentFlow Admin

Admin access required

Step 1: Setup Google Cloud Run Function

Install Prerequisites

Install gcloud CLI

# macOS
brew install google-cloud-sdk

# Linux
curl https://sdk.cloud.google.com | bash

# Windows
# Download installer from cloud.google.com/sdk

Authenticate

gcloud auth login
gcloud config set project YOUR_PROJECT_ID

Enable APIs

gcloud services enable run.googleapis.com
gcloud services enable cloudbuild.googleapis.com

Create Cloud Run Function

Initialize Project

mkdir agentflow-processor
cd agentflow-processor

Create main.py

import os
import json
from flask import Flask, request, jsonify
from google.cloud import firestore
import openai

app = Flask(__name__)

@app.route('/process', methods=['POST'])
def process_message():
    try:
        # Get request data
        data = request.get_json()
        message = data.get('payload', {}).get('message_content')
        user_id = data.get('payload', {}).get('user_id')

        # Process with AI
        openai.api_key = os.environ.get('OPENAI_API_KEY')
        response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": message}
            ]
        )

        result = response.choices[0].message.content

        # Store in Firestore (optional)
        db = firestore.Client()
        db.collection('responses').add({
            'user_id': user_id,
            'message': message,
            'result': result,
            'timestamp': firestore.SERVER_TIMESTAMP
        })

        return jsonify({
            'data': {
                'result': result,
                'status': 'success',
                'metadata': {
                    'user_id': user_id,
                    'processed_at': 'timestamp'
                }
            }
        }), 200

    except Exception as e:
        return jsonify({
            'error': str(e),
            'status': 'failed'
        }), 500

if __name__ == '__main__':
    port = int(os.environ.get('PORT', 8080))
    app.run(host='0.0.0.0', port=port)

Create requirements.txt

Flask==3.0.0
gunicorn==21.2.0
openai==1.3.0
google-cloud-firestore==2.13.1
google-auth==2.23.4

Create Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:app

Deploy to Cloud Run

Build Container

gcloud builds submit --tag gcr.io/YOUR_PROJECT_ID/agentflow-processor

Deploy Service

gcloud run deploy agentflow-processor \
  --image gcr.io/YOUR_PROJECT_ID/agentflow-processor \
  --platform managed \
  --region us-central1 \
  --allow-unauthenticated \
  --set-env-vars OPENAI_API_KEY=your_api_key \
  --memory 1Gi \
  --cpu 1 \
  --max-instances 10 \
  --timeout 300

Get Service URL

gcloud run services describe agentflow-processor \
  --platform managed \
  --region us-central1 \
  --format 'value(status.url)'

Output: https://agentflow-processor-xyz-uc.a.run.app

Step 2: Setup Authentication

Create Service Account

# Create service account
gcloud iam service-accounts create agentflow-invoker \
  --display-name "AgentFlow Cloud Run Invoker"

# Grant permissions
gcloud run services add-iam-policy-binding agentflow-processor \
  --member="serviceAccount:agentflow-invoker@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/run.invoker" \
  --region us-central1

# Create key
gcloud iam service-accounts keys create key.json \
  --iam-account agentflow-invoker@YOUR_PROJECT_ID.iam.gserviceaccount.com

Generate Auth Token

from google.oauth2 import service_account
from google.auth.transport.requests import Request

# Load credentials
credentials = service_account.Credentials.from_service_account_file(
    'key.json',
    scopes=['https://www.googleapis.com/auth/cloud-platform']
)

# Get token
credentials.refresh(Request())
token = credentials.token
print(f"Bearer {token}")

Step 3: Create AI Connection in AgentFlow

Manual Configuration

Admin Dashboard → AI Models → Add Model
Basic Info:
- Name: Cloud Run AI Processor
- Model ID: cloud-run-function-executor
- Description: Serverless AI processing via Google Cloud Run
API Settings:
- Endpoint: https://agentflow-processor-xyz-uc.a.run.app/process
- Method: POST

Headers:

{
  "Authorization": "Bearer {{cloud_run_token}}",
  "Content-Type": "application/json",
  "User-Agent": "ChatPlatform/1.0"
}

Request Schema:

{
  "function_name": "agentflow-processor",
  "project_id": "YOUR_PROJECT_ID",
  "region": "us-central1",
  "endpoint_path": "/process",
  "payload": {
    "message": "{{message}}",
    "user_id": "{{user_id}}",
    "timestamp": "{{timestamp}}",
    "context": "{{context}}"
  },
  "execution_config": {
    "timeout": 300,
    "memory": "1Gi",
    "cpu": "1",
    "concurrency": 100,
    "max_instances": 10
  }
}

Response Path: data.result
Save

Step 4: Import via YAML

YAML Configuration

Create cloud-run-function-config.yaml:

name: "Google Cloud Run Function"
model_id: "cloud-run-function-executor"
description: "Execute Google Cloud Run functions for serverless AI processing and automation"
endpoint: "https://{{function_name}}-{{project_id}}.{{region}}.run.app/{{endpoint_path}}"
method: "POST"

headers:
  Authorization: "Bearer {{cloud_run_token}}"
  Content-Type: "application/json"
  User-Agent: "ChatPlatform/1.0"

request_schema:
  function_name: "{{function_name}}"
  project_id: "{{project_id}}"
  region: "{{region}}"
  endpoint_path: "{{endpoint_path}}"
  payload:
    message: "{{message}}"
    user_id: "{{user_id}}"
    timestamp: "{{timestamp}}"
    context: "{{context}}"
  execution_config:
    timeout: 300
    memory: "1Gi"
    cpu: "1"
    concurrency: 100
    max_instances: 10

response_path: "data.result"

message_format:
  preset: "custom"
  mapping:
    role:
      source: "role"
      target: "payload.user_role"
      transform: "lowercase"
    content:
      source: "content"
      target: "payload.message_content"
      transform: "none"
    timestamp:
      source: "timestamp"
      target: "payload.timestamp"
      transform: "iso8601"
    context:
      source: "context"
      target: "payload.additional_context"
      transform: "json_stringify"
  customFields:
    - name: "cloud_run_metadata"
      value:
        platform: "google_cloud_run"
        version: "2.0"
        execution_type: "serverless"
        region: "{{region}}"
        project: "{{project_id}}"
      type: "object"
    - name: "function_configuration"
      value:
        timeout: 300
        memory: "1Gi"
        cpu: "1"
        concurrency: 100
        max_instances: 10
        min_instances: 0
      type: "object"

suggestion_prompts:
  - "Create a Cloud Run function for real-time image processing and analysis"
  - "Set up a function for automated document parsing and data extraction"
  - "Build a function for real-time language translation services"
  - "Create a function for automated email processing and routing"
  - "Set up a function for real-time data validation and transformation"

Import Process

Update all {{placeholder}} values
Admin Dashboard → AI Models → Import Model
Upload YAML
Enter credentials
Import

Step 5: Assign to Group

Admin Dashboard → Groups
Select group (e.g., “Development Team”)
Manage Models → Enable Cloud Run Function
Configure:
- Execution Limit: 10,000/day
- Timeout: 5 minutes
- Priority: High
Save

Step 6: Use in Chat

Trigger Functions

Chat → New Conversation
Select Cloud Run AI Processor
Send message

Example Use Cases

Analyze this image URL for objects, text, and sentiment:
https://example.com/image.jpg

Process this CSV data and generate summary statistics with visualizations
[paste CSV data]

Translate this text to Spanish, French, and German:
"Welcome to our platform. Please complete your profile."

Extract structured data from this invoice:
[upload or paste invoice content]

Advanced Function Examples

Image Processing Function

import os
from flask import Flask, request, jsonify
from google.cloud import vision
from google.cloud import storage

app = Flask(__name__)

@app.route('/analyze-image', methods=['POST'])
def analyze_image():
    data = request.get_json()
    image_url = data.get('payload', {}).get('image_url')

    # Initialize Vision AI
    client = vision.ImageAnnotatorClient()

    # Analyze image
    image = vision.Image()
    image.source.image_uri = image_url

    response = client.annotate_image({
        'image': image,
        'features': [
            {'type_': vision.Feature.Type.LABEL_DETECTION},
            {'type_': vision.Feature.Type.TEXT_DETECTION},
            {'type_': vision.Feature.Type.FACE_DETECTION},
        ]
    })

    results = {
        'labels': [label.description for label in response.label_annotations],
        'text': response.text_annotations[0].description if response.text_annotations else '',
        'faces': len(response.face_annotations)
    }

    return jsonify({'data': {'result': results}})

Data Processing Function

import pandas as pd
import json
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/process-data', methods=['POST'])
def process_data():
    data = request.get_json()
    csv_data = data.get('payload', {}).get('csv_data')

    # Parse CSV
    df = pd.read_csv(io.StringIO(csv_data))

    # Generate statistics
    stats = {
        'summary': df.describe().to_dict(),
        'correlations': df.corr().to_dict(),
        'missing_values': df.isnull().sum().to_dict(),
        'data_types': df.dtypes.astype(str).to_dict()
    }

    return jsonify({'data': {'result': stats}})

Translation Function

from google.cloud import translate_v2 as translate

app = Flask(__name__)

@app.route('/translate', methods=['POST'])
def translate_text():
    data = request.get_json()
    text = data.get('payload', {}).get('text')
    target_languages = data.get('payload', {}).get('languages', ['es', 'fr', 'de'])

    # Initialize translator
    client = translate.Client()

    # Translate to multiple languages
    translations = {}
    for lang in target_languages:
        result = client.translate(text, target_language=lang)
        translations[lang] = result['translatedText']

    return jsonify({'data': {'result': translations}})

Scaling & Performance

Auto-Scaling Configuration

gcloud run services update agentflow-processor \
  --min-instances 0 \
  --max-instances 100 \
  --concurrency 80 \
  --cpu-throttling \
  --region us-central1

Performance Optimization

Cold Start Reduction

Use minimum instances (min-instances=1)
Optimize container size
Implement connection pooling

Memory Optimization

# Adjust based on workload
--memory 512Mi   # Light processing
--memory 2Gi     # Heavy processing
--memory 4Gi     # ML inference

CPU Allocation

--cpu 1      # Standard
--cpu 2      # CPU-intensive
--cpu 4      # Maximum performance

Concurrency Tuning

# Requests per instance
--concurrency 80   # I/O bound
--concurrency 1    # CPU bound

Regional Deployment

Deploy to multiple regions:

# Deploy to us-central1
gcloud run deploy agentflow-processor \
  --image gcr.io/PROJECT_ID/agentflow-processor \
  --region us-central1

# Deploy to europe-west1
gcloud run deploy agentflow-processor \
  --image gcr.io/PROJECT_ID/agentflow-processor \
  --region europe-west1

# Deploy to asia-east1
gcloud run deploy agentflow-processor \
  --image gcr.io/PROJECT_ID/agentflow-processor \
  --region asia-east1

Monitoring & Logging

Cloud Logging

View logs:

# Stream logs
gcloud run services logs tail agentflow-processor \
  --region us-central1

# Filter by severity
gcloud logging read "resource.type=cloud_run_revision AND severity>=ERROR" \
  --limit 50

Cloud Monitoring

Setup alerts:

# Create alert policy
gcloud alpha monitoring policies create \
  --notification-channels=CHANNEL_ID \
  --display-name="Cloud Run High Error Rate" \
  --condition-display-name="Error rate > 5%" \
  --condition-threshold-value=5 \
  --condition-threshold-duration=60s

Custom Metrics

from google.cloud import monitoring_v3

def write_custom_metric(project_id, value):
    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{project_id}"

    series = monitoring_v3.TimeSeries()
    series.metric.type = "custom.googleapis.com/agentflow/processing_time"
    series.resource.type = "cloud_run_revision"

    point = series.points.add()
    point.value.double_value = value
    point.interval.end_time.seconds = int(time.time())

    client.create_time_series(request={"name": project_name, "time_series": [series]})

Security Best Practices

Authentication

Always require authentication for sensitive endpoints

Service Accounts

Use least-privilege service accounts

Secrets Management

Store secrets in Secret Manager, not environment variables

Network Security

Use VPC connectors for private resources

Input Validation

Validate and sanitize all inputs

Audit Logging

Enable Cloud Audit Logs

Using Secret Manager

from google.cloud import secretmanager

def access_secret(project_id, secret_id):
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{project_id}/secrets/{secret_id}/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return response.payload.data.decode("UTF-8")

# Usage
openai_key = access_secret("my-project", "openai-api-key")

Cost Optimization

Pricing Breakdown

Resource	Cost
CPU	$0.00002400/vCPU-second
Memory	$0.00000250/GiB-second
Requests	$0.40/million requests
Networking	$0.12/GB egress

Optimization Strategies

Right-Size Resources

Match CPU/memory to actual needs

Minimize Cold Starts

Use min-instances only when necessary

Optimize Container

Reduce image size and dependencies

Cache Responses

Implement caching for repeated queries

Batch Processing

Process multiple items per request

Troubleshooting

503 Service Unavailable

Causes:

Cold start timeout
Resource limits exceeded
Deployment issues

Fix:

Increase timeout
Add min-instances
Check deployment logs

429 Too Many Requests

Cause: Rate limit exceededFix:

Increase max-instances
Implement request queuing
Distribute across regions

Memory Exceeded

Symptoms: OOM errors in logsFix:

Increase memory allocation
Optimize code for memory efficiency
Implement streaming for large data

Authentication Failed

Check:

Service account permissions
Token validity
IAM policies

Fix: Regenerate credentials

CI/CD Integration

GitHub Actions Deployment

name: Deploy to Cloud Run

on:
  push:
    branches: [ main ]

jobs:
  deploy:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v2

    - name: Setup Cloud SDK
      uses: google-github-actions/setup-gcloud@v0
      with:
        service_account_key: ${{ secrets.GCP_SA_KEY }}
        project_id: ${{ secrets.GCP_PROJECT_ID }}

    - name: Build and Deploy
      run: |
        gcloud builds submit --tag gcr.io/${{ secrets.GCP_PROJECT_ID }}/agentflow-processor
        gcloud run deploy agentflow-processor \
          --image gcr.io/${{ secrets.GCP_PROJECT_ID }}/agentflow-processor \
          --platform managed \
          --region us-central1 \
          --allow-unauthenticated

Next Steps

LangChain Agents

Build AI agents with tools

Workflow Automation

Combine with n8n workflows

OpenAI Assistants

Advanced AI agents

Analytics

Monitor function performance

​Overview

​Prerequisites

Google Cloud Account

Cloud Run Access

gcloud CLI

AgentFlow Admin

​Step 1: Setup Google Cloud Run Function

​Install Prerequisites

​Create Cloud Run Function

​Deploy to Cloud Run

​Step 2: Setup Authentication

​Create Service Account

​Generate Auth Token

​Step 3: Create AI Connection in AgentFlow

​Manual Configuration

​Step 4: Import via YAML

​YAML Configuration

​Import Process

​Step 5: Assign to Group

​Step 6: Use in Chat

​Trigger Functions

​Example Use Cases

​Advanced Function Examples

​Image Processing Function

​Data Processing Function

​Translation Function

​Scaling & Performance

​Auto-Scaling Configuration

​Performance Optimization

​Regional Deployment

​Monitoring & Logging

​Cloud Logging

​Cloud Monitoring

​Custom Metrics

​Security Best Practices