> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.emtelligent.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.emtelligent.com/_mcp/server.

# Quickstart Code Example

The Extract API runs a named pipeline over a set of documents listed in a
**manifest** in your own S3 bucket. Like the Prep API it reads and writes your
storage directly, so a job needs storage credentials as well as your API key.

```bash
export EXTRACT_API_KEY='...'
export AWS_ACCESS_KEY_ID='...'
export AWS_SECRET_ACCESS_KEY='...'
```

## Create a session

Authentication is an API key in the `X-API-Key` header. Create one with the
`extract:process` scope on the [emtelligent platform](/platform/api-keys).

```python
import os
import requests

BASE = "https://extraction-api.emtelligent.com"

session = requests.Session()
session.headers["X-API-Key"] = os.environ["EXTRACT_API_KEY"]
```

## Find a pipeline

Pipelines are named, and which ones exist depends on your deployment — so ask
rather than hard-coding a name.

```python
pipelines = session.get(f"{BASE}/api/v1/pipelines").json()
print(pipelines)
```

Each pipeline takes its own parameters, which you can inspect before submitting:

```python
params = session.get(
    f"{BASE}/api/v1/pipelines/core_clinical/pipeline_params"
).json()
```

## Create a job

`manifest_key` is the object key of the manifest **within your input bucket**, and
`aws_region` is the region that bucket lives in.

No bucket name appears in the request. Your input and output buckets are fixed
during [onboarding](/extract/workflow/setup) and resolved from your API key, so a
job names only the key inside them — the credentials you send are what lets the
service read and write those buckets on your behalf.

```python
response = session.post(
    f"{BASE}/api/v1/jobs",
    json={
        "manifest_key": "manifests/batch-001.json",
        "pipeline_name": "core_clinical",
        "aws_region": "us-east-1",
        "user_credentials": {
            "aws_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
            "aws_secret_access_key": os.environ["AWS_SECRET_ACCESS_KEY"],
        },
        "job_name": "batch-001",
    },
)
response.raise_for_status()
job_id = response.json()["job_id"]
```

> **Warning**
>
> These credentials are used by the service to read your documents and write your
> results. Scope them to the buckets this job needs.

## Wait for it to finish

A job moves through `queued`, `validating`, `ingesting` and `running` before it
reaches `completed`, `failed` or `cancelled`; it can also sit at `paused`.
Collecting adds three more — see below. Once the retention window has passed the
job reports `output_expired`: it succeeded, but its output can no longer be
collected, so treat that as terminal too or a polling loop will never exit.

```python
import time

TERMINAL = {"completed", "failed", "cancelled", "output_expired"}

while True:
    status = session.get(f"{BASE}/api/v1/jobs/{job_id}/status").json()
    if status["job_status"] in TERMINAL:
        break
    time.sleep(10)

if status["job_status"] != "completed":
    raise RuntimeError(f"job {job_id} ended as {status['job_status']}")
```

## Collect the output

Collecting is a transfer, not a download: the service copies the results into
**your** bucket under `output_prefix`, with the job id appended, so it needs the
credentials and region again.

```python
response = session.post(
    f"{BASE}/api/v1/jobs/{job_id}/collect",
    json={
        "output_prefix": "extract-output",
        "aws_region": "us-east-1",
        "user_credentials": {
            "aws_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
            "aws_secret_access_key": os.environ["AWS_SECRET_ACCESS_KEY"],
        },
    },
)
response.raise_for_status()
```

The job then reports `transferring`, and finishes at `transfer_completed` or
`transfer_failed`. Your results land in your **output** bucket — again the one from
onboarding, not named here — under `extract-output/<job_id>/`.

## The whole thing

**`extract_quickstart.py`**

```python title="extract_quickstart.py"
import os
import time

import requests

BASE = "https://extraction-api.emtelligent.com"
REGION = "us-east-1"

CREDENTIALS = {
    "aws_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
    "aws_secret_access_key": os.environ["AWS_SECRET_ACCESS_KEY"],
}

session = requests.Session()
session.headers["X-API-Key"] = os.environ["EXTRACT_API_KEY"]


def wait_for(job_id, done, every=10):
    """Poll a job until its status is in `done`, then return that status."""
    while True:
        status = session.get(f"{BASE}/api/v1/jobs/{job_id}/status").json()
        if status["job_status"] in done:
            return status
        time.sleep(every)


def extract(manifest_key, pipeline_name, output_prefix, job_name=None):
    """Run a pipeline over a manifest and copy the output to your bucket."""
    response = session.post(
        f"{BASE}/api/v1/jobs",
        json={
            "manifest_key": manifest_key,
            "pipeline_name": pipeline_name,
            "aws_region": REGION,
            "user_credentials": CREDENTIALS,
            "job_name": job_name,
        },
    )
    response.raise_for_status()
    job_id = response.json()["job_id"]

    # `output_expired` is terminal as well: the job succeeded long enough ago that
    # its output is gone, and waiting for `completed` would never return.
    status = wait_for(job_id, {"completed", "failed", "cancelled", "output_expired"})
    if status["job_status"] != "completed":
        raise RuntimeError(f"job {job_id} ended as {status['job_status']}")

    response = session.post(
        f"{BASE}/api/v1/jobs/{job_id}/collect",
        json={
            "output_prefix": output_prefix,
            "aws_region": REGION,
            "user_credentials": CREDENTIALS,
        },
    )
    response.raise_for_status()

    status = wait_for(job_id, {"transfer_completed", "transfer_failed"})
    if status["job_status"] != "transfer_completed":
        raise RuntimeError(f"collecting job {job_id} ended as {status['job_status']}")

    return f"{output_prefix}/{job_id}/"


if __name__ == "__main__":
    print(extract("manifests/batch-001.json", "core_clinical",
                  "extract-output", job_name="batch-001"))
```

## Next

#### [API Reference](/extract/api)

Every operation, including cancel and job listing.

#### [Output reference](/extract/reference/output)

What the extracted data looks like, field by field.