How to Integrate AI Agents with Shopify

Shopify powers over two million merchants worldwide, and every one of those stores generates a continuous stream of events: new orders, inventory changes, customer messages, abandoned carts, and sales data. An AI agent with Shopify access can monitor all of this, make decisions, and take action — handling customer queries instantly, alerting you before stockouts happen, and surfacing sales insights that would take hours to compile manually.

This guide covers the complete integration: Custom App setup, Admin API tooling with Python, four production-ready use cases, and Shopify webhook patterns for event-driven agents.

What AI Agents Can Do with Shopify#

Before writing any code, map the full scope of what becomes automatable once an agent has API access:

Order management

• Look up order status and tracking information for customer queries
• Process refunds and create draft orders
• Flag high-risk orders for fraud review
• Generate order confirmation or shipping delay notifications

Inventory operations

• Monitor stock levels and trigger reorder alerts
• Update inventory counts across locations
• Identify products approaching stockout based on sales velocity
• Create purchase orders to suppliers when thresholds are crossed

Customer data

• Retrieve customer order history for personalized service
• Update customer tags for segmentation
• Create customer notes from support interactions
• Identify high-LTV customers for VIP treatment

Products and pricing

• Read product details for Q&A (sizing, compatibility, specs)
• Update product descriptions or metafields
• Create and manage discount codes
• Analyze product performance metrics

Analytics and reporting

• Pull sales data for period-over-period analysis
• Identify top products, slow movers, and seasonal trends
• Generate automated weekly/monthly sales summaries
• Calculate conversion metrics from funnel data

---

Step 1: Create a Shopify Custom App#

Custom Apps are the right authentication method for AI agents — they're store-specific, use permanent access tokens, and let you scope permissions precisely.

Create the App in Shopify Admin#

1. Go to your Shopify Admin → Settings → Apps and sales channels → Develop apps
2. Click Create an app and give it a descriptive name (e.g., "AI Agent - Ops")
3. Click Configure Admin API scopes and select only what you need

Recommended scopes by use case:

| Use Case | Scopes | |----------|--------| | Customer service | read_orders, read_customers, write_draft_orders | | Inventory monitoring | read_inventory, write_inventory, read_products | | Sales analytics | read_orders, read_products, read_analytics | | Abandoned cart recovery | read_orders, write_customers, read_checkouts | | Discount management | write_price_rules, write_discounts, read_customers |

4. Click Save, then go to API credentials → Install app
5. Copy the Admin API access token — it's shown only once

export SHOPIFY_STORE_URL="your-store.myshopify.com"
export SHOPIFY_ACCESS_TOKEN="shpat_your_token_here"

---

Step 2: Build Shopify Tools for LangChain#

Install the Shopify Python library:

pip install ShopifyAPI langchain langchain-openai python-dotenv requests

import shopify
import requests
from langchain.tools import tool
from dotenv import load_dotenv
import os
import json
from datetime import datetime, timedelta

load_dotenv()

# Initialize Shopify session
STORE_URL = os.getenv("SHOPIFY_STORE_URL")
ACCESS_TOKEN = os.getenv("SHOPIFY_ACCESS_TOKEN")
API_VERSION = "2025-01"

shopify.ShopifyResource.set_site(
    f"https://{STORE_URL}/admin/api/{API_VERSION}"
)
shopify.ShopifyResource.set_headers({"X-Shopify-Access-Token": ACCESS_TOKEN})


@tool
def get_order_by_name(order_name: str) -> str:
    """
    Look up a Shopify order by order number (e.g., '#1042' or '1042').
    Returns order status, items, fulfillment tracking, and customer details.
    """
    # Normalize order name format
    order_name = order_name.strip().lstrip("#")
    orders = shopify.Order.find(name=f"#{order_name}", status="any")

    if not orders:
        return f"No order found with number #{order_name}."

    order = orders[0]
    items = [f"{li.quantity}x {li.title}" for li in order.line_items]

    fulfillments = []
    for f in order.fulfillments:
        tracking = f.tracking_number or "No tracking number"
        fulfillments.append(f"Tracking: {tracking} (Status: {f.status})")

    return (
        f"Order #{order.name}:\n"
        f"Status: {order.financial_status} / {order.fulfillment_status or 'unfulfilled'}\n"
        f"Items: {', '.join(items)}\n"
        f"Total: {order.currency} {order.total_price}\n"
        f"Customer: {order.email}\n"
        f"Fulfillment: {'; '.join(fulfillments) if fulfillments else 'Not yet fulfilled'}"
    )


