GCP AgentFlow

Agentic AI orchestration toolkit for Google Cloud Workflows, Pub/Sub, BigQuery, Datastore, and ML pipelines.

gcp-agentflow helps engineers design and operate event-driven Google Cloud automation patterns where Pub/Sub events trigger workflow decisions, Datastore stores operational state, BigQuery stores analytics, and ML/Agentic AI logic recommends the next action.

What it is for

• Google Cloud event-driven architectures
• Agentic AI workflow orchestration
• Pub/Sub message routing
• Datastore operational state tracking
• BigQuery analytics event logging
• ML-based routing and decision support
• Workflow-ready payload generation

Install

pip install gcp-agentflow

Python Usage

from gcp_agentflow import AgentDecisionInput, decide_next_action

event = AgentDecisionInput(
    event_type="file_arrived",
    source="pubsub",
    risk_score=72,
    payload={"bucket": "incoming", "name": "file.csv"}
)

decision = decide_next_action(event)
print(decision.action)
print(decision.reason)

CLI Usage

gcp-agentflow decide --event-type file_arrived --risk-score 72

Key Concepts

gcp-agentflow is intentionally lightweight. It does not force a specific architecture. It provides reusable building blocks:

• Decision Engine: Determines whether to route, retry, quarantine, approve, or alert.
• Workflow Payload Builder: Creates clean payloads for Google Workflows or Cloud Run services.
• Pub/Sub Publisher Wrapper: Publishes JSON events safely.
• BigQuery Event Logger: Inserts structured analytics events.
• Datastore State Store: Saves and retrieves operational state by key.

Build and Publish

python -m pip install --upgrade build twine
python -m build
twine check dist/*
twine upload dist/*

License

MIT

Author: Raghava Chellu