Magento Custom Actions

Use this playbook to implement Magento automations through Copilot or MCP. The installation guide covers merchant credential setup.

Setup sequence

1. Identify the business and read its existing actions and workflows. Avoid replacing unrelated customer logic or enabling demo data.
2. Read the custom-action, Python context and workflow contracts.
3. Use the connected Magento credentials at runtime. Do not ask the customer to paste them into chat or copy them into another secret.
4. Create a business-scoped Python action with clear required and optional inputs. For customer-facing order lookup, require the order email and optionally narrow by order number.
5. Test a non-matching lookup and, with an appropriate order reference, a matching lookup. Distinguish API failures from valid empty results.
6. Enable the action in a workflow, refer to its callable toolName, and test a bot conversation. Inspect its actual tool calls, not just its final text.

New Magento connections do not create legacy order configurations. Keep existing legacy configurations working. A workflow with loadingRequirements.source = legacyOrderLookup performs a built-in lookup before custom actions become available. For a new custom lookup, create a workflow without that prerequisite. Adding an action to a legacy workflow does not remove its prerequisite.

Credentials and scope

slug = (context.get("business") or {}).get("slug")
if not slug:
    slug = (context.get("conversation") or {}).get("businessSlug")
magento = (get_integration_credentials(context).get(slug) or {}).get("magento2")

The magento2 object contains storeUrl, consumerKey, consumerSecret, accessToken, accessTokenSecret, optional numeric storeId, and boolean useAccessTokenAsBearerToken.

Select the current business explicitly: this helper covers the whole organization. Never fall back to another business or expose credentials in output, metadata, errors or instructions.

URL and authentication rules

• Build the connector's endpoint as magento["storeUrl"].rstrip("/") + "/rest/V1/orders". Preserve the configured path. For example, a configured /fr prefix must remain in the request and OAuth signature; do not reduce the URL to its hostname.
• /fr is an example of a deployment-specific prefix, not a universal Magento requirement. Adobe also documents store-view scopes after /rest/, as /rest/<store_view_code>/V1/.... A numeric storeId, a language prefix and a store-view code are different things. Confirm the store's routing if the configured URL does not work; do not guess path changes silently.
• OAuth 1.0a with HMAC-SHA256 is the default. Sign the exact final URL and query parameters. Do not use HMAC-SHA1. Encode and sort parameters according to OAuth rules.
• Use Authorization: Bearer <accessToken> only when useAccessTokenAsBearerToken is true. Magento installations can disable standalone integration tokens.
• Do not follow redirects with signed credentials. Correct the configured base URL instead.
• Use finite HTTP timeouts and bounded pages within the action's 120-second budget. Use plain requests by default. If the store requires Octocom's allowlisted egress or blocks direct calls, use proxy_request consistently. The dashboard connection test uses this proxy.

These rules apply to Magento Open Source and Adobe Commerce on-premises/PaaS. Adobe Commerce as a Cloud Service uses a different endpoint layout and IMS authentication; do not assume this OAuth connector applies to it.

Read-only order lookup example

Save this as a Python action with required email and optional orderId. The code uses only Python's standard library and the provided requests package. Adapt the returned fields and policy checks to the customer's needs.

import base64
import hashlib
import hmac
import secrets
import time
from urllib.parse import quote
import requests


def magento_get(config, resource, params):
    url = config["storeUrl"].rstrip("/") + "/rest/V1/" + resource
    headers = {"Accept": "application/json"}
    if config.get("useAccessTokenAsBearerToken"):
        headers["Authorization"] = "Bearer " + config["accessToken"]
    else:
        oauth = {
            "oauth_consumer_key": config["consumerKey"],
            "oauth_token": config["accessToken"],
            "oauth_nonce": secrets.token_hex(16),
            "oauth_timestamp": str(int(time.time())),
            "oauth_signature_method": "HMAC-SHA256",
            "oauth_version": "1.0",
        }
        encode = lambda value: quote(str(value), safe="~")
        pairs = sorted((encode(k), encode(v)) for k, v in list(params.items()) + list(oauth.items()))
        normalized = "&".join(k + "=" + v for k, v in pairs)
        base = "GET&" + encode(url) + "&" + encode(normalized)
        key = encode(config["consumerSecret"]) + "&" + encode(config["accessTokenSecret"])
        oauth["oauth_signature"] = base64.b64encode(
            hmac.new(key.encode(), base.encode(), hashlib.sha256).digest()
        ).decode()
        headers["Authorization"] = "OAuth " + ", ".join(
            encode(k) + '="' + encode(v) + '"' for k, v in sorted(oauth.items())
        )
    response = requests.get(url, params=params, headers=headers, timeout=20, allow_redirects=False)
    # Replace requests.get with proxy_request("GET", ...) if this store requires it.
    if response.status_code != 200:
        return None, {"error": "Magento API access failed", "httpStatus": response.status_code}
    payload = response.json()
    if not isinstance(payload, dict) or not isinstance(payload.get("items"), list):
        return None, {"error": "Unexpected Magento API response"}
    return payload, None


