Call the Cheerful backend API for Shopify order creation, outbound outreach, draft management, and workflow triggers. Use when the user asks to create gifting orders, add creators to campaigns for outreach, review/amend/bulk-edit email drafts, or trigger backend workflows. For creator search and list management, use the creator-search skill instead.
Use this skill for operations that must go through the Cheerful backend (order creation, outbound outreach, workflow triggers). These endpoints enforce business logic, queue population, and Shopify integration that should not be bypassed via direct DB writes.
import urllib.request, json, os
BACKEND_URL = os.environ.get('CHEERFUL_BACKEND_URL', 'https://prd-cheerful.fly.dev')
SUPABASE_URL = os.environ.get('SUPABASE_URL', '')
SERVICE_KEY = os.environ.get('SUPABASE_SERVICE_ROLE_KEY', '')
All endpoints require a Supabase user JWT. Use the admin generate_link + verify flow to get one:
def get_user_jwt(email: str) -> str:
"""Generate a JWT for a Supabase user via admin magic link + verify."""
# Step 1: Generate a magic link (gives us an OTP)
url = f"{SUPABASE_URL}/auth/v1/admin/generate_link"
req = urllib.request.Request(
url,
data=json.dumps({'type': 'magiclink', 'email': email}).encode(),
method='POST',
headers={
'apikey': SERVICE_KEY,
'Authorization': f'Bearer {SERVICE_KEY}',
'Content-Type': 'application/json'
}
)
with urllib.request.urlopen(req) as r:
link_data = json.loads(r.read())
otp = link_data['email_otp']
# Step 2: Verify the OTP to get an access token
verify_url = f"{SUPABASE_URL}/auth/v1/verify"
req = urllib.request.Request(
verify_url,
data=json.dumps({'type': 'magiclink', 'token': otp, 'email': email}).encode(),
method='POST',
headers={
'apikey': SERVICE_KEY,
'Content-Type': 'application/json'
}
)
with urllib.request.urlopen(req) as r:
tokens = json.loads(r.read())
return tokens['access_token']
Use the campaign owner's email — the order must be created under their account. Resolve from CLIENT_IDS in your CLAUDE.md if needed.
def get_user_email(user_id: str) -> str:
url = f"{SUPABASE_URL}/auth/v1/admin/users/{user_id}"
req = urllib.request.Request(url, headers={
'apikey': SERVICE_KEY,
'Authorization': f'Bearer {SERVICE_KEY}',
})
with urllib.request.urlopen(req) as r:
user = json.loads(r.read())
return user['email']
Look up all active Shopify products for a campaign's store. Uses the GoAffPro store proxy (same source as the frontend UI). Useful when you need to find a product ID or variant ID by name.
def list_shopify_products(campaign_id: str) -> list:
"""Fetch all active Shopify products for a campaign's connected store.
Returns list of dicts: {id, name, handle, vendor, product_type, variations: [{id, name, price, sku}]}
"""
# Step 1: Get the GoAffPro API key from campaign_workflow config
url = f"{SUPABASE_URL}/rest/v1/campaign_workflow?campaign_id=eq.{campaign_id}&select=config"
req = urllib.request.Request(url, headers={
'apikey': SERVICE_KEY,
'Authorization': f'Bearer {SERVICE_KEY}',
})
with urllib.request.urlopen(req) as r:
workflows = json.loads(r.read())
api_key = None
for w in workflows:
if w.get('config') and w['config'].get('goaffpro_api_key'):
api_key = w['config']['goaffpro_api_key']
break
if not api_key:
raise ValueError(f"No goaffpro_api_key found in campaign_workflow config for campaign {campaign_id}")
# Step 2: Fetch products via GoAffPro store proxy (Shopify GraphQL)
PROXY_URL = 'https://api.goaffpro.com/v1/admin/store/system/api'
PAGE_SIZE = 50
MAX_PAGES = 20
all_products = []
cursor = None
has_next = True
page = 0
while has_next and page < MAX_PAGES:
after = f', after: "{cursor}"' if cursor else ''
query = '''{
products(first: %d, query: "status:active"%s) {
edges {
node { id title handle vendor productType
variants(first: 20) { edges { node { id title price sku } } }
}
cursor
}
pageInfo { hasNextPage endCursor }
}
}''' % (PAGE_SIZE, after)
req = urllib.request.Request(
PROXY_URL,
data=json.dumps({'method': 'POST', 'url': '/graphql.json', 'body': {'query': query}}).encode(),
method='POST',
headers={
'x-goaffpro-access-token': api_key,
'Content-Type': 'application/json',
}
)
with urllib.request.urlopen(req) as r:
data = json.loads(r.read())
edges = data['result']['data']['products']['edges']
page_info = data['result']['data']['products']['pageInfo']
for edge in edges:
node = edge['node']
numeric_id = node['id'].split('/')[-1] # gid://shopify/Product/123 -> 123
all_products.append({
'id': numeric_id,
'name': node['title'],
'handle': node['handle'],
'vendor': node['vendor'],
'product_type': node.get('productType', ''),
'variations': [
{
'id': ve['node']['id'].split('/')[-1],
'name': ve['node']['title'],
'price': ve['node']['price'],
'sku': ve['node'].get('sku', ''),
}
for ve in node['variants']['edges']
],
})
has_next = page_info['hasNextPage']
cursor = page_info.get('endCursor')
page += 1
return all_products
# Example: find a product by name
products = list_shopify_products('your-campaign-uuid')
for p in products:
if 'essential blend' in p['name'].lower():
print(f"{p['name']} — Product ID: {p['id']}")
for v in p['variations']:
print(f" Variant: {v['name']} — ID: {v['id']} — £{v['price']}")
Creates a Shopify order for a creator from their completed workflow execution. After success, automatically updates campaign_creator: gifting_status → ORDERED, shopify_order_id → set, slack_approval_status → approved.
Prerequisite: The creator must have a campaign_workflow_execution with status='completed' and output_data populated. Use the cheerful-supabase skill's get_order_execution() helper to find the execution ID.
def create_order(execution_id: str, user_email: str) -> dict:
"""Create a Shopify order for a creator from their workflow execution."""
jwt = get_user_jwt(user_email)
url = f"{BACKEND_URL}/api/v1/shopify/workflow-executions/{execution_id}/orders"
req = urllib.request.Request(
url,
data=json.dumps({}).encode(),
method='POST',
headers={
'Authorization': f'Bearer {jwt}',
'Content-Type': 'application/json'
}
)
with urllib.request.urlopen(req) as r:
return json.loads(r.read())
# Example usage — use campaign owner's email (resolve from CLIENT_IDS if needed)
result = create_order('ec286204-5ded-4124-835b-1d5073751b96', '[email protected]')
print(f"Order {result['order_name']} created — ID: {result['order_id']}, Amount: {result['total_amount']} {result['currency_code']}")
Success response (HTTP 201):
{
"order_id": "gid://shopify/Order/6052407705783",
"order_name": "#687202",
"total_amount": "49.00",
"currency_code": "GBP",
"workflow_execution_id": "ec286204-..."
}
def create_orders_for_ready_creators(campaign_id: str, owner_email: str) -> list:
"""Create Shopify orders for all READY_TO_SHIP creators with completed executions."""
from cheerful_supabase import supabase_get, get_order_execution # helpers from cheerful-supabase skill
# Find all ready creators with no order yet
creators = supabase_get('campaign_creator',
f'campaign_id=eq.{campaign_id}&gifting_status=eq.READY_TO_SHIP&shopify_order_id=is.null&select=id,name,email')
results = []
for creator in creators:
execution = get_order_execution(creator['id'])
if not execution:
results.append({'creator': creator['name'], 'status': 'skipped', 'reason': 'no completed execution'})
continue
try:
order = create_order(execution['id'], owner_email)
results.append({'creator': creator['name'], 'status': 'success', 'order_name': order['order_name']})
except Exception as e:
results.append({'creator': creator['name'], 'status': 'error', 'reason': str(e)})
return results
import urllib.error