parse-integration skill

Parse: document data extraction

Parse turns documents into structured JSON. You describe the fields you want, send a PDF, scan, or photo, and get those fields back with their values. It handles the layout variation that breaks template-based parsers: the same schema works across suppliers, form revisions, and scan quality.

Two ways to use it, and this skill covers both.

• Run an extraction right now with the parse_* MCP tools bundled in this plugin.
• Build an integration in the user's own codebase against the HTTP API.

With a shell, the ctio CLI installed and a Parse key in CT_PARSE_TOKEN, ctio parse <file> --schema fields.json runs an extraction in one command and waits for the result (see the ctio section of the convert skill).

The one rule that decides whether an integration works

POST /v1/extract is asynchronous by default. It answers 202 with an extraction id and status: "processing" immediately, and the extracted data is not in that response. You then poll GET /v1/extractions/{id} until status is completed or failed.

Branch on status, never on timing. Every response carries status, so one code path handles all of them. Integrations that assume "if the call took a while, the data must be in there" break the first time a document is slow or fast in the wrong direction, and they break silently.

There are two shortcuts, and both still report status, so the same branch handles them:

• wait=N holds the request open for up to N seconds (max 120) and returns the result inline if it finishes in time. It only applies when the document is uploaded in the same call. If the window expires you get the usual 202 and id, and you poll.
• A cache hit returns status: "completed" with cached: true instantly. The same document with the same field definitions, already extracted on that account, is served from storage and costs no pages.

---

Using Parse from this agent (MCP tools)

Tool	What it does
parse_extract	Submit a document. Returns an extraction id and status, not the data.
parse_extraction_status	Poll one extraction. Returns the data once status is completed.
parse_list_schemas	List saved field definitions on the account.
parse_create_schema	Create a reusable field definition.
parse_export	Turn a completed extraction into CSV or XLSX.
parse_usage	Pages used, page limit, remaining, reset date.

These tools are in the server's parse toolset, which this plugin turns on. On claude.ai or in Cowork, connect the plugin's connector once first: Customize > Plugins > Conversion Tools > Connectors > Connect. Added the server by hand and the tools are missing? Re-add it as https://mcp.conversiontools.io/mcp?toolsets=all (or ?toolsets=convert,parse).

The loop is always: parse_extract, then parse_extraction_status every 2 to 3 seconds until terminal, then optionally parse_export. Check parse_list_schemas before inventing a new field list - reusing a schema keeps results consistent between runs and lets identical documents hit the cache.

---

Building an integration

Base URL: https://api-parse.conversiontools.io Auth: an Authorization: Bearer <your API key> header on every request.

Step 1: get an API key

Create an account at parse.conversiontools.io, then go to Dashboard, API Keys (direct link) and create a key.

The key belongs to the user's application: it loads the key from its own secret manager at startup and passes it to the client (configure(api_key) in the reference implementations; <your API key> in the curl examples). Never commit it, never log it, never put it in client-side code - the key carries the account's full page allowance. Never ask the user to paste the key into the conversation.

Step 2: define a schema

A schema is a named list of fields. Save it once and reference it by schema_id, or pass an inline fields array for a genuine one-off.

Field rules, enforced server-side:

• name must be snake_case: lowercase letters, digits, underscores, starting with a letter.
• type is one of string, number, date, boolean, array, object.
• An array needs items (string, number, date, boolean, or object). An object, and an array of object, needs a nested fields list.
• Optional per field: description, required, examples, aliases, and validation (regex or enum for strings, min or max for numbers).
• Limits: 50 fields per level, 200 fields total, 4 levels of nesting.

description is the single biggest accuracy lever. "Invoice total including tax, bottom right, may be labelled Amount Due" beats total on its own. aliases covers the other labels a document might use for the same thing.

