flowflex 0.1.0

Official FlowFlex SDK — fire custom-integration events and attach private files without handling presigned URLs yourself.

flowflex

Official Python SDK for firing FlowFlex custom-integration events and attaching private files to them — without ever dealing with presigned URLs, Basic-auth headers, or the multi-step upload dance yourself.

pip install flowflex

Requires Python 3.9+.

---

Quick start

from flowflex import FlowFlex

ff = FlowFlex(
    api_key="cik_xxx",            # from your custom integration
    api_secret="yyy",
    integration_code="ic_abc123", # the code in your event URL: /v1/events/<code>
    base_url="https://api.flowflex.ai",
)

# Plain event, no file
ff.send_event("order.placed", payload={"name": "Ada", "order_id": "ord_42"})

In your flow builder, the values land under trigger — e.g. {{trigger.name}}, {{trigger.order_id}}.

---

Attaching a private file

Wrap any file in ff.file(...) and drop it into the payload. The SDK uploads it through a presigned URL and replaces it with its opaque assetId before the event is sent. The file bytes go straight to storage — they never pass through the FlowFlex app server.

ff.send_event("invoice.created", payload={
    "assetId": ff.file("./invoice.pdf"),  # ← becomes "asset_..." on the wire
    "customer_id": "cust_123",
})

Then in your flow's Media Message node:

1. Set File source → Private file (assetId)
2. Set the Asset ID field → {{trigger.assetId}}

The key you use in the payload is the key you reference in the flow. If you send {"invoice": ff.file(...)}, reference it as {{trigger.invoice}}.

Multiple files

A flow can have as many media nodes as you like — put a file() anywhere in the payload (top-level, nested, or in lists) and the SDK uploads them all in parallel and swaps each for its assetId. The shape is entirely up to you; reference each one by its path in the flow builder.

ff.send_event("order.shipped", payload={
    "invoice": ff.file("./invoice.pdf"),                    # {{trigger.invoice}}
    "label":   ff.file("./label.png"),                      # {{trigger.label}}
    "gallery": [ff.file("./a.jpg"), ff.file("./b.jpg")],    # {{trigger.gallery[0]}}, {{trigger.gallery[1]}}
    "order":   {"receipt": ff.file("./receipt.pdf")},       # {{trigger.order.receipt}}
    "note":    "non-file values pass through untouched",
})
# → {"invoice": "asset_a", "label": "asset_b",
#    "gallery": ["asset_c", "asset_d"],
#    "order": {"receipt": "asset_e"}, "note": "..."}

result.uploaded_assets maps each payload path to its assetId, e.g. {"invoice": "asset_a", "gallery[0]": "asset_c", "order.receipt": "asset_e"}.

Reusing one file across nodes: if you pass the same file() instance in multiple places, it's uploaded only once and the same assetId is used everywhere:

banner = ff.file("./banner.png")
ff.send_event("promo.sent", payload={
    "header": banner, "footer": banner,  # one upload, same assetId in both
})

---

Supplying file bytes other ways

file() accepts a path (str or pathlib.Path), raw bytes, or any binary file-like object. When the type can't be inferred, pass filename and mime:

# Raw bytes
ff.file(pdf_bytes, filename="invoice.pdf", mime="application/pdf")

# An open file object
with open("invoice.pdf", "rb") as fh:
    ff.send_event("invoice.created", payload={"assetId": ff.file(fh)})

# Override the stored filename
ff.file("./tmp-7f3a.pdf", filename="Invoice-2026.pdf")

Allowed types: PDF, DOC(X), XLS(X), PPT(X), TXT, JPEG, PNG, MP4, 3GPP, AAC, AMR, MP3, OGG. Max size: 25 MB.

---

API

FlowFlex(...)

Option	Type	Required	Notes
api_key	str	yes	Custom-integration key (cik_…).
api_secret	str	yes	Custom-integration secret.
integration_code	str	yes	The <code> in /v1/events/<code>.
base_url	str	yes	FlowFlex host. Must be https (except localhost). Trailing /api stripped.
timeout_seconds	float	no	Per-request timeout. Default 30.0.
max_file_bytes	int	no	Client-side size cap. Default 26214400 (25 MB).
session	requests.Session	no	Custom session for connection pooling / proxies.

ff.send_event(event, payload=None, idempotency_key=None)

Uploads any file() in payload, then POSTs the event. Returns a SendEventResult:

result.response         # raw body from the events endpoint
result.uploaded_assets  # dict: payload path → assetId

An idempotency_key is auto-generated (UUID) if you don't pass one — safe to retry the same call.

ff.file(source, filename=None, mime=None, size=None)

Returns a lazy FileRef. Bytes are read only when the event is sent.

Lower-level helpers

info = ff.create_upload_url(filename="x.pdf", mime="application/pdf")
# → {"assetId": ..., "uploadUrl": ..., "token": ...}

asset_id = ff.upload_file(ff.file("./x.pdf"))  # upload, get assetId

---

Errors

All errors extend FlowFlexError (.message, .status, .code, .details):

• FlowFlexConfigError — bad/missing constructor options.
• FlowFlexUploadError — a file couldn't be read or storage rejected it.
• FlowFlexError — API or network failure (.code is the backend error code, e.g. MIME_NOT_ALLOWED, ASSET_FILE_MISSING).

from flowflex import FlowFlex, FlowFlexError

try:
    ff.send_event("invoice.created", payload={"assetId": ff.file("./big.pdf")})
except FlowFlexError as err:
    print(err.code, err.status, err.message)

---

File lifetime & storage cleanup

Important — read before using file attachments in production.

When you call ff.file(...), the file is uploaded to private storage that only the FlowFlex backend can read. It is not a public URL and the caller cannot access it after upload.

How long does the file stay?

Phase	Duration
Presigned upload URL valid	2 hours from create_upload_url
File kept in storage	48 hours from upload
After 48 hours	File deleted from storage + record removed

What this means for you

• Do not store assetId long-term expecting to reuse it. It expires in 48h.
• Each event send should get a fresh assetId by calling send_event with a new ff.file(...). The SDK handles the upload automatically.
• If you fire the event more than 48h after uploading, the file will be gone and the flow will fail with ASSET_FILE_MISSING. Keep your event send close to the upload.
• Re-sending the same message to a different recipient after 48h requires a fresh upload — call send_event again with the file, don't reuse the old assetId.

Typical correct pattern

# ✅ Upload + fire in the same operation — always fresh
ff.send_event("invoice.created", payload={"assetId": ff.file("./invoice.pdf")})

# ❌ Don't store assetId and reuse it hours later
result = ff.send_event(...)
# ... 50 hours later ...
# result.uploaded_assets["assetId"] is now expired and deleted

---

Security

Your api_key/api_secret are integration-wide credentials — anyone who obtains them can fire events and upload files as you. Keep them in environment variables or a secrets manager, never in source control.

Protections built in:

• HTTPS enforced. base_url must be https:// (only localhost may use http), so Basic-auth credentials are never sent in cleartext.
• No credential leakage. The Authorization header is never included in error messages or FlowFlexError.details.
• Header-injection safe. The event name and idempotency_key are rejected if they contain control characters (CRLF).
• Path-injection safe. integration_code is URL-encoded into the request path.
• Per-request timeouts (default 30 s) on every network call.
• Client-side size cap (default 25 MB) so oversized files fail before upload.