@tool
def get_customer_order_history(customer_email: str) -> str:
    """
    Retrieve the order history for a customer by email address.
    Returns the last 5 orders with status and totals.
    """
    customers = shopify.Customer.find(email=customer_email)

    if not customers:
        return f"No customer found with email {customer_email}."

    customer = customers[0]
    orders = shopify.Order.find(customer_id=customer.id, status="any", limit=5)

    if not orders:
        return f"Customer {customer.first_name} {customer.last_name} has no orders."

    order_list = []
    for o in orders:
        order_list.append(
            f"#{o.name} ({o.created_at[:10]}): {o.currency} {o.total_price} — {o.financial_status}"
        )

    lifetime_value = float(customer.total_spent)
    order_count = customer.orders_count

    return (
        f"Customer: {customer.first_name} {customer.last_name} ({customer_email})\n"
        f"Lifetime value: {customer.currency} {lifetime_value:.2f} ({order_count} orders)\n"
        f"Recent orders:\n" + "\n".join(order_list)
    )


@tool
def check_inventory_levels(product_id: str = "", low_stock_threshold: int = 10) -> str:
    """
    Check current inventory levels.
    If product_id is provided, check that specific product.
    Otherwise, return all products with inventory below low_stock_threshold.
    """
    if product_id:
        product = shopify.Product.find(product_id)
        variants_info = []
        for variant in product.variants:
            inv = shopify.InventoryLevel.find(inventory_item_ids=variant.inventory_item_id)
            for level in inv:
                variants_info.append(
                    f"  {variant.title}: {level.available} units (Location: {level.location_id})"
                )
        return f"Inventory for {product.title}:\n" + "\n".join(variants_info)
    else:
        # Scan all products for low stock
        products = shopify.Product.find(limit=250, status="active")
        low_stock_items = []

        for product in products:
            for variant in product.variants:
                if variant.inventory_management == "shopify":
                    inv = shopify.InventoryLevel.find(
                        inventory_item_ids=variant.inventory_item_id
                    )
                    for level in inv:
                        if level.available is not None and level.available <= low_stock_threshold:
                            low_stock_items.append(
                                f"{product.title} — {variant.title}: {level.available} units remaining"
                            )

        if not low_stock_items:
            return f"No products below {low_stock_threshold} units."

        return f"Products below {low_stock_threshold} units ({len(low_stock_items)} items):\n" + "\n".join(low_stock_items)


@tool
def create_discount_code(
    title: str,
    code: str,
    discount_type: str,
    value: float,
    customer_email: str = "",
    expiry_days: int = 30
) -> str:
    """
    Create a Shopify discount code.
    discount_type: 'percentage' or 'fixed_amount'
    value: percentage (0-100) or fixed amount in store currency
    customer_email: if provided, restricts code to this customer only
    """
    from datetime import timezone

    ends_at = (datetime.now(timezone.utc) + timedelta(days=expiry_days)).isoformat()

    price_rule_data = {
        "title": title,
        "value_type": discount_type,
        "value": f"-{value}",
        "customer_selection": "specific_customers" if customer_email else "all",
        "target_type": "line_item",
        "target_selection": "all",
        "allocation_method": "across",
        "starts_at": datetime.now(timezone.utc).isoformat(),
        "ends_at": ends_at,
        "usage_limit": 1 if customer_email else 100,
        "once_per_customer": True
    }

    # Create price rule
    price_rule = shopify.PriceRule()
    for key, val in price_rule_data.items():
        setattr(price_rule, key, val)
    price_rule.save()

    # Create discount code for the price rule
    discount = shopify.DiscountCode()
    discount.code = code
    discount.prefix("/admin/api/{API_VERSION}/price_rules/{price_rule.id}/discount_codes.json")
    discount.save()

    discount_label = f"{value}% off" if discount_type == "percentage" else f"{value} off"
    expiry_label = f"(expires {ends_at[:10]})"

    return (
        f"Discount code created: {code} — {discount_label} {expiry_label}\n"
        f"{'Restricted to: ' + customer_email if customer_email else 'Available to all customers'}"
    )


