Lord1Egypt The Agent SaaS Factory: Build, Deploy, and Monetize Software Products with Autonomous AI Agents
Notice: This is an educational guide with illustrative code examples. It does not execute code or install dependencies. All examples use the GreenHelix sandbox (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required to get started.
Referenced credentials (you supply these in your own environment):
GREENHELIX_API_KEY: API authentication for GreenHelix gateway (read/write access to purchased API tools only)STRIPE_API_KEY: Stripe API key for card payment processing (scoped to payment intents only)
When Stripe announced the Agentic Commerce Suite in March 2026 -- with Visa, OpenAI, and Samsung as launch partners -- they framed it as "AI agents that can buy things." That is the merchant side of the equation. An agent browses a storefront, picks a product, and pays with a human's saved card. But what about the other side? What about an agent that builds the product, sets up the billing, deploys the database, handles the disputes, and operates the entire business -- end to end, without a human writing a single line of code or clicking a single button in a dashboard?
That is what this guide builds. A factory pattern where autonomous AI agents receive a product specification and produce a running, revenue-generating micro-SaaS: repository scaffolded on GitHub, schema designed and migrated in Postgres, billing configured in Stripe, dispute resolution automated, and the whole pipeline orchestrated by a single AgentSaaSFactory class. The BotStall marketplace demonstrated in late 2025 that agents could list and sell digital goods. We go further: agents that build, operate, and defend the entire software business.
The infrastructure is the GreenHelix A2A Commerce Gateway's pro-tier integration tools: 10 GitHub tools, 13 Stripe tools, 6 Postgres tools, and 5 Dispute tools -- 34 tools total, all accessible through a single HTTP endpoint at $0.005-$0.01 per call. This guide covers every one of them.
What You'll Learn
- Chapter 1: Why Agents Should Build SaaS
- Chapter 2: Setting Up the Agent's Development Environment
- Chapter 3: The Data Layer: Agents Managing Postgres
- Chapter 4: Monetization: Agents Running Stripe
- Chapter 5: The Full Factory: Orchestrating the Pipeline
- Chapter 6: Dispute Resolution and Customer Support Automation
- Chapter 7: Security, Cost Governance, and Guardrails
- Chapter 8: Production Recipes
- Chapter 9: What's Next
Full Guide
The Agent SaaS Factory: Build, Deploy, and Monetize Software Products with Autonomous AI Agents
When Stripe announced the Agentic Commerce Suite in March 2026 -- with Visa, OpenAI, and Samsung as launch partners -- they framed it as "AI agents that can buy things." That is the merchant side of the equation. An agent browses a storefront, picks a product, and pays with a human's saved card. But what about the other side? What about an agent that builds the product, sets up the billing, deploys the database, handles the disputes, and operates the entire business -- end to end, without a human writing a single line of code or clicking a single button in a dashboard?
That is what this guide builds. A factory pattern where autonomous AI agents receive a product specification and produce a running, revenue-generating micro-SaaS: repository scaffolded on GitHub, schema designed and migrated in Postgres, billing configured in Stripe, dispute resolution automated, and the whole pipeline orchestrated by a single AgentSaaSFactory class. The BotStall marketplace demonstrated in late 2025 that agents could list and sell digital goods. We go further: agents that build, operate, and defend the entire software business.
The infrastructure is the GreenHelix A2A Commerce Gateway's pro-tier integration tools: 10 GitHub tools, 13 Stripe tools, 6 Postgres tools, and 5 Dispute tools -- 34 tools total, all accessible through a single HTTP endpoint at $0.005-$0.01 per call. This guide covers every one of them.
Getting started: All examples in this guide work with the GreenHelix sandbox (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required.
Table of Contents
- Why Agents Should Build SaaS
- Setting Up the Agent's Development Environment
- The Data Layer: Agents Managing Postgres
- Monetization: Agents Running Stripe
- The Full Factory: Orchestrating the Pipeline
- Dispute Resolution and Customer Support Automation
- Security, Cost Governance, and Guardrails
- Production Recipes
- What's Next
Chapter 1: Why Agents Should Build SaaS
From Tools to Operators
The 2025 frame was "agents as tools." Developer writes prompt, agent generates code, developer reviews and deploys. Faster pair programming. Useful. Not transformative.
The shift is "agents as operators." The agent does not help you build a SaaS product -- it builds the SaaS product. Creates the repository. Designs the schema. Sets up Stripe pricing. Writes code, submits PRs, reviews diffs, merges. Monitors disputes and resolves them. The human's role shifts from implementer to investor: you provide the specification and the API credits. The agent does the rest.
Stripe's Agentic Commerce Suite already envisions agents that autonomously complete purchases and handle disputes. The missing piece is the supply side -- agents that build and operate the businesses those buyer agents purchase from.
The BotStall Precedent
BotStall (late 2025) proved one half: agents listing and selling digital goods on a marketplace. But BotStall agents were vendors, not builders. A human created the product; the agent handled distribution. The Agent SaaS Factory closes the loop: the agent creates, operates, monetizes, and defends the entire business.
Architecture Overview
The factory runs on four services, all accessed through the GreenHelix A2A Commerce Gateway:
+------------------------------------------------------------+
| AgentSaaSFactory |
| |
| +------------------+ +------------------+ |
| | AgentDeveloper | | AgentDBA | |
| | 10 GitHub tools | | 6 Postgres tools | |
| | $0.005/call | | $0.01/call | |
| +--------+---------+ +--------+---------+ |
| | | |
| +--------+---------+ +--------+---------+ |
| | AgentBilling | | AgentArbitrator | |
| | 13 Stripe tools | | 5 Dispute tools | |
| | $0.01/call | | varies | |
| +--------+---------+ +--------+---------+ |
| | | |
+-----------|----------------------|-------------------------+
| |
+--------v----------------------v---------+
| GreenHelix A2A Commerce Gateway |
| Pro-tier: GitHub + Stripe + Postgres |
| + Disputes |
+------------------------------------------+
Pro-Tier Cost Model
Every tool call has a fixed cost. There is no per-seat pricing, no monthly minimum, no overage surprise. You pay exactly for what the agent does.
| Service | Tools | Cost per Call | Typical Calls per SaaS Build |
|---|---|---|---|
| GitHub | 10 | $0.005 | 40-60 |
| Stripe | 13 | $0.01 | 15-25 |
| Postgres | 6 | $0.01 | 20-40 |
| Disputes | 5 | varies | 0-10 (reactive) |
A complete micro-SaaS build costs $1.00-$1.50 in tool calls. Ongoing operation runs $0.10-$0.30/day. At $29/month from a single customer, the unit economics are immediately positive.
Chapter 2: Setting Up the Agent's Development Environment
The AgentDeveloper Class
This class wraps all 10 GitHub tools into a development interface. The agent uses it to create repositories, scaffold projects, manage issues, submit pull requests, and perform code review.
import requests
import json
import time
from typing import Optional, List
class AgentDeveloper:
"""GitHub development client for the GreenHelix A2A Commerce Gateway.
Wraps the 10 GitHub integration tools into a development workflow
interface for autonomous repository management, code review, and
issue tracking.
"""
def __init__(
self,
api_key: str,
agent_id: str,
base_url: str = "https://api.greenhelix.net/v1",
):
self.base_url = base_url
self.agent_id = agent_id
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
})
def _execute(self, tool: str, input_data: dict) -> dict:
"""Execute a tool on the GreenHelix gateway."""
resp = self.session.post(
f"{self.base_url}/v1",
json={"tool": tool, "input": input_data},
)
if resp.status_code == 402:
raise BudgetExhaustedError(
f"Agent {self.agent_id} budget exceeded: {resp.text}"
)
resp.raise_for_status()
return resp.json()
# -- Repository Management -----------------------------------------
def create_repo(
self,
name: str,
description: str,
private: bool = True,
) -> dict:
"""Create a new GitHub repository."""
return self._execute("create_repo", {
"name": name,
"description": description,
"private": private,
"auto_init": True,
})
def get_repo(self, owner: str, repo: str) -> dict:
"""Get repository metadata."""
return self._execute("get_repo", {
"owner": owner,
"repo": repo,
})
# -- File Operations -----------------------------------------------
def get_file_contents(
self,
owner: str,
repo: str,
path: str,
ref: Optional[str] = None,
) -> dict:
"""Read a file from a repository."""
payload = {"owner": owner, "repo": repo, "path": path}
if ref:
payload["ref"] = ref
return self._execute("get_file_contents", payload)
def create_or_update_file(
self,
owner: str,
repo: str,
path: str,
content: str,
message: str,
branch: Optional[str] = None,
sha: Optional[str] = None,
) -> dict:
"""Write or update a file in a repository."""
payload = {
"owner": owner,
"repo": repo,
"path": path,
"content": content,
"message": message,
}
if branch:
payload["branch"] = branch
if sha:
payload["sha"] = sha
return self._execute("create_or_update_file", payload)
# -- Issue Tracking ------------------------------------------------
def create_issue(
self,
owner: str,
repo: str,
title: str,
body: str,
labels: Optional[List[str]] = None,
) -> dict:
"""File a bug report or feature request."""
payload = {
"owner": owner,
"repo": repo,
"title": title,
"body": body,
}
if labels:
payload["labels"] = labels
return self._execute("create_issue", payload)
def get_issue(
self,
owner: str,
repo: str,
issue_number: int,
) -> dict:
"""Retrieve issue details."""
return self._execute("get_issue", {
"owner": owner,
"repo": repo,
"issue_number": issue_number,
})
# -- Code Review ---------------------------------------------------
def create_pull_request(
self,
owner: str,
repo: str,
title: str,
body: str,
head: str,
base: str = "main",
) -> dict:
"""Submit code changes for review."""
return self._execute("create_pull_request", {
"owner": owner,
"repo": repo,
"title": title,
"body": body,
"head": head,
"base": base,
})
def merge_pull_request(
self,
owner: str,
repo: str,
pull_number: int,
merge_method: str = "squash",
) -> dict:
"""Merge an approved pull request."""
return self._execute("merge_pull_request", {
"owner": owner,
"repo": repo,
"pull_number": pull_number,
"merge_method": merge_method,
})
# -- Code Intelligence ---------------------------------------------
def search_code(
self,
query: str,
owner: Optional[str] = None,
repo: Optional[str] = None,
) -> dict:
"""Search across codebases."""
payload = {"query": query}
if owner and repo:
payload["query"] = f"{query} repo:{owner}/{repo}"
return self._execute("search_code", payload)
def list_commits(
self,
owner: str,
repo: str,
sha: Optional[str] = None,
per_page: int = 30,
) -> dict:
"""View commit history."""
payload = {
"owner": owner,
"repo": repo,
"per_page": per_page,
}
if sha:
payload["sha"] = sha
return self._execute("list_commits", payload)
# -- High-Level Workflows ------------------------------------------
def scaffold_project(
self,
owner: str,
repo: str,
files: dict,
branch: str = "main",
) -> List[dict]:
"""Write multiple files to scaffold a project.
Args:
files: dict mapping file paths to content strings.
Example: {"src/app.py": "...", "requirements.txt": "..."}
"""
results = []
for path, content in files.items():
result = self.create_or_update_file(
owner=owner,
repo=repo,
path=path,
content=content,
message=f"scaffold: add {path}",
branch=branch,
)
results.append(result)
return results
def review_and_merge(
self,
owner: str,
repo: str,
head_branch: str,
title: str,
description: str,
) -> dict:
"""Create a PR, perform automated review, and merge.
The agent creates the PR, searches the diff for known
anti-patterns, then merges if clean.
"""
pr = self.create_pull_request(
owner=owner,
repo=repo,
title=title,
body=description,
head=head_branch,
base="main",
)
pull_number = pr["number"]
# Agent-driven code review: search for common issues
issues_found = []
for pattern in ["eval(", "exec(", "password =", "secret ="]:
hits = self.search_code(
query=pattern,
owner=owner,
repo=repo,
)
if hits.get("total_count", 0) > 0:
issues_found.append(pattern)
if issues_found:
self.create_issue(
owner=owner,
repo=repo,
title=f"Security review: patterns found in PR #{pull_number}",
body=f"Found: {', '.join(issues_found)}. Review before merge.",
labels=["security", "automated-review"],
)
return {"status": "blocked", "pr": pr, "issues": issues_found}
merged = self.merge_pull_request(
owner=owner,
repo=repo,
pull_number=pull_number,
)
return {"status": "merged", "pr": pr, "merge": merged}
Pattern: Agent-Driven Code Review
The agent reads the diff, searches the codebase for related code, and judges whether the change is safe -- beyond what linters and CI catch.
import os
dev = AgentDeveloper(
api_key=os.environ["GREENHELIX_API_KEY"],
agent_id="dev-agent-01",
)
# Agent reviews its own work before merging
owner, repo = "myorg", "newsletter-saas"
# 1. Check recent commits on the feature branch
commits = dev.list_commits(owner, repo, sha="feature/add-billing")
print(f"Feature branch has {len(commits)} commits")
# 2. Search for any hardcoded secrets that slipped in
secrets_check = dev.search_code(
query="STRIPE_SECRET password API_KEY",
owner=owner,
repo=repo,
)
if secrets_check.get("total_count", 0) > 0:
dev.create_issue(
owner=owner,
repo=repo,
title="CRITICAL: Hardcoded secrets detected",
body="Automated scan found potential secrets in codebase.",
labels=["security", "P0"],
)
raise RuntimeError("Secrets detected -- aborting merge")
# 3. If clean, create PR and merge
result = dev.review_and_merge(
owner=owner,
repo=repo,
head_branch="feature/add-billing",
title="Add Stripe billing integration",
description="Adds subscription management with usage-based pricing.",
)
print(f"PR status: {result['status']}")
Pattern: Automated Issue Tracking
The agent tracks its own work with issues, creating a paper trail for every feature and bug fix.
# Agent creates a tracking issue before starting work
issue = dev.create_issue(
owner="myorg",
repo="newsletter-saas",
title="Implement subscriber management API",
body=(
"## Requirements\n"
"- POST /api/subscribers to add new subscriber\n"
"- GET /api/subscribers to list all subscribers\n"
"- DELETE /api/subscribers/:id to unsubscribe\n"
"- Store in Postgres with email uniqueness constraint\n\n"
"## Acceptance Criteria\n"
"- All endpoints return JSON\n"
"- Email validation on POST\n"
"- Idempotent DELETE\n"
),
labels=["feature", "api", "agent-created"],
)
issue_number = issue["number"]
# Agent reads the issue back to confirm requirements
details = dev.get_issue("myorg", "newsletter-saas", issue_number)
print(f"Working on: {details['title']}")
# ... agent implements the feature ...
# Agent closes the issue by referencing it in the PR body
dev.create_pull_request(
owner="myorg",
repo="newsletter-saas",
title="Implement subscriber management API",
body=f"Closes #{issue_number}\n\nAdds CRUD endpoints for subscribers.",
head="feature/subscriber-api",
)
Cost for a typical development session: 40-60 GitHub tool calls at $0.005 each = $0.20-$0.30. That covers repo creation, 10-15 file writes, 3-5 PR cycles, issue tracking, and code search.
Chapter 3: The Data Layer: Agents Managing Postgres
The AgentDBA Class
This class wraps all 6 Postgres tools. The agent uses it to design schemas, run migrations, seed data, and validate data integrity.
import requests
import json
from typing import Optional, List
class AgentDBA:
"""Postgres database administration client for the GreenHelix
A2A Commerce Gateway.
Wraps the 6 Postgres integration tools into a database management
interface for schema design, migrations, queries, and data validation.
"""
def __init__(
self,
api_key: str,
agent_id: str,
base_url: str = "https://api.greenhelix.net/v1",
):
self.base_url = base_url
self.agent_id = agent_id
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
})
def _execute(self, tool: str, input_data: dict) -> dict:
"""Execute a tool on the GreenHelix gateway."""
resp = self.session.post(
f"{self.base_url}/v1",
json={"tool": tool, "input": input_data},
)
if resp.status_code == 402:
raise BudgetExhaustedError(
f"Agent {self.agent_id} budget exceeded: {resp.text}"
)
resp.raise_for_status()
return resp.json()
# -- Schema Inspection ---------------------------------------------
def get_schema(self) -> dict:
"""Inspect the full database schema."""
return self._execute("get_schema", {})
def list_tables(self) -> dict:
"""List all tables in the database."""
return self._execute("list_tables", {})
def describe_table(self, table_name: str) -> dict:
"""Get the structure of a specific table."""
return self._execute("describe_table", {
"table_name": table_name,
})
# -- Query Execution -----------------------------------------------
def query(self, sql: str, params: Optional[List] = None) -> dict:
"""Run a SELECT query. Read-only."""
payload = {"sql": sql}
if params:
payload["params"] = params
return self._execute("query", payload)
def execute_query(
self,
sql: str,
params: Optional[List] = None,
) -> dict:
"""Run any SQL statement (INSERT, UPDATE, DELETE, DDL)."""
payload = {"sql": sql}
if params:
payload["params"] = params
return self._execute("execute_query", payload)
# -- Migrations ----------------------------------------------------
def execute_migration(
self,
migration_sql: str,
version: str,
description: str,
) -> dict:
"""Run a schema migration with version tracking."""
return self._execute("execute_migration", {
"migration_sql": migration_sql,
"version": version,
"description": description,
})
# -- High-Level Workflows ------------------------------------------
def schema_first_design(
self,
tables: dict,
) -> List[dict]:
"""Design and create a full schema from a table specification.
Args:
tables: dict mapping table names to column definitions.
Example:
{
"subscribers": [
"id SERIAL PRIMARY KEY",
"email VARCHAR(255) UNIQUE NOT NULL",
"created_at TIMESTAMPTZ DEFAULT NOW()",
]
}
"""
results = []
for table_name, columns in tables.items():
col_defs = ", ".join(columns)
sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({col_defs})"
result = self.execute_query(sql)
results.append({"table": table_name, "result": result})
return results
def safe_migration_chain(
self,
migrations: List[dict],
) -> List[dict]:
"""Execute a chain of migrations in order, stopping on failure.
Args:
migrations: list of dicts with keys:
- version: str (e.g., "001")
- description: str
- up: str (SQL to apply)
"""
results = []
for m in migrations:
try:
result = self.execute_migration(
migration_sql=m["up"],
version=m["version"],
description=m["description"],
)
results.append({
"version": m["version"],
"status": "applied",
"result": result,
})
except Exception as e:
results.append({
"version": m["version"],
"status": "failed",
"error": str(e),
})
break # Stop chain on failure
return results
def validate_data(
self,
table: str,
checks: List[dict],
) -> List[dict]:
"""Run validation checks against a table.
Args:
checks: list of dicts with keys:
- name: str (human-readable check name)
- sql: str (query that should return 0 rows if valid)
"""
results = []
for check in checks:
result = self.query(check["sql"])
row_count = len(result.get("rows", []))
results.append({
"check": check["name"],
"passed": row_count == 0,
"violations": row_count,
})
return results
Pattern: Schema-First Development
The agent designs the schema before writing application code, ensuring the data layer is correct before business logic is built on top.
import os
dba = AgentDBA(
api_key=os.environ["GREENHELIX_API_KEY"],
agent_id="dba-agent-01",
)
# Agent designs the schema for a newsletter SaaS
schema = dba.schema_first_design({
"subscribers": [
"id SERIAL PRIMARY KEY",
"email VARCHAR(255) UNIQUE NOT NULL",
"name VARCHAR(255)",
"status VARCHAR(20) DEFAULT 'active' CHECK (status IN ('active', 'unsubscribed', 'bounced'))",
"stripe_customer_id VARCHAR(255)",
"subscribed_at TIMESTAMPTZ DEFAULT NOW()",
"unsubscribed_at TIMESTAMPTZ",
],
"newsletters": [
"id SERIAL PRIMARY KEY",
"subject VARCHAR(500) NOT NULL",
"body TEXT NOT NULL",
"status VARCHAR(20) DEFAULT 'draft' CHECK (status IN ('draft', 'scheduled', 'sent'))",
"scheduled_for TIMESTAMPTZ",
"sent_at TIMESTAMPTZ",
"created_at TIMESTAMPTZ DEFAULT NOW()",
],
"send_log": [
"id SERIAL PRIMARY KEY",
"newsletter_id INTEGER REFERENCES newsletters(id)",
"subscriber_id INTEGER REFERENCES subscribers(id)",
"status VARCHAR(20) DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed', 'bounced'))",
"sent_at TIMESTAMPTZ",
"error_message TEXT",
"UNIQUE(newsletter_id, subscriber_id)",
],
})
# Verify the schema was created correctly
tables = dba.list_tables()
print(f"Tables created: {[t['table_name'] for t in tables['rows']]}")
for table in ["subscribers", "newsletters", "send_log"]:
desc = dba.describe_table(table)
print(f"\n{table}:")
for col in desc["columns"]:
print(f" {col['column_name']} {col['data_type']} {col.get('constraints', '')}")
Pattern: Safe Migration Chains
Versioned migrations tracked in order. The chain stops on the first failure to prevent partial schema states.
# Agent evolves the schema over time
migrations = dba.safe_migration_chain([
{
"version": "002",
"description": "Add subscriber tags for segmentation",
"up": """
CREATE TABLE IF NOT EXISTS subscriber_tags (
id SERIAL PRIMARY KEY,
subscriber_id INTEGER REFERENCES subscribers(id) ON DELETE CASCADE,
tag VARCHAR(100) NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(subscriber_id, tag)
);
CREATE INDEX idx_subscriber_tags_tag ON subscriber_tags(tag);
""",
},
{
"version": "003",
"description": "Add open tracking to send_log",
"up": """
ALTER TABLE send_log ADD COLUMN IF NOT EXISTS opened_at TIMESTAMPTZ;
ALTER TABLE send_log ADD COLUMN IF NOT EXISTS open_count INTEGER DEFAULT 0;
CREATE INDEX idx_send_log_opened ON send_log(opened_at)
WHERE opened_at IS NOT NULL;
""",
},
{
"version": "004",
"description": "Add revenue tracking table",
"up": """
CREATE TABLE IF NOT EXISTS revenue_events (
id SERIAL PRIMARY KEY,
subscriber_id INTEGER REFERENCES subscribers(id),
stripe_payment_intent_id VARCHAR(255) UNIQUE,
amount_cents INTEGER NOT NULL,
currency VARCHAR(3) DEFAULT 'usd',
event_type VARCHAR(50) NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_revenue_subscriber ON revenue_events(subscriber_id);
CREATE INDEX idx_revenue_created ON revenue_events(created_at);
""",
},
])
for m in migrations:
status_marker = "OK" if m["status"] == "applied" else "FAIL"
print(f" [{status_marker}] v{m['version']}: {m.get('error', 'applied')}")
Pattern: Data Validation Pipeline
After seeding data or processing a batch, the agent runs validation checks to catch integrity issues early.
# After importing subscriber data, validate
validations = dba.validate_data(
table="subscribers",
checks=[
{
"name": "No duplicate emails",
"sql": """
SELECT email, COUNT(*) as cnt FROM subscribers
GROUP BY email HAVING COUNT(*) > 1
""",
},
{
"name": "All active subscribers have email",
"sql": """
SELECT id FROM subscribers
WHERE status = 'active' AND (email IS NULL OR email = '')
""",
},
{
"name": "No future subscription dates",
"sql": """
SELECT id FROM subscribers
WHERE subscribed_at > NOW() + INTERVAL '1 minute'
""",
},
{
"name": "Unsubscribed have unsubscribed_at set",
"sql": """
SELECT id FROM subscribers
WHERE status = 'unsubscribed' AND unsubscribed_at IS NULL
""",
},
],
)
all_passed = all(v["passed"] for v in validations)
for v in validations:
marker = "PASS" if v["passed"] else f"FAIL ({v['violations']} violations)"
print(f" [{marker}] {v['check']}")
if not all_passed:
raise DataValidationError("Validation failed -- see results above")
Cost for a typical database session: 20-40 Postgres tool calls at $0.01 each = $0.20-$0.40. That covers schema creation, 3-5 migrations, data seeding, and validation.
Chapter 4: Monetization: Agents Running Stripe
The AgentBilling Class
This class wraps all 13 Stripe tools. The agent uses it to create products, set pricing, generate checkout sessions, manage invoices, and track revenue.
import requests
import json
from typing import Optional, List, Literal
class AgentBilling:
"""Stripe billing client for the GreenHelix A2A Commerce Gateway.
Wraps the 13 Stripe integration tools into a monetization interface
for product creation, pricing, checkout, invoicing, and revenue tracking.
"""
def __init__(
self,
api_key: str,
agent_id: str,
base_url: str = "https://api.greenhelix.net/v1",
):
self.base_url = base_url
self.agent_id = agent_id
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
})
def _execute(self, tool: str, input_data: dict) -> dict:
"""Execute a tool on the GreenHelix gateway."""
resp = self.session.post(
f"{self.base_url}/v1",
json={"tool": tool, "input": input_data},
)
if resp.status_code == 402:
raise BudgetExhaustedError(
f"Agent {self.agent_id} budget exceeded: {resp.text}"
)
resp.raise_for_status()
return resp.json()
# -- Customer Management -------------------------------------------
def create_customer(
self,
email: str,
name: Optional[str] = None,
metadata: Optional[dict] = None,
) -> dict:
"""Register a new Stripe customer."""
payload = {"email": email}
if name:
payload["name"] = name
if metadata:
payload["metadata"] = metadata
return self._execute("create_customer", payload)
def get_customer(self, customer_id: str) -> dict:
"""Retrieve customer details."""
return self._execute("get_customer", {
"customer_id": customer_id,
})
# -- Product & Pricing ---------------------------------------------
def create_product(
self,
name: str,
description: str,
metadata: Optional[dict] = None,
) -> dict:
"""Define a new product or service."""
payload = {"name": name, "description": description}
if metadata:
payload["metadata"] = metadata
return self._execute("create_product", payload)
def get_product(self, product_id: str) -> dict:
"""Retrieve product details."""
return self._execute("get_product", {
"product_id": product_id,
})
def create_price(
self,
product_id: str,
unit_amount: int,
currency: str = "usd",
recurring: Optional[dict] = None,
usage_type: Optional[str] = None,
) -> dict:
"""Set pricing for a product.
Args:
unit_amount: Price in cents (e.g., 2900 = $29.00)
recurring: dict with "interval" key ("month", "year")
usage_type: "metered" for usage-based billing
"""
payload = {
"product_id": product_id,
"unit_amount": unit_amount,
"currency": currency,
}
if recurring:
payload["recurring"] = recurring
if usage_type:
payload["usage_type"] = usage_type
return self._execute("create_price", payload)
# -- Checkout & Payments -------------------------------------------
def create_checkout_session(
self,
price_id: str,
success_url: str,
cancel_url: str,
mode: Literal["payment", "subscription"] = "subscription",
customer_id: Optional[str] = None,
) -> dict:
"""Generate a Stripe checkout page."""
payload = {
"price_id": price_id,
"success_url": success_url,
"cancel_url": cancel_url,
"mode": mode,
}
if customer_id:
payload["customer_id"] = customer_id
return self._execute("create_checkout_session", payload)
def get_checkout_session(self, session_id: str) -> dict:
"""Check the status of a checkout session."""
return self._execute("get_checkout_session", {
"session_id": session_id,
})
# -- Invoicing -----------------------------------------------------
def create_invoice(
self,
customer_id: str,
auto_advance: bool = True,
collection_method: str = "charge_automatically",
) -> dict:
"""Generate an invoice for a customer."""
return self._execute("create_invoice", {
"customer_id": customer_id,
"auto_advance": auto_advance,
"collection_method": collection_method,
})
def get_invoice(self, invoice_id: str) -> dict:
"""Retrieve invoice details."""
return self._execute("get_invoice", {
"invoice_id": invoice_id,
})
def list_invoices(
self,
customer_id: Optional[str] = None,
status: Optional[str] = None,
limit: int = 10,
) -> dict:
"""List invoices with optional filtering."""
payload = {"limit": limit}
if customer_id:
payload["customer_id"] = customer_id
if status:
payload["status"] = status
return self._execute("list_invoices", payload)
# -- Payment Intents -----------------------------------------------
def create_payment_intent(
self,
amount: int,
currency: str = "usd",
customer_id: Optional[str] = None,
metadata: Optional[dict] = None,
) -> dict:
"""Initiate a payment."""
payload = {"amount": amount, "currency": currency}
if customer_id:
payload["customer_id"] = customer_id
if metadata:
payload["metadata"] = metadata
return self._execute("create_payment_intent", payload)
def get_payment_intent(self, payment_intent_id: str) -> dict:
"""Check the status of a payment."""
return self._execute("get_payment_intent", {
"payment_intent_id": payment_intent_id,
})
# -- Revenue Tracking ----------------------------------------------
def get_balance(self) -> dict:
"""Check the Stripe account balance."""
return self._execute("get_balance", {})
# -- High-Level Workflows ------------------------------------------
def setup_subscription_product(
self,
name: str,
description: str,
monthly_price_cents: int,
yearly_price_cents: Optional[int] = None,
) -> dict:
"""Create a product with monthly (and optional yearly) pricing.
Returns dict with product_id, monthly_price_id, yearly_price_id.
"""
product = self.create_product(name=name, description=description)
product_id = product["id"]
monthly = self.create_price(
product_id=product_id,
unit_amount=monthly_price_cents,
recurring={"interval": "month"},
)
result = {
"product_id": product_id,
"monthly_price_id": monthly["id"],
}
if yearly_price_cents:
yearly = self.create_price(
product_id=product_id,
unit_amount=yearly_price_cents,
recurring={"interval": "year"},
)
result["yearly_price_id"] = yearly["id"]
return result
def setup_usage_based_product(
self,
name: str,
description: str,
per_unit_cents: int,
) -> dict:
"""Create a metered billing product.
Returns dict with product_id and price_id for usage reporting.
"""
product = self.create_product(name=name, description=description)
price = self.create_price(
product_id=product["id"],
unit_amount=per_unit_cents,
usage_type="metered",
recurring={"interval": "month"},
)
return {
"product_id": product["id"],
"price_id": price["id"],
}
Pattern: Subscription Lifecycle
Complete subscription setup with monthly/annual pricing and customer onboarding through checkout.
import os
billing = AgentBilling(
api_key=os.environ["GREENHELIX_API_KEY"],
agent_id="billing-agent-01",
)
# 1. Create the subscription product
product = billing.setup_subscription_product(
name="NewsletterPro",
description="Automated newsletter platform with analytics",
monthly_price_cents=2900, # $29/month
yearly_price_cents=29000, # $290/year (2 months free)
)
print(f"Product: {product['product_id']}")
print(f"Monthly: {product['monthly_price_id']}")
print(f"Yearly: {product['yearly_price_id']}")
# 2. Onboard a customer
customer = billing.create_customer(
email="alic
…(truncated)