curl -X POST https://api-parse.conversiontools.io/v1/schemas \
  -H "Authorization: Bearer <your API key>" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Supplier invoice",
    "description": "Fields we post into the ledger",
    "fields": [
      { "name": "invoice_number", "type": "string", "required": true,
        "description": "Supplier invoice reference, usually top right",
        "aliases": ["Invoice No", "Rechnungsnummer"] },
      { "name": "invoice_date", "type": "date", "required": true },
      { "name": "currency", "type": "string",
        "validation": { "enum": ["EUR", "USD", "GBP"] } },
      { "name": "total_amount", "type": "number",
        "description": "Total including tax", "validation": { "min": 0 } },
      { "name": "line_items", "type": "array", "items": "object",
        "fields": [
          { "name": "description", "type": "string" },
          { "name": "quantity", "type": "number" },
          { "name": "unit_price", "type": "number" },
          { "name": "amount", "type": "number" }
        ] }
    ]
  }'

A bad field definition comes back as 400 with a details array naming the exact path, for example fields[0].name. Fix and resend. Nothing is billed for a rejected request.

Other schema endpoints: GET /v1/schemas, GET /v1/schemas/{id}, PUT /v1/schemas/{id} (full replace), DELETE /v1/schemas/{id}.

Step 3: submit a document

One call, multipart/form-data:

curl -X POST https://api-parse.conversiontools.io/v1/extract \
  -H "Authorization: Bearer <your API key>" \
  -F "file=@invoice.pdf" \
  -F "schema_id=SCHEMA_ID"

{ "success": true, "id": "65aa28ae...", "status": "processing" }

Form fields on this call: file (required), schema_id or fields (a JSON string), wait, webhook_url, no_cache, output_format.

Or split it into upload and extract, which is the better shape when you extract the same file more than once, or want the upload and the extraction on different code paths:

# 1. upload -> file_id
curl -X POST https://api-parse.conversiontools.io/v1/upload \
  -H "Authorization: Bearer <your API key>" \
  -F "file=@invoice.pdf"

# 2. extract by file_id (JSON body)
curl -X POST https://api-parse.conversiontools.io/v1/extract \
  -H "Authorization: Bearer <your API key>" \
  -H "Content-Type: application/json" \
  -d '{ "file_id": "FILE_ID", "schema_id": "SCHEMA_ID" }'

JSON body fields: file_id (required), schema_id or fields (an array), webhook_url, no_cache, output_format. Note that wait is not available on this path - submit, then poll.

Send no_cache only when you actually want to bypass the cache. Any value present is read as "bypass", so a literal false still disables caching. Leave the field out instead.

Step 4: poll until terminal, with bounded retries

curl https://api-parse.conversiontools.io/v1/extractions/EXTRACTION_ID \
  -H "Authorization: Bearer <your API key>"

{
  "success": true,
  "id": "65aa28ae...",
  "filename": "invoice.pdf",
  "status": "completed",
  "data": { "invoice_number": "INV-2026-0042", "total_amount": 1284.5, "line_items": [ ... ] },
  "pages_used": 2,
  "validation": { "status": "passed", "warnings": [], "coercions": [] },
  "error": null,
  "created_at": "2026-07-26T10:00:00.000Z",
  "completed_at": "2026-07-26T10:00:37.000Z"
}

Poll every 2 to 3 seconds with a small backoff, and stop on a wall-clock deadline rather than a retry count - a long document legitimately takes a minute or more, and a fixed count silently means different things at different intervals. Treat hitting the deadline as its own outcome, not as a failure of the extraction: the extraction keeps running server-side and the id stays valid, so you can requeue the poll later.

validation reports how well the result matched the requested fields: passed, partial (usable, but read warnings), failed (required fields missing or wrong-typed, treat the data as unreliable), or unvalidated (no field definitions were supplied). Surface partial and failed to a human instead of writing them straight into a ledger.

Step 5: handle failed

{ "status": "failed", "error": "...", "pages_used": 1 }

A failed extraction is not billed - the reserved page is refunded. Do not blind-retry the same bytes with the same schema: it will fail the same way. Log the id and the error, and route it to a human or to a fallback. Retrying is only sensible after the input or the field definitions change.

Re-sending the same document while an extraction is still running returns the same id instead of starting a duplicate, so a retry after a network blip is safe and costs nothing extra.

Step 6: webhooks instead of polling

Pass webhook_url at submit time and Parse posts to it when the extraction reaches a terminal state:

