Example integration

A typical integration: your application already holds a taxpayer's documents and wants the finished return back automatically — no manual uploading, no copy‑pasting status. The whole exchange is a handful of API calls.

Reference implementation

A complete, runnable version of this integration lives on GitHub — ushabti-org/integrations-example. It wires the flow below into a small existing app without changing that app's data model: a per‑organization encrypted credential store, the presigned upload handshake, and a signed‑webhook receiver, all behind a single sendToMagnetic() seam you can lift into your own codebase.

To see exactly what an integration adds to an existing product — the diff a partner engineer would open as a pull request — read integrations-example #1: the base app stays untouched and everything Magnetic‑specific is isolated to the integration layer.

Flow

1. Create a client for the taxpayer.
2. Upload their documents (presigned, so bytes go straight to storage).
3. Submit for prep and store the returned expected_completion_at to show your user.
4. Get notified when it's completed — via a webhook (recommended) or by polling as a fallback.
5. Download the output documents and surface them back in your app.

End-to-end (pseudocode)

import requests

# `MAGNETIC_API_KEY`, `documents`, and the `save_expected_date(...)` / `store_in_your_app(...)`
# helpers are YOUR application's — they're not part of the API or an SDK. Swap in your own.
BASE = "https://api.magnetictax.com"
H = {"Authorization": f"Bearer {MAGNETIC_API_KEY}"}

# 1. Create the client
client = requests.post(f"{BASE}/v1/clients",
                       headers=H, json={"friendly_name": taxpayer_name}).json()
cid = client["id"]

# 2. Presigned upload of each document
manifest = [{"filename": d.name, "content_type": d.mime} for d in documents]
prep = requests.post(f"{BASE}/v1/clients/{cid}/documents/prepare",
                     headers=H, json={"documents": manifest}).json()
failed_ids = prep.get("failed_document_ids", [])
failed_indices = prep.get("failed_document_indices", [])       # deprecated; also covers id-less failures
if failed_ids or failed_indices:                              # some manifest entries were rejected
    raise RuntimeError(f"prepare failed: ids={failed_ids} indices={failed_indices}")
for target, doc in zip(prep["upload_targets"], documents):
    requests.post(target["upload_url"], data=target["fields"],
                  files={"file": (doc.name, doc.bytes)})              # → 204
requests.post(f"{BASE}/v1/clients/{cid}/documents/finalize",
              headers=H, json={"document_ids": [t["document_id"] for t in prep["upload_targets"]]})

# 3. Submit for prep
sub = requests.post(f"{BASE}/v1/clients/{cid}/submissions",
                    headers=H, json={"tax_software": "Drake", "tax_year": 2025}).json()
save_expected_date(sub["expected_completion_at"])     # show your user the ETA

# 4. Learn when it's completed, then ingest outputs.
#    RECOMMENDED: do this from your submission.updated webhook handler (see webhooks docs)
#    — no polling at all. Shown here as the polling fallback: processing takes ~3 business
#    days, so poll a few times a day (not in a tight loop) until it reaches a terminal state.
import time

while True:
    r = requests.get(f"{BASE}/v1/submissions/{sub['id']}", headers=H, timeout=30)
    if r.status_code == 429:                          # rate-limited — back off as told
        time.sleep(int(r.headers.get("Retry-After", "60")))
        continue
    r.raise_for_status()                              # surface 4xx/5xx, don't parse an error body
    s = r.json()
    if s["status"] in ("completed", "cancelled"):
        break
    time.sleep(6 * 60 * 60)  # ~4x/day

if s["status"] == "completed":
    for out in s["output_documents"]:
        dl = requests.get(out["download_url"], timeout=60)
        dl.raise_for_status()
        store_in_your_app(out["filename"], dl.content)

That's the entire integration — a few endpoints, no bespoke file plumbing on either side. For the recommended completion path, register a webhook; see the Quickstart for curl versions of each call.