MuleSoft Anypoint Integration

Build API-led connectivity flows in MuleSoft Anypoint Platform that transform enterprise events into blockchain-anchored Certyo records using DataWeave and Mule 4 flows.

Overview

MuleSoft Anypoint Platform is the industry-leading integration platform for API-led connectivity. This guide shows how to position Certyo as a System API within MuleSoft's three-layer architecture, enabling any enterprise system to anchor records on the blockchain through a standardized process layer.

The integration uses Mule 4's HTTP Connector and DataWeave 2.0 to transform payloads from your source systems into the Certyo record format, then calls POST /api/v1/records asynchronously.

Prerequisites

MuleSoft Anypoint Platform account (Trial or Enterprise)
Anypoint Studio 7.x IDE installed locally
Certyo API key (obtain from the Quickstart guide)
Java 8 or 11 (required by Anypoint Studio)
Access to your source ERP/CRM system APIs

Architecture

MuleSoft's API-led connectivity pattern organizes integrations into three layers. Certyo sits at the System API layer, providing blockchain anchoring as a reusable service.

API-Led Connectivity Architecturebash

┌─────────────────────────────────────────────────────────────────┐
│                     EXPERIENCE LAYER                            │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐  │
│  │ Mobile App   │  │  Web Portal  │  │  Partner Dashboard   │  │
│  └──────┬───────┘  └──────┬───────┘  └──────────┬───────────┘  │
│         └─────────────────┼─────────────────────┘              │
├───────────────────────────┼─────────────────────────────────────┤
│                     PROCESS LAYER                               │
│  ┌────────────────────────┴────────────────────────────────┐   │
│  │          Certyo Record Ingestion Process API            │   │
│  │  ┌──────────────────────────────────────────────────┐   │   │
│  │  │  1. Receive event from Experience / System API   │   │   │
│  │  │  2. DataWeave: transform to Certyo record format │   │   │
│  │  │  3. Enrich with tenant metadata                  │   │   │
│  │  │  4. Route to Certyo System API                   │   │   │
│  │  │  5. Return record hash to caller                 │   │   │
│  │  └──────────────────────────────────────────────────┘   │   │
│  └────────────────────────┬────────────────────────────────┘   │
├───────────────────────────┼─────────────────────────────────────┤
│                     SYSTEM LAYER                                │
│  ┌─────────────┐  ┌──────┴──────┐  ┌──────────────────────┐   │
│  │  SAP S/4    │  │   Certyo    │  │  Salesforce CRM      │   │
│  │  System API │  │  System API │  │  System API          │   │
│  │             │  │  (Records)  │  │                      │   │
│  └─────────────┘  └─────────────┘  └──────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

Data Flow:
  Source System ──► MuleSoft Process API ──► Certyo System API
                    (DataWeave transform)    (POST /api/v1/records)
                                                    │
                                             202 Accepted + hash
                                                    │
                                             Kafka ──► Accumulate
                                                ──► Merkle ──► IPFS
                                                ──► Polygon anchor

API-Led Connectivity Layers

System APIs — Direct connections to backend systems. Certyo exposes a System API for record ingestion (POST /api/v1/records) and verification (POST /api/v1/verify/record). Your ERP, CRM, and database systems each have their own System APIs.
Process APIs — Orchestrate business logic across System APIs. The Certyo Process API transforms source payloads into the Certyo record format, enriches with tenant metadata, handles errors, and routes to the correct endpoint.
Experience APIs — Tailored for specific consumers (mobile, web, partners). These call the Process API without needing to know about Certyo's record format.

Mule 4 Flow Configuration

The following Mule 4 flow receives an HTTP POST, transforms the payload with DataWeave, and calls the Certyo API. This is the core of your Process API implementation.

mule-certyo-flow.xml — Mule 4 Flowbash