{
  "event": "extraction.completed",
  "id": "65aa28ae...",
  "status": "completed",
  "pages_used": 2,
  "error": null,
  "created_at": "...",
  "completed_at": "..."
}

The payload deliberately carries the id and status only, never the extracted data. Your handler treats it as a trigger and fetches the result over the authenticated API. That way a misconfigured or hijacked endpoint cannot leak anyone's documents.

Requirements and behaviour:

• The URL must be publicly reachable over http or https. Private and internal addresses are rejected at submit time with 400.
• Delivery is fire and forget with a short timeout and a few bounded retries. A webhook can be missed, so keep a slow reconciliation poll for anything you have not heard about.
• There is no signature on the payload. Use an unguessable path, treat the URL itself as a secret, and re-fetch through the API before acting on anything. Never trust the body as the source of truth.

Step 7: export to a spreadsheet

Exports run the extracted JSON through a normal conversion. They are free, repeatable, and never re-run the extraction.

# start it (idempotent per format - a running export is reused)
curl -X POST https://api-parse.conversiontools.io/v1/extractions/EXTRACTION_ID/export \
  -H "Authorization: Bearer <your API key>" \
  -H "Content-Type: application/json" \
  -d '{ "format": "xlsx" }'

Then poll GET /v1/extractions/EXTRACTION_ID/export?format=xlsx with the same Authorization header and save the body once it is the file; the Python and Node clients below do exactly that. The GET speaks two languages: application/json with {"status": "processing", "progress": N} while it converts, and the file's own bytes once it is ready. Check the response Content-Type, do not assume you got a file. Nested arrays are flattened into rows with the parent values repeated, which is the layout spreadsheet users expect.

You can also pass output_format: "csv" | "xlsx" at submit time to start the export automatically as soon as the extraction completes.

Other endpoints

• GET /v1/usage - plan, pages_used, pages_limit, pages_remaining, reset_date. One page is billed per document page. Failed extractions are refunded, cache hits cost nothing.
• DELETE /v1/extractions/{id} - deletes the stored result and its exports. Results are kept until you delete them, so wire this into the user's data-retention policy.

---

Errors worth handling explicitly

Status	Meaning	What to do
400	Bad request. details names the exact field path.	Fix the schema or parameters and resend. Nothing was billed.
401	Missing, malformed, or revoked key.	Check the Authorization header and the key. Do not retry.
404	No such extraction or schema on this account. Ids are per-account.	Check the id. Do not retry.
409	Export requested before the extraction completed.	Poll until completed, then export.
410	The stored result is gone.	Re-submit the document with no_cache: true.
429	Monthly page allowance used up.	Stop. Retrying fails the same way. Check GET /v1/usage, raise the plan at Dashboard, Billing.
5xx	Server-side.	Retry a couple of times with backoff, then give up and report.

Rule of thumb: retry 5xx and network errors, never 4xx.

---

Reference implementation: Python

Python 3.9+ with the requests package.

"""Minimal, production-shaped Parse client: submit, poll until terminal, export."""

import json
import time
from pathlib import Path
from typing import Any, Optional

import requests

BASE_URL = "https://api-parse.conversiontools.io/v1"
HEADERS: dict[str, str] = {}


def configure(api_key: str) -> None:
    """Call once at startup. The key comes from your app's secret manager, never from this file."""
    HEADERS["Authorization"] = f"Bearer {api_key}"


class ParseError(RuntimeError):
    """An API call was rejected."""

    def __init__(self, status: int, message: str, details: Any = None):
        super().__init__(f"HTTP {status}: {message}")
        self.status = status
        self.details = details


class ExtractionFailed(RuntimeError):
    """The document was processed but produced no usable result."""


class ExtractionPending(RuntimeError):
    """Still processing when our deadline ran out. The id stays valid."""


def _unwrap(response: requests.Response) -> dict:
    if response.ok:
        return response.json()
    try:
        body = response.json()
    except ValueError:
        body = {}
    raise ParseError(
        response.status_code,
        body.get("message") or body.get("error") or response.reason,
        body.get("details"),
    )