@tool
def get_sales_summary(days: int = 7) -> str:
    """
    Generate a sales summary for the last N days.
    Returns total revenue, order count, top products, and average order value.
    """
    from datetime import timezone

    since_date = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    orders = shopify.Order.find(
        status="any",
        financial_status="paid",
        created_at_min=since_date,
        limit=250
    )

    if not orders:
        return f"No paid orders in the last {days} days."

    total_revenue = sum(float(o.total_price) for o in orders)
    order_count = len(orders)
    aov = total_revenue / order_count

    # Count items sold per product
    product_counts: dict[str, int] = {}
    for order in orders:
        for item in order.line_items:
            product_counts[item.title] = product_counts.get(item.title, 0) + item.quantity

    top_products = sorted(product_counts.items(), key=lambda x: x[1], reverse=True)[:5]
    top_products_str = "\n".join(f"  {title}: {qty} units" for title, qty in top_products)

    currency = orders[0].currency if orders else "USD"

    return (
        f"Sales Summary — Last {days} days:\n"
        f"Revenue: {currency} {total_revenue:,.2f}\n"
        f"Orders: {order_count}\n"
        f"Average order value: {currency} {aov:.2f}\n"
        f"Top products:\n{top_products_str}"
    )

---

Use Case 1: Customer Service Agent — Order Status Queries#

This is the highest-ROI use case for most Shopify stores. An agent handles order status questions, tracking inquiries, and basic support without human involvement.

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [get_order_by_name, get_customer_order_history, create_discount_code]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a customer service agent for a Shopify store.

    When customers ask about orders:
    1. Look up their order using the order number they provide
    2. Give a clear, friendly status update including tracking info if available
    3. If the order is significantly delayed (>5 days since purchase, not yet fulfilled), offer a 10% discount code as an apology

    When customers ask about their history:
    1. Look up their account by email
    2. Summarize their order history warmly

    Tone: Friendly, professional, concise.
    Never make up order information — only share what the API returns.
    If you cannot find an order, ask the customer to verify their order number."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Simulate customer message
result = executor.invoke({
    "input": "Hi! I placed order #2847 five days ago and haven't received a shipping notification yet. Can you check on it?"
})
print(result["output"])

---

Use Case 2: Inventory Monitoring and Reorder Alert Agent#

This agent runs on a schedule (daily or twice daily) to identify impending stockouts and draft purchase order notifications.

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

inventory_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an inventory management agent for a Shopify store.

    Your daily task:
    1. Check all inventory levels and find items below 15 units
    2. For critical items (below 5 units), flag as URGENT
    3. Generate a structured reorder report with:
       - Product name and variant
       - Current stock level
       - Priority: URGENT (<5) or LOW STOCK (5-15)
       - Suggested reorder quantity (3x current weekly sales velocity estimate)
    4. End with an action summary

    Be specific and actionable — operations teams use this report directly."""),
    ("human", "Run the daily inventory check."),
    ("placeholder", "{agent_scratchpad}"),
])

inventory_agent = AgentExecutor(
    agent=create_tool_calling_agent(llm, [check_inventory_levels, get_sales_summary], inventory_prompt),
    tools=[check_inventory_levels, get_sales_summary],
    verbose=True
)

report = inventory_agent.invoke({"input": "Run the daily inventory check."})
print(report["output"])

---

Use Case 3: Sales Analytics and Reporting Agent#

This agent pulls sales data, identifies trends, and generates a structured weekly business summary.

analytics_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a sales analytics agent. Generate a weekly performance report.

    Structure your report as:
    ## Weekly Sales Report — [date range]

    ### Key Metrics
    - Total revenue vs previous week
    - Order count vs previous week
    - Average order value trend

    ### Top Performing Products
    [List with units sold]

    ### Insights
    [2-3 specific, actionable observations]

    ### Recommendations
    [1-2 concrete actions based on the data]

    Use the sales summary tool for both this week (7 days) and last week (7-14 days ago)."""),
    ("human", "Generate this week's sales report."),
    ("placeholder", "{agent_scratchpad}"),
])

analytics_agent = AgentExecutor(
    agent=create_tool_calling_agent(llm, [get_sales_summary], analytics_prompt),
    tools=[get_sales_summary],
    verbose=True
)

weekly_report = analytics_agent.invoke({"input": "Generate this week's sales report."})

---

Use Case 4: Abandoned Cart Recovery Agent (CrewAI)#

For multi-step abandoned cart recovery, CrewAI's crew pattern works well: one agent analyzes the cart value and customer history, a second drafts the personalized recovery email, and a third decides whether to include a discount code.

from crewai import Agent, Task, Crew

# Reuse the Shopify tools defined earlier
analyzer = Agent(
    role="Cart Analyst",
    goal="Analyze abandoned cart value and customer history to determine recovery strategy",
    backstory="Expert at evaluating customer lifetime value and cart abandonment patterns",
    tools=[get_customer_order_history, get_order_by_name],
    llm="gpt-4o"
)

copywriter = Agent(
    role="Email Copywriter",
    goal="Write personalized, conversion-focused recovery emails",
    backstory="E-commerce copywriter specializing in recovery and retention emails",
    tools=[],
    llm="gpt-4o"
)