<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns="http://www.mulesoft.org/schema/mule/core"
      xmlns:http="http://www.mulesoft.org/schema/mule/http"
      xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core"
      xmlns:secure-properties="http://www.mulesoft.org/schema/mule/secure-properties"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="
        http://www.mulesoft.org/schema/mule/core
        http://www.mulesoft.org/schema/mule/core/current/mule.xsd
        http://www.mulesoft.org/schema/mule/http
        http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
        http://www.mulesoft.org/schema/mule/ee/core
        http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd
        http://www.mulesoft.org/schema/mule/secure-properties
        http://www.mulesoft.org/schema/mule/secure-properties/current/mule-secure-properties.xsd">

    <!-- Load encrypted properties (API keys, credentials) -->
    <secure-properties:config name="secureConfig"
        file="secure-${env}.yaml"
        key="${secure.key}" />

    <!-- Certyo API HTTP configuration -->
    <http:request-config name="certyo-api-config">
        <http:request-connection
            host="www.certyos.com"
            port="443"
            protocol="HTTPS">
            <http:default-headers>
                <http:default-header
                    key="X-API-Key"
                    value="${secure::certyo.api.key}" />
                <http:default-header
                    key="Content-Type"
                    value="application/json" />
            </http:default-headers>
        </http:request-connection>
    </http:request-config>

    <!-- Inbound HTTP listener for Process API -->
    <http:listener-config name="process-api-listener">
        <http:listener-connection host="0.0.0.0" port="8081" />
    </http:listener-config>

    <!-- Main flow: Receive event, transform, send to Certyo -->
    <flow name="certyo-record-ingestion-flow">
        <http:listener config-ref="process-api-listener"
            path="/api/certyo/ingest"
            method="POST" />

        <!-- Transform source payload to Certyo record format -->
        <ee:transform>
            <ee:message>
                <ee:set-payload><![CDATA[%dw 2.0
output application/json
---
{
    tenantId: vars.tenantId default p('certyo.tenant.id'),
    database: payload.sourceSystem default "erp",
    collection: payload.entityType default "records",
    recordId: payload.id as String,
    recordVersion: payload.version default "1",
    operationType: payload.operation default "upsert",
    recordPayload: payload.data,
    sourceTimestamp: payload.timestamp
        default now() as String {format: "yyyy-MM-dd'T'HH:mm:ss'Z'"},
    idempotencyKey: payload.id ++ "-" ++ (payload.version default "1")
        ++ "-" ++ now() as String {format: "yyyy-MM-dd"}
}]]></ee:set-payload>
            </ee:message>
        </ee:transform>

        <!-- Call Certyo Records API -->
        <http:request config-ref="certyo-api-config"
            method="POST"
            path="/api/v1/records" />

        <!-- Log the accepted response -->
        <logger level="INFO"
            message="Record ingested: #[payload.recordHash]" />
    </flow>
</mule>

DataWeave Transformation

DataWeave 2.0 is MuleSoft's expression language for transforming data between formats. Below are examples of transforming common ERP payloads into the Certyo record format, plus an equivalent Python script for teams that prefer to test transformations outside Anypoint Studio.

certyo-transform.dwl — ERP Order to Certyo Recordjavascript

%dw 2.0
output application/json

// Map ERP-specific field names to Certyo's canonical record format.
// This transform handles SAP, D365, and generic ERP payloads.

var sourceSystem = payload.sourceSystem default "unknown"
var entityMap = {
    "SAP":  { db: "sap-s4", collection: "material-documents" },
    "D365": { db: "dynamics-365", collection: "sales-orders" },
    "ERP":  { db: payload.databaseName default "erp", collection: payload.tableName default "records" }
}
var mapping = entityMap[sourceSystem] default entityMap["ERP"]

---
{
    tenantId: p('certyo.tenant.id'),
    database: mapping.db,
    collection: mapping.collection,
    recordId: payload.documentNumber as String,
    recordVersion: payload.version default "1",
    operationType: if (payload.isNew == true) "insert"
                   else if (payload.isDeleted == true) "delete"
                   else "upsert",
    recordPayload: {
        documentNumber: payload.documentNumber,
        documentDate: payload.documentDate,
        (plant: payload.plant) if sourceSystem == "SAP",
        (businessUnit: payload.businessUnit) if sourceSystem == "D365",
        lineItems: payload.items map (item) -> {
            itemId: item.itemNumber as String,
            material: item.materialCode default item.productId,
            quantity: item.quantity as Number,
            unit: item.unitOfMeasure default "EA",
            amount: item.netAmount as Number default 0
        },
        totalAmount: sum(payload.items.netAmount default [0]),
        currency: payload.currency default "USD"
    },
    sourceTimestamp: payload.changedAt
        default payload.createdAt
        default now() as String {format: "yyyy-MM-dd'T'HH:mm:ss'Z'"},
    idempotencyKey: payload.documentNumber ++ "-"
        ++ (payload.version default "1") ++ "-"
        ++ now() as String {format: "yyyy-MM-dd"}
}

certyo_transform.py — Equivalent Python logicpython

"""
Certyo record transformer — equivalent to the DataWeave 2.0 script.
Use this for testing transformations outside Anypoint Studio
or as a standalone microservice alternative.
"""
import os
import json
from datetime import datetime, timezone
from typing import Any

