ShippedFlow Partner API

Authorization: Bearer <your_api_key>

{
  "correlationId": "550e8400-e29b-41d4-a716-446655440000",
  "customerRef": "your-ref",
  "status": "error",
  "jobId": "sf-job-550e8400",
  "completedAt": "2026-06-14T10:23:00.000Z",
  "error": {
    "code": "submission_too_large",
    "message": "Combined document size exceeds processing limit. Please split the submission into smaller batches."
  }
}

import hmac
import hashlib
from flask import request, abort

SHIPPEDFLOW_HMAC_SECRET = "your-shared-secret-here"

def verify_shippedflow_signature():
    received = request.headers.get("X-Shippedflow-Signature", "")
    if not received.startswith("sha256="):
        abort(401)
    expected = "sha256=" + hmac.new(
        SHIPPEDFLOW_HMAC_SECRET.encode(),
        request.get_data(),   # raw body, NOT request.get_json()
        hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(received, expected):
        abort(401)

const crypto = require("crypto");

const SHIPPEDFLOW_HMAC_SECRET = "your-shared-secret-here";

function verifyShippedflowSignature(req) {
  const received = req.header("X-Shippedflow-Signature") || "";
  if (!received.startsWith("sha256=")) return false;
  const expected = "sha256=" + crypto
    .createHmac("sha256", SHIPPEDFLOW_HMAC_SECRET)
    .update(req.rawBody)       // raw body buffer, NOT req.body parsed
    .digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(received),
    Buffer.from(expected)
  );
}

{
  "error": "validation_failed",
  "message": "documents[2].filedata is not valid base64",
  "correlationId": "550e8400-e29b-41d4-a716-446655440000"
}

curl -X POST https://api.shippedflow.info/api/v1/partner/submit \
  -H "Authorization: Bearer $SHIPPEDFLOW_API_KEY" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d @- <<EOF
{
  "correlationId": "$(uuidgen)",
  "customerRef": "futurbano-sl",
  "webhookUrl": "https://your-system.example.com/webhooks/shippedflow",
  "operationType": "IMPORT",
  "documents": [
    {
      "name": "invoice.pdf",
      "type": "INVOICE",
      "filedata": "$(base64 -i invoice.pdf)"
    }
  ]
}
EOF

import base64
import os
import uuid
import requests

API_KEY = os.environ["SHIPPEDFLOW_API_KEY"]
API_BASE = "https://api.shippedflow.info"

def submit(pdf_paths, customer_ref, webhook_url):
    documents = []
    for path in pdf_paths:
        with open(path, "rb") as f:
            documents.append({
                "name": os.path.basename(path),
                "type": "INVOICE",        # or PACKING_LIST / BILL_OF_LADING
                "filedata": base64.b64encode(f.read()).decode("ascii"),
            })

    correlation_id = str(uuid.uuid4())
    response = requests.post(
        f"{API_BASE}/api/v1/partner/submit",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
            "Idempotency-Key": str(uuid.uuid4()),
        },
        json={
            "correlationId": correlation_id,
            "customerRef": customer_ref,
            "webhookUrl": webhook_url,
            "operationType": "IMPORT",
            "documents": documents,
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

if __name__ == "__main__":
    result = submit(
        ["invoice.pdf", "packing-list.pdf"],
        customer_ref="futurbano-sl",
        webhook_url="https://your-system.example.com/webhooks/shippedflow",
    )
    print(f"Accepted: jobId={result['jobId']} correlationId={result['correlationId']}")

import hmac
import hashlib
import os
from flask import Flask, request, jsonify, abort

app = Flask(__name__)
SECRET = os.environ["SHIPPEDFLOW_HMAC_SECRET"].encode()

@app.route("/webhooks/shippedflow", methods=["POST"])
def shippedflow_webhook():
    # 1. Verify signature against raw body
    received = request.headers.get("X-Shippedflow-Signature", "")
    expected = "sha256=" + hmac.new(SECRET, request.get_data(), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(received, expected):
        abort(401)

    # 2. Deduplicate by delivery id (in case of retries)
    delivery_id = request.headers.get("X-Shippedflow-Delivery-Id")
    if already_processed(delivery_id):
        return jsonify({"status": "duplicate_ignored"}), 200

    # 3. Process payload
    payload = request.get_json()
    save_result(payload["correlationId"], payload)

    return jsonify({"status": "ok"}), 200