About This Code Showcase
This curated code walkthrough shows how Stock Manager orchestrates three AI agents to analyze sales, check inventory, and generate purchase orders for a steam bun shop.
The focus is on the multi-agent architecture (Google ADK), the SQL-based analysis tools, and the Google Sheets integration. Environment setup and deployment scripts are omitted for clarity.
Project Structure
projects/stock-manager/
├── main.py
├── agents.py
├── tools.py
├── schema.sql
├── db_setup.py
├── static/
│ └── index.html
├── dataset/
│ ├── suppliers.csv
│ ├── products.csv
│ ├── sales.csv
│ └── inventory.csv
├── Dockerfile
└── requirements.txt
Multi-Agent Architecture (agents.py)
The core of Stock Manager is a SequentialAgent that chains three specialized sub-agents. Each agent has its own instruction prompt and tool set:
from google.adk.agents import Agent, SequentialAgent
from tools import (
get_sales_summary, get_sales_trends,
get_inventory_status, get_low_stock_items,
create_purchase_order,
)
sales_analyst = Agent(
name="sales_analyst",
model="gemini-2.5-flash",
instruction="""You are the Sales Analyst agent for a steam bun shop.
Analyze sales data by calling BOTH tools:
1. Call get_sales_summary for average daily sales per product.
2. Call get_sales_trends to detect rising or falling demand.
Return a summary listing:
- Top 5 fast movers (highest daily sales)
- Top 5 slow movers (lowest daily sales)
- All products with RISING trends (10%+ increase)""",
tools=[get_sales_summary, get_sales_trends],
)
inventory_checker = Agent(
name="inventory_checker",
model="gemini-2.5-flash",
instruction="""You are the Inventory Checker agent.
Check stock levels by calling BOTH tools:
1. Call get_inventory_status for all products.
2. Call get_low_stock_items for items below reorder point.
Return: critical items, low items, and healthy items.""",
tools=[get_inventory_status, get_low_stock_items],
)
restock_decider = Agent(
name="restock_decider",
model="gemini-2.5-flash",
instruction="""You are the Restock Decision agent.
Review sales analysis and inventory check. Decide restocking:
1. LOW STOCK: below reorder point -> ALWAYS restock
2. TRENDING UP: 10%+ increase AND stock within 1.5x reorder -> restock early
3. HEALTHY + STABLE: skip
Call create_purchase_order with the final restock list.""",
tools=[create_purchase_order],
)
manager_agent = SequentialAgent(
name="stock_manager",
sub_agents=[sales_analyst, inventory_checker, restock_decider],
)
SQL Analysis Tools (tools.py)
Each agent calls Python functions that query AlloyDB. Here are the two sales analysis tools used by the Sales Analyst agent:
def get_sales_trends() -> dict:
"""Compare sales from the most recent 7 days vs the prior 7 days
to detect rising or falling demand."""
rows = _query("""
WITH date_range AS (
SELECT MAX(sale_date) AS max_date FROM sales
),
recent AS (
SELECT product_id, SUM(quantity_sold) AS recent_sold
FROM sales, date_range
WHERE sale_date > max_date - INTERVAL '7 days'
GROUP BY product_id
),
previous AS (
SELECT product_id, SUM(quantity_sold) AS prev_sold
FROM sales, date_range
WHERE sale_date > max_date - INTERVAL '14 days'
AND sale_date <= max_date - INTERVAL '7 days'
GROUP BY product_id
)
SELECT p.product_name,
COALESCE(r.recent_sold, 0) AS recent_sold,
COALESCE(pr.prev_sold, 0) AS prev_sold,
ROUND(((r.recent_sold - pr.prev_sold)::numeric
/ pr.prev_sold) * 100, 1) AS pct_change
FROM products p
LEFT JOIN recent r ON p.product_id = r.product_id
LEFT JOIN previous pr ON p.product_id = pr.product_id
ORDER BY pct_change DESC
""")
for row in rows:
pct = float(row["pct_change"])
if pct >= 10:
row["trend"] = "rising"
elif pct <= -10:
row["trend"] = "falling"
else:
row["trend"] = "stable"
return {"products": rows}
Google Sheets Integration (tools.py)
The Restock Decider agent calls this tool to push a purchase order to Google Sheets. Each order creates a new worksheet named by date:
def create_purchase_order(order_items_json: str) -> dict:
"""Create a purchase order and push it to Google Sheets."""
items = json.loads(order_items_json)
creds, _ = google.auth.default(
scopes=["https://www.googleapis.com/auth/spreadsheets"]
)
gc = gspread.authorize(creds)
sh = gc.open_by_key(os.environ["GOOGLE_SHEETS_ID"])
sheet_title = f"PO-{date.today().isoformat()}"
try:
worksheet = sh.worksheet(sheet_title)
worksheet.clear()
except WorksheetNotFound:
worksheet = sh.add_worksheet(title=sheet_title, rows=100, cols=10)
header = ["Product", "Quantity", "Supplier", "Contact Phone"]
rows = [header]
current_supplier = None
for item in sorted(items, key=lambda x: x["supplier_name"]):
supplier = item["supplier_name"]
if supplier != current_supplier:
if current_supplier:
rows.append(["", "", "", ""])
rows.append([f"--- {supplier} ---", "", "", ""])
current_supplier = supplier
rows.append([item["product_name"], str(item["quantity"]),
supplier, item["contact_phone"]])
worksheet.update(range_name="A1", values=rows)
return {"status": "success", "sheet_url": f"https://docs.google.com/spreadsheets/d/{sheets_id}/edit"}
FastAPI Endpoint (main.py)
The API endpoint receives user messages, runs the ADK agent pipeline, and returns the final response:
@app.post("/api/restock")
async def restock(request: Request):
data = await request.json()
user_message = data.get("message", "What should I restock this week?")
content = types.Content(
role="user", parts=[types.Part(text=user_message)]
)
all_texts = []
async for event in runner.run_async(
user_id=user_id, session_id=session_id, new_message=content
):
if event.content and event.content.parts:
for part in event.content.parts:
if part.text and part.text.strip():
all_texts.append(part.text.strip())
response_text = all_texts[-1] if all_texts else "Could not process request."
return JSONResponse({"response": response_text})
Database Schema (schema.sql)
Four tables model the shop's data: suppliers, products, sales transactions, and current inventory:
CREATE TABLE suppliers (
supplier_id VARCHAR(10) PRIMARY KEY,
supplier_name VARCHAR(100),
contact_phone VARCHAR(20),
email VARCHAR(100),
lead_time_days INTEGER
);
CREATE TABLE products (
product_id VARCHAR(10) PRIMARY KEY,
product_name VARCHAR(100),
category VARCHAR(50),
unit_price DECIMAL(10,2),
reorder_point INTEGER,
reorder_quantity INTEGER,
supplier_id VARCHAR(10) REFERENCES suppliers
);
CREATE TABLE sales (
sale_id VARCHAR(20) PRIMARY KEY,
product_id VARCHAR(10) REFERENCES products,
quantity_sold INTEGER,
sale_date DATE
);
CREATE TABLE inventory (
product_id VARCHAR(10) PRIMARY KEY REFERENCES products,
current_stock INTEGER,
last_updated DATE
);