TENANT_ID = os.environ.get("CERTYO_TENANT_ID", "acme-corp")

ENTITY_MAP = {
    "SAP":  {"db": "sap-s4", "collection": "material-documents"},
    "D365": {"db": "dynamics-365", "collection": "sales-orders"},
}


def transform_to_certyo_record(payload: dict[str, Any]) -> dict[str, Any]:
    """Transform an ERP payload into a Certyo record."""
    source = payload.get("sourceSystem", "ERP")
    mapping = ENTITY_MAP.get(source, {
        "db": payload.get("databaseName", "erp"),
        "collection": payload.get("tableName", "records"),
    })

    items = payload.get("items", [])
    line_items = [
        {
            "itemId": str(item.get("itemNumber", "")),
            "material": item.get("materialCode") or item.get("productId", ""),
            "quantity": float(item.get("quantity", 0)),
            "unit": item.get("unitOfMeasure", "EA"),
            "amount": float(item.get("netAmount", 0)),
        }
        for item in items
    ]

    record_payload: dict[str, Any] = {
        "documentNumber": payload["documentNumber"],
        "documentDate": payload.get("documentDate"),
        "lineItems": line_items,
        "totalAmount": sum(i["amount"] for i in line_items),
        "currency": payload.get("currency", "USD"),
    }

    # Add source-specific fields
    if source == "SAP" and "plant" in payload:
        record_payload["plant"] = payload["plant"]
    if source == "D365" and "businessUnit" in payload:
        record_payload["businessUnit"] = payload["businessUnit"]

    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    version = payload.get("version", "1")
    doc_number = str(payload["documentNumber"])

    op = "upsert"
    if payload.get("isNew"):
        op = "insert"
    elif payload.get("isDeleted"):
        op = "delete"

    return {
        "tenantId": TENANT_ID,
        "database": mapping["db"],
        "collection": mapping["collection"],
        "recordId": doc_number,
        "recordVersion": version,
        "operationType": op,
        "recordPayload": record_payload,
        "sourceTimestamp": payload.get("changedAt")
            or payload.get("createdAt")
            or now,
        "idempotencyKey": f"{doc_number}-{version}-{today}",
    }


if __name__ == "__main__":
    sample = {
        "sourceSystem": "SAP",
        "documentNumber": "4900012345",
        "documentDate": "2026-04-14",
        "version": "1",
        "plant": "1000",
        "currency": "EUR",
        "items": [
            {"itemNumber": 10, "materialCode": "MAT-001",
             "quantity": 100, "unitOfMeasure": "KG", "netAmount": 5000.00},
            {"itemNumber": 20, "materialCode": "MAT-002",
             "quantity": 50, "unitOfMeasure": "EA", "netAmount": 2500.00},
        ],
    }
    print(json.dumps(transform_to_certyo_record(sample), indent=2))

Test the MuleSoft Process API endpointbash

# Send a test event to your MuleSoft Process API
# (running locally on port 8081)

curl -X POST http://localhost:8081/api/certyo/ingest \
  -H "Content-Type: application/json" \
  -d '{
    "sourceSystem": "SAP",
    "documentNumber": "4900012345",
    "documentDate": "2026-04-14",
    "version": "1",
    "plant": "1000",
    "currency": "EUR",
    "isNew": true,
    "items": [
      {
        "itemNumber": 10,
        "materialCode": "MAT-001",
        "quantity": 100,
        "unitOfMeasure": "KG",
        "netAmount": 5000.00
      },
      {
        "itemNumber": 20,
        "materialCode": "MAT-002",
        "quantity": 50,
        "unitOfMeasure": "EA",
        "netAmount": 2500.00
      }
    ]
  }'

# Expected response (proxied from Certyo):
# {
#   "recordId": "4900012345",
#   "recordHash": "a1b2c3d4...",
#   "tenantId": "acme-corp",
#   "acceptedAt": "2026-04-14T10:30:00.000Z",
#   "idempotencyReplayed": false
# }

Error Handling

Production Mule flows must handle transient failures (network timeouts, 429 rate limits) and permanent failures (400 bad request, 401 unauthorized) differently. The flow below adds retry logic and a dead letter queue for unrecoverable errors.

Error handling with retry and dead letter queuebash

