GCS for Workflows

Agno supports using Google Cloud Storage (GCS) as a storage backend for Workflows using the GcsJsonDb class. This storage backend stores session data as JSON blobs in a GCS bucket.

Usage

Configure your workflow with GCS storage to enable cloud-based session persistence.

gcs_for_workflow.py

"""
Run: `uv pip install agno openai google-auth google-cloud-storage ddgs agno` to install the dependencies
"""

import os
import google.auth
from google.auth.credentials import AnonymousCredentials
from agno.agent import Agent
from agno.db.gcs_json import GcsJsonDb
from agno.models.openai import OpenAIResponses
from agno.team import Team
from agno.tools.hackernews import HackerNewsTools
from agno.tools.websearch import WebSearchTools
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow

# Local emulator uses anonymous credentials; real GCS uses configured ADC.
if os.environ.get("STORAGE_EMULATOR_HOST"):
    credentials, project_id = AnonymousCredentials(), "local-test"
else:
    credentials, project_id = google.auth.default()

# The bucket must already exist. Keep this name stable across restarts.
bucket_name = os.environ["GCS_BUCKET_NAME"]

db = GcsJsonDb(
    bucket_name=bucket_name,
    prefix="workflow/",
    project=project_id,
    credentials=credentials,
)

# Define agents
hackernews_agent = Agent(
    name="Hackernews Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[HackerNewsTools()],
    role="Extract key insights and content from Hackernews posts",
)
web_agent = Agent(
    name="Web Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[WebSearchTools()],
    role="Search the web for the latest news and trends",
)

# Define research team for complex analysis
research_team = Team(
    name="Research Team",
    members=[hackernews_agent, web_agent],
    instructions="Research tech topics from Hackernews and the web",
)

content_planner = Agent(
    name="Content Planner",
    model=OpenAIResponses(id="gpt-5.2"),
    instructions=[
        "Plan a content schedule over 4 weeks for the provided topic and research content",
        "Ensure that I have posts for 3 posts per week",
    ],
)

# Define steps
research_step = Step(
    name="Research Step",
    team=research_team,
)

content_planning_step = Step(
    name="Content Planning Step",
    agent=content_planner,
)

# Create and use workflow
if __name__ == "__main__":
    content_creation_workflow = Workflow(
        name="Content Creation Workflow",
        description="Automated content creation from blog posts to social media",
        db=db,
        steps=[research_step, content_planning_step],
    )
    content_creation_workflow.print_response(
        input="AI trends in 2024",
        markdown=True,
    )

Prerequisites

Install dependencies and configure the model

Use a virtual environment:

uv pip install -U agno google-auth google-cloud-storage openai ddgs
export OPENAI_API_KEY="your-openai-api-key"

Choose an existing bucket

The examples use GCS_BUCKET_NAME. Create a bucket once using the Google Cloud bucket setup guide, or ask your administrator for an existing bucket. Grant the application identity the object permissions it needs for that bucket. GcsJsonDb obtains a bucket handle; it does not create the bucket.

export GCS_BUCKET_NAME="your-existing-bucket"

Keep the bucket and prefix stable to read earlier runs. GCS JSON storage rewrites blobs; it does not provide the same concurrent-write semantics as a transactional database.

Configure real GCS authentication

For local development, install the Google Cloud CLI, select your project, and configure Application Default Credentials:

gcloud init
gcloud auth application-default login

For a service account, set GOOGLE_APPLICATION_CREDENTIALS to its credential file instead. The examples use google.auth.default() when STORAGE_EMULATOR_HOST is absent. The model still needs its own OpenAI key.

Local Testing with Fake GCS

To use fake-gcs-server, prepare a preloaded bucket before starting the server:

mkdir -p fake-gcs-data/example-gcs-bucket
echo "Local test bucket" > fake-gcs-data/example-gcs-bucket/seed.txt

Save as compose.yaml:

services:
  fake-gcs-server:
    image: fsouza/fake-gcs-server:latest
    ports:
      - "127.0.0.1:4443:4443"
    command: ["-scheme", "http", "-port", "4443", "-public-host", "localhost:4443"]
    volumes:
      - ./fake-gcs-data:/data

Start Docker Compose and point the example at the local bucket:

docker compose up -d
export STORAGE_EMULATOR_HOST="http://localhost:4443"
export GCS_BUCKET_NAME="example-gcs-bucket"
python gcs_for_agent.py

Use the filename from the Team or Workflow page when running those examples. Their explicit emulator branch supplies AnonymousCredentials and project="local-test", so it does not call google.auth.default(). The emulator is for local storage tests; model and web-search calls still use their real services.

Params

Parameter	Type	Default	Description
`id`	`Optional[str]`	-	Database ID. Derived deterministically from bucket, project and prefix when omitted.
`bucket_name`	`str`	-	Name of the GCS bucket where JSON files will be stored.
`prefix`	`Optional[str]`	-	Path prefix for organizing files in the bucket. Defaults to "agno/".
`session_table`	`Optional[str]`	-	Name of the JSON file to store sessions (without .json extension).
`runs_table`	`Optional[str]`	`None`	Storage name for individual runs. Defaults to `agno_runs`, or `<session_table>_runs` when a custom session name is supplied.
`memory_table`	`Optional[str]`	-	Name of the JSON file to store user memories.
`metrics_table`	`Optional[str]`	-	Name of the JSON file to store metrics.
`eval_table`	`Optional[str]`	-	Name of the JSON file to store evaluation runs.
`knowledge_table`	`Optional[str]`	-	Name of the JSON file to store knowledge content.
`traces_table`	`Optional[str]`	-	Name of the JSON file to store traces.
`spans_table`	`Optional[str]`	-	Name of the JSON file to store spans.
`project`	`Optional[str]`	-	GCP project ID. If None, uses default project.
`credentials`	`Optional[Any]`	-	GCP credentials. If None, uses default credentials.

Run the Example

Save the code as gcs_for_workflow.py, complete the prerequisites above, then run:

python gcs_for_workflow.py