retention_agent = Agent(
    role="Retention Manager",
    goal="Decide on discount offers and create codes for high-value recovery scenarios",
    backstory="Responsible for balancing retention cost against customer lifetime value",
    tools=[create_discount_code],
    llm="gpt-4o"
)

# Define tasks
analyze_task = Task(
    description="""Analyze this abandoned cart scenario:
    Customer: {customer_email}
    Cart value: {cart_value}
    Items: {cart_items}

    Check their order history. Determine: Is this a new or returning customer?
    What is their LTV? Is this cart worth offering a discount?""",
    agent=analyzer,
    expected_output="Customer segment, LTV, and recovery strategy recommendation"
)

email_task = Task(
    description="Write a personalized recovery email based on the analyst's findings. Be genuine, not pushy. Mention the specific items left behind.",
    agent=copywriter,
    expected_output="Subject line and email body for the recovery campaign"
)

discount_task = Task(
    description="If the analyst recommended a discount, create a unique 10% discount code for this customer. Report the code to include in the email.",
    agent=retention_agent,
    expected_output="Discount code (if applicable) and final email ready to send"
)

cart_recovery_crew = Crew(
    agents=[analyzer, copywriter, retention_agent],
    tasks=[analyze_task, email_task, discount_task],
    verbose=True
)

result = cart_recovery_crew.kickoff(inputs={
    "customer_email": "customer@example.com",
    "cart_value": "89.00",
    "cart_items": "Blue Merino Wool Sweater (Large), Leather Belt"
})

---

Shopify Webhooks: Event-Driven Agent Triggers#

Polling is inefficient for time-sensitive workflows. Shopify webhooks push events to your agent endpoint the moment they happen.

Register a Webhook in Shopify Admin#

Go to Settings → Notifications → Webhooks → Create webhook. Choose your event (e.g., orders/created) and point it to your agent's public endpoint (a FastAPI or Flask server, or a cloud function URL).

from fastapi import FastAPI, Request
import hmac, hashlib, base64

app = FastAPI()
SHOPIFY_WEBHOOK_SECRET = os.getenv("SHOPIFY_WEBHOOK_SECRET")

@app.post("/webhooks/shopify/orders-created")
async def handle_new_order(request: Request):
    # Verify webhook authenticity
    hmac_header = request.headers.get("X-Shopify-Hmac-Sha256", "")
    body = await request.body()
    digest = hmac.new(
        SHOPIFY_WEBHOOK_SECRET.encode(),
        body,
        hashlib.sha256
    ).digest()
    computed = base64.b64encode(digest).decode()

    if not hmac.compare_digest(computed, hmac_header):
        return {"status": "unauthorized"}, 401

    # Parse order and trigger agent asynchronously
    order_data = await request.json()
    order_id = order_data["id"]
    order_total = float(order_data["total_price"])

    # Trigger agent for high-value orders (e.g., >$200 gets VIP treatment)
    if order_total > 200:
        # Queue agent run — use Celery, background task, or cloud function
        print(f"VIP order detected: #{order_data['order_number']} (${order_total})")

    return {"status": "received"}

Key webhook events for agents:

• orders/created — Trigger fulfillment checks or VIP notifications
• orders/cancelled — Trigger recovery workflows
• inventory_levels/update — Trigger reorder alerts in real time
• checkouts/create — Start abandoned cart timer
• customers/create — Trigger onboarding sequences

---

API Rate Limits Reference#

| API | Limit | Notes | |-----|-------|-------| | REST Admin API | 2 req/sec (40 bucket) | Per store, per app | | GraphQL Admin API | 1,000 points/bucket, 50/sec refill | Query cost varies | | Webhooks | No limit on receiving | Shopify retries failed deliveries | | Bulk Operations API | 1 concurrent job | For large data exports |

For agents making frequent inventory or order calls, implement a simple token bucket rate limiter and cache product/customer data that changes infrequently.

---

Next Steps#

Your Shopify agent is the foundation of a fully automated e-commerce operations layer. Extend it with these resources:

• What Are AI Agents — Core concepts behind the agents powering your Shopify automation
• Build an AI Agent with LangChain — Full LangChain setup if you're new to the framework
• Build Multi-Agent Systems with CrewAI — Deploy the cart recovery crew pattern and others
• AI Agents in Business Examples — How e-commerce companies are deploying agents in production
• AI Agent Workflow Examples — Operational workflow patterns adaptable to Shopify automation
• AI Agents Salesforce Integration — Connect Shopify agent outputs to your CRM for full customer journey visibility
• Integration Templates — Pre-built blueprints for Shopify agent workflow patterns