def create_schema(name: str, fields: list[dict], description: str = "") -> str:
    payload = {"name": name, "fields": fields, "description": description}
    body = _unwrap(requests.post(f"{BASE_URL}/schemas", headers=HEADERS, json=payload, timeout=30))
    return body["schema"]["id"]


def submit(
    path: str | Path,
    *,
    schema_id: Optional[str] = None,
    fields: Optional[list[dict]] = None,
    webhook_url: Optional[str] = None,
    no_cache: bool = False,
) -> dict:
    """Submit a document. Returns the envelope with `id` and `status`.

    This does NOT return the extracted data - poll wait_for_result() next.
    """
    form: dict[str, str] = {}
    if schema_id:
        form["schema_id"] = schema_id
    elif fields:
        form["fields"] = json.dumps(fields)
    if webhook_url:
        form["webhook_url"] = webhook_url
    if no_cache:
        # Send the field only when bypassing the cache: any value present counts.
        form["no_cache"] = "true"

    path = Path(path)
    with path.open("rb") as handle:
        response = requests.post(
            f"{BASE_URL}/extract",
            headers=HEADERS,
            files={"file": (path.name, handle)},
            data=form,
            timeout=120,
        )
    return _unwrap(response)


def get_extraction(extraction_id: str) -> dict:
    return _unwrap(requests.get(f"{BASE_URL}/extractions/{extraction_id}", headers=HEADERS, timeout=30))


def wait_for_result(extraction_id: str, *, timeout_s: float = 600.0) -> dict:
    """Poll until terminal. Branches on `status`, never on elapsed time."""
    deadline = time.monotonic() + timeout_s
    interval = 2.0

    while time.monotonic() < deadline:
        body = get_extraction(extraction_id)
        status = body.get("status")

        if status == "completed":
            return body
        if status == "failed":
            raise ExtractionFailed(body.get("error") or "extraction failed")
        # "pending" / "processing" -> keep waiting

        time.sleep(interval)
        interval = min(interval * 1.5, 10.0)   # gentle backoff, capped

    raise ExtractionPending(
        f"{extraction_id} still processing after {timeout_s:.0f}s. "
        "It is still running server-side; poll this id again later."
    )


def export(extraction_id: str, fmt: str, out_path: str | Path, *, timeout_s: float = 300.0) -> Path:
    """Export a COMPLETED extraction to csv or xlsx and save it."""
    _unwrap(requests.post(
        f"{BASE_URL}/extractions/{extraction_id}/export",
        headers=HEADERS, json={"format": fmt}, timeout=30,
    ))

    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        response = requests.get(
            f"{BASE_URL}/extractions/{extraction_id}/export",
            headers=HEADERS, params={"format": fmt}, timeout=60,
        )
        if not response.ok:
            _unwrap(response)

        # JSON means "still converting" or "failed"; anything else is the file.
        if response.headers.get("content-type", "").startswith("application/json"):
            state = response.json()
            if state.get("status") == "failed":
                raise ExtractionFailed(state.get("error") or "export failed")
            time.sleep(2.0)
            continue

        out_path = Path(out_path)
        out_path.write_bytes(response.content)
        return out_path

    raise ExtractionPending(f"export of {extraction_id} still converting after {timeout_s:.0f}s")


def main(api_key: str) -> None:
    configure(api_key)
    submitted = submit("invoice.pdf", schema_id="<your schema id>")
    print("submitted:", submitted["id"], submitted["status"])

    try:
        # A cache hit already comes back completed - the same branch handles it.
        result = submitted if submitted["status"] == "completed" else wait_for_result(submitted["id"])
    except ExtractionFailed as exc:
        raise SystemExit(f"extraction failed, not retrying: {exc}")

    print("pages billed:", result.get("pages_used"), "cached:", result.get("cached", False))
    print(json.dumps(result["data"], indent=2, ensure_ascii=False))

    if (result.get("validation") or {}).get("status") in {"partial", "failed"}:
        print("needs review:", result["validation"].get("warnings"))

    export(result["id"], "xlsx", "invoice.xlsx")


if __name__ == "__main__":
    from getpass import getpass

    main(getpass("Parse API key: "))   # for a one-off run; a service injects it from its secret manager