def execute_action(context):
    args = context.get("args") or {}
    email = str(args.get("email") or "").strip().lower()
    if not email or "@" not in email:
        return {"error": "Ask for the email used to place the order"}
    slug = (context.get("business") or {}).get("slug") or (context.get("conversation") or {}).get("businessSlug")
    if not slug:
        return {"error": "Business context is missing"}
    config = (get_integration_credentials(context).get(slug) or {}).get("magento2")
    if not config:
        return {"error": "Magento is not connected for this business"}
    filters = [("customer_email", email)]
    if config.get("storeId") is not None:
        filters.append(("store_id", config["storeId"]))
    if args.get("orderId"):
        filters.append(("increment_id", str(args["orderId"]).strip()))
    params = {
        "searchCriteria[pageSize]": "5",
        "searchCriteria[currentPage]": "1",
        "searchCriteria[sortOrders][0][field]": "created_at",
        "searchCriteria[sortOrders][0][direction]": "DESC",
    }
    for index, (field, value) in enumerate(filters):
        prefix = "searchCriteria[filterGroups][%s][filters][0]" % index
        params[prefix + "[field]"] = field
        params[prefix + "[value]"] = str(value)
        params[prefix + "[conditionType]"] = "eq"
    try:
        payload, error = magento_get(config, "orders", params)
        if error:
            return error
        orders = [order for order in payload["items"]
                  if str(order.get("customer_email", "")).lower() == email
                  and (config.get("storeId") is None or str(order.get("store_id")) == str(config["storeId"]))]
        return {"found": bool(orders), "totalMatches": payload.get("total_count"), "orders": [
            {"orderId": order.get("increment_id"), "entityId": order.get("entity_id"),
             "status": order.get("status"), "createdAt": order.get("created_at"),
             "total": order.get("grand_total"), "currency": order.get("order_currency_code"),
             "items": [{"name": item.get("name"), "sku": item.get("sku"), "quantity": item.get("qty_ordered")}
                       for item in order.get("items", []) if not item.get("parent_item_id")]}
            for order in orders]}
    except (requests.RequestException, ValueError):
        return {"error": "Magento could not be queried. Offer a customer-service handoff."}

Order and shipment patterns

Use increment_id for the customer-visible order number. Preserve its real format, prefixes and leading zeros. Do not invent demo prefixes or assume every order number is nine digits. Pagination is required when customers have more orders than your page size; return an explicit limit or next-page indication instead of claiming the first page is complete.

For shipment queries, construct the same search criteria with field order_id and the matched order's internal ID. Handle a shipment API error separately: order lookup can succeed while shipment access is denied. Return a flag such as trackingUnavailable, and instruct the bot to say it cannot consult tracking. Never treat denied access as proof that a parcel has not shipped or that tracking has not been issued.

Do not invent delivery dates, tracking links, product images, return eligibility or modification eligibility from an order status alone. Keep totals and currency from the actual response. Monetary item prices may exclude tax; use clearly labelled tax-inclusive or tax-exclusive fields rather than presenting an exclusive-tax item price as the final customer price.

For write automations such as cancellations or refunds, consult the operation's documentation, apply the customer's authorization and eligibility rules, and test with controlled fixtures before live use. Keep these writes out of a read-only lookup.

Further API documentation

This is a starting playbook, not a copy of the Magento API reference. Consult Adobe's REST overview and endpoint structure, OAuth signing guide, search criteria guide, and the store's own API schema for operations beyond these examples.