<!-- Add inside <flow name="certyo-record-ingestion-flow"> -->
<error-handler>
    <!-- Retry on transient errors: 429, 500, 502, 503, 504, timeouts -->
    <on-error-continue
        type="HTTP:TIMEOUT, HTTP:CONNECTIVITY, HTTP:RETRY_EXHAUSTED"
        enableNotifications="true">
        <until-successful
            maxRetries="3"
            millisBetweenRetries="2000">
            <http:request config-ref="certyo-api-config"
                method="POST"
                path="/api/v1/records" />
        </until-successful>
    </on-error-continue>

    <!-- Handle rate limiting (429) with exponential backoff -->
    <on-error-continue
        type="HTTP:TOO_MANY_REQUESTS"
        enableNotifications="true">
        <set-variable variableName="retryAfter"
            value="#[attributes.headers.'Retry-After' default '5']" />
        <logger level="WARN"
            message="Rate limited. Retry after #[vars.retryAfter]s" />
        <!-- Wait and retry -->
        <until-successful
            maxRetries="5"
            millisBetweenRetries="#[vars.retryAfter as Number * 1000]">
            <http:request config-ref="certyo-api-config"
                method="POST"
                path="/api/v1/records" />
        </until-successful>
    </on-error-continue>

    <!-- Dead letter queue for permanent failures -->
    <on-error-propagate
        type="HTTP:BAD_REQUEST, HTTP:UNAUTHORIZED, HTTP:FORBIDDEN">
        <logger level="ERROR"
            message="Permanent failure: #[error.description]" />
        <ee:transform>
            <ee:message>
                <ee:set-payload><![CDATA[%dw 2.0
output application/json
---
{
    originalPayload: vars.originalPayload,
    error: error.description,
    errorType: error.errorType.identifier,
    timestamp: now() as String {format: "yyyy-MM-dd'T'HH:mm:ss'Z'"},
    flowName: flow.name
}]]></ee:set-payload>
            </ee:message>
        </ee:transform>
        <!-- Publish to dead letter queue for manual review -->
        <jms:publish config-ref="activemq-config"
            destination="certyo.dlq" />
    </on-error-propagate>
</error-handler>

Authentication

The integration uses two sets of credentials:

MuleSoft Anypoint Connected App — For deploying and managing your Mule applications via the Anypoint Platform APIs (CI/CD). Configured in your Anypoint Platform account under Access Management > Connected Apps.
Certyo API Key — Passed as the X-API-Key header on every request to Certyo. Store this in Mule's Secure Properties file, never in plain text.

Secure Properties Configuration

secure-dev.yaml — Encrypted properties filebash

# Generate an encrypted value using Anypoint Studio's
# Secure Properties Tool (or the CLI):
#
#   mule-credential-vault-cli encrypt \
#     --algorithm AES --mode CBC \
#     --key "${MULE_SECURE_KEY}" \
#     --value "your-certyo-api-key-here"

certyo:
  api:
    key: "![encryptedValue]"
  tenant:
    id: "acme-corp"

# Reference in your flow XML:
#   ${secure::certyo.api.key}
#   ${secure::certyo.tenant.id}

Custom MuleSoft Connector

For teams that want a reusable, shareable connector (instead of raw HTTP requests), you can build a custom MuleSoft connector using the Mule SDK. This packages the Certyo API as a first-class Mule component with typed operations and autocomplete in Anypoint Studio.

Connector Overview

SDK: Mule SDK (Java-based), available via org.mule.sdk:mule-sdk-api Maven dependency
Operations: ingestRecord, bulkIngest, verifyRecord, queryRecords
Connection Provider: Handles the X-API-Key header and base URL configuration. Supports Mule's connection testing and reconnection strategies.
Distribution: Publish to Anypoint Exchange as a custom asset, making it available to all developers in your organization.

Scaffold a Mule connector projectbash

# Generate the connector skeleton using the Mule SDK archetype
mvn archetype:generate \
  -DarchetypeGroupId=org.mule.extensions \
  -DarchetypeArtifactId=mule-extensions-archetype \
  -DarchetypeVersion=1.4.0 \
  -DgroupId=com.certyos \
  -DartifactId=certyo-mule-connector \
  -Dversion=1.0.0-SNAPSHOT \
  -Dpackage=com.certyos.mule.connector

# Key files to implement:
#   src/main/java/.../CertyoConnection.java        — API key + base URL
#   src/main/java/.../CertyoConnectionProvider.java — connection lifecycle
#   src/main/java/.../CertyoOperations.java         — ingest, verify, query
#   src/main/java/.../CertyoExtension.java          — @Extension metadata

# Deploy to Anypoint Exchange:
mvn deploy -DaltDeploymentRepository=anypoint-exchange::default::https://maven.anypoint.mulesoft.com/api/v3/organizations/${orgId}/maven

Bulk Ingestion with Batch Job

For high-volume scenarios (e.g., nightly ERP data exports), use MuleSoft's Batch Job component to process thousands of records efficiently via POST /api/v1/records/bulk.

Batch job flow for bulk ingestionbash

<flow name="certyo-bulk-ingestion-flow">
    <!-- Trigger: scheduled or event-driven -->
    <scheduler>
        <scheduling-strategy>
            <cron expression="0 0 2 * * ?" timeZone="UTC" />
        </scheduling-strategy>
    </scheduler>

    <!-- Fetch records from source system -->
    <http:request config-ref="erp-api-config"
        method="GET"
        path="/api/export/pending-records" />

    <!-- Split into batches of 1000 (Certyo bulk limit) -->
    <batch:job jobName="certyo-bulk-batch">
        <batch:process-records>
            <batch:step name="transform-records">
                <ee:transform>
                    <ee:message>
                        <ee:set-payload><![CDATA[%dw 2.0
output application/json
---
{
    tenantId: p('certyo.tenant.id'),
    records: payload map (record) -> {
        database: record.sourceSystem default "erp",
        collection: record.entityType default "records",
        recordId: record.id as String,
        recordPayload: record.data,
        sourceTimestamp: record.updatedAt
    }
}]]></ee:set-payload>
                    </ee:message>
                </ee:transform>
            </batch:step>

            <batch:step name="send-to-certyo"
                acceptExpression="#[sizeOf(payload.records) > 0]">
                <http:request config-ref="certyo-api-config"
                    method="POST"
                    path="/api/v1/records/bulk" />
                <logger level="INFO"
                    message="Bulk batch sent: #[sizeOf(payload.records)] records" />
            </batch:step>
        </batch:process-records>

        <batch:on-complete>
            <logger level="INFO"
                message="Batch complete. Processed: #[payload.processedRecords], Failed: #[payload.failedRecords]" />
        </batch:on-complete>
    </batch:job>
</flow>

MuleSoft A2A Connector

MuleSoft's Agent-to-Agent (A2A) Connector supports the Google A2A protocol, enabling AI agents to discover and interact with MuleSoft APIs. As Certyo exposes its own A2A Agent Card, future versions of this integration will support autonomous AI-driven record ingestion and verification through MuleSoft's agent orchestration layer.

AI Integration · v1.0.0

AI Integration Skill

Download a skill file that enables AI agents to generate working MuleSoft + Certyo integration code for any language or framework.

v1.0.0

What is this?

A markdown file containing MuleSoft-specific field mappings, authentication setup, code examples, and integration patterns for Certyo. Drop it into your AI agent's context and ask it to generate integration code.

What's inside

Authentication — Anypoint Connected App and Mule Secure Properties for API key
Architecture — API-led connectivity — System, Process, and Experience layers
Field mapping — Generic ERP payload to Certyo record format via DataWeave
Code examples — DataWeave 2.0 transforms, Mule 4 flow XML, Batch Job for bulk
Error handling — Retry policies, exponential backoff, and dead letter queue patterns
Connector guide — Custom MuleSoft connector development for Exchange marketplace

How to use

Claude Code

Place the file in your project's .claude/commands/ directory, then use it as a slash command:

# Download the skill file
mkdir -p .claude/commands
curl -o .claude/commands/certyo-mulesoft.md \
  https://www.certyos.com/developers/skills/certyo-mulesoft-skill.md

# Use it in Claude Code
/certyo-mulesoft "Generate a Mule 4 flow with DataWeave that routes ERP events to Certyo"

Cursor / Copilot / Any AI Agent

Add the file to your project root or attach it to a conversation. The AI agent will use the MuleSoft-specific patterns, field mappings, and code examples to generate correct integration code.

# Add to your project
curl -o CERTYO_MULESOFT.md \
  https://www.certyos.com/developers/skills/certyo-mulesoft-skill.md

# Then in your AI agent:
"Using the Certyo MuleSoft spec in CERTYO_MULESOFT.md,
 generate a mule 4 flow with dataweave that routes erp events to certyo"

CLAUDE.md Context File

Append the skill file to your project's CLAUDE.md so every Claude conversation has MuleSoft + Certyo context automatically.

# Append to your project's CLAUDE.md
echo "" >> CLAUDE.md
echo "## Certyo MuleSoft Integration" >> CLAUDE.md
cat CERTYO_MULESOFT.md >> CLAUDE.md