> ## Documentation Index
> Fetch the complete documentation index at: https://docs.markpdf.tech/llms.txt
> Use this file to discover all available pages before exploring further.

# FastAPI

> Secure proxy from a API Python, with job polling, idiomatic HTTPException and production.

# FastAPI

FastAPI is already async by nature, so `httpx.AsyncClient` is the natural choice to talk to the API conversion without blocking the event loop.

## Basic route with `UploadFile`

```python theme={null}
import httpx
from fastapi import FastAPI, UploadFile
from fastapi.responses import Response

app = FastAPI()

FLASH_MD_URL = "https://api.markpdf.tech"
FLASH_MD_KEY = "YOUR_API_KEY"


@app.post("/ai/read-document")
async def read_document(file: UploadFile):
    content = await file.read()
    async with httpx.AsyncClient(timeout=300) as client:
        res = await client.post(
            f"{FLASH_MD_URL}/convert/raw",
            params={"filename": file.filename, "mode": "fast"},
            headers={
                "x-api-key": FLASH_MD_KEY,
                "content-type": file.content_type or "application/octet-stream",
            },
            content=content,
        )

    return Response(
        content=res.content,
        status_code=res.status_code,
        media_type=res.headers.get("content-type", "text/markdown"),
    )
```

<Note>
  For production, vavalidate the size of `UploadFile` before reading it in its entirety into memory (see Production section below).
</Note>

## Language exceptions with `HTTPException`

Instead of forwarding the raw body, map the API status code to a FastAPI `HTTPException`. This gives you error responses consistent with the rest of your API and integrates them with FastAPI/OpenAPI's automatic error handling.

```python theme={null}
import httpx
from fastapi import FastAPI, UploadFile, HTTPException

app = FastAPI()

FLASH_MD_URL = "https://api.markpdf.tech"
FLASH_MD_KEY = "YOUR_API_KEY"


def _detail_from_response(res: httpx.Response) -> str:
    try:
        return res.json().get("detail", res.text)
    except Exception:
        return res.text


@app.post("/ai/read-document")
async def read_document(file: UploadFile):
    content = await file.read()

    async with httpx.AsyncClient(timeout=300) as client:
        try:
            res = await client.post(
                f"{FLASH_MD_URL}/convert/raw",
                params={"filename": file.filename, "mode": "fast"},
                headers={
                    "x-api-key": FLASH_MD_KEY,
                    "content-type": file.content_type or "application/octet-stream",
                },
                content=content,
            )
        except httpx.TimeoutException:
            raise HTTPException(status_code=504, detail="Timeout calling the conversion API")
        except httpx.ConnectError:
            raise HTTPException(status_code=502, detail="Could not connect to the conversion API")

    if res.status_code == 200:
        return Response(content=res.content, media_type="text/markdown")

    if res.status_code == 202:
        job = res.json()
        markdown = await poll_job(job["job_id"])
        return Response(content=markdown.encode(), media_type="text/markdown")

    # 400, 401, 403, 413, 415, 422, 429, 500 -> HTTPException with the same status
    raise HTTPException(status_code=res.status_code, detail=_detail_from_response(res))
```

## Background polling for 202 responses

When all backends are saturated, API responds `202` with `job_id`. The following helper reuses the `httpx.AsyncClient` client to poll `GET /jobs/{job_id}` every \~5 seconds until `completed` or `failed`:

```python theme={null}
import asyncio
import httpx
from fastapi import HTTPException

FLASH_MD_URL = "https://api.markpdf.tech"
FLASH_MD_KEY = "YOUR_API_KEY"


async def poll_job(job_id: str, timeout_seconds: int = 300) -> str:
    deadline = asyncio.get_event_loop().time() + timeout_seconds

    async with httpx.AsyncClient(timeout=30) as client:
        while True:
            if asyncio.get_event_loop().time() > deadline:
                raise HTTPException(status_code=504, detail=f"Timeout esperando el job {job_id}")

            await asyncio.sleep(5)
            res = await client.get(
                f"{FLASH_MD_URL}/jobs/{job_id}",
                headers={"x-api-key": FLASH_MD_KEY},
            )
            if res.status_code != 200:
                raise HTTPException(status_code=res.status_code, detail=res.text)

            data = res.json()
            if data["status"] == "completed":
                return data["body"]
            if data["status"] == "failed":
                raise HTTPException(status_code=500, detail=data.get("error", "El job falló"))
            # "queued" o "processing": seguimos esperando
```

Ruta completa combinando ambos:

```python theme={null}
from fastapi import FastAPI, UploadFile, HTTPException
from fastapi.responses import PlainTextResponse

app = FastAPI()


@app.post("/ai/read-document")
async def read_document(file: UploadFile):
    content = await file.read()
    async with httpx.AsyncClient(timeout=300) as client:
        res = await client.post(
            f"{FLASH_MD_URL}/convert/raw",
            params={"filename": file.filename, "mode": "fast"},
            headers={
                "x-api-key": FLASH_MD_KEY,
                "content-type": file.content_type or "application/octet-stream",
            },
            content=content,
        )

        if res.status_code == 200:
            return PlainTextResponse(res.text, media_type="text/markdown")

        if res.status_code == 202:
            job = res.json()
            markdown = await poll_job(job["job_id"])
            return PlainTextResponse(markdown, media_type="text/markdown")

        raise HTTPException(status_code=res.status_code, detail=_detail_from_response(res))
```

## Conversion by URL (`/convert/from-url`)

```python theme={null}
from pydantic import BaseModel
from fastapi import FastAPI, HTTPException

app = FastAPI()


class ConvertFromUrlBody(BaseModel):
    url: str
    filename: str | None = None
    mode: str = "fast"


@app.post("/ai/read-from-url")
async def read_from_url(body: ConvertFromUrlBody):
    async with httpx.AsyncClient(timeout=300) as client:
        res = await client.post(
            f"{FLASH_MD_URL}/convert/from-url",
            headers={"x-api-key": FLASH_MD_KEY},
            json=body.model_dump(),
        )

    if res.status_code == 200:
        return PlainTextResponse(res.text, media_type="text/markdown")
    if res.status_code == 202:
        markdown = await poll_job(res.json()["job_id"])
        return PlainTextResponse(markdown, media_type="text/markdown")

    raise HTTPException(status_code=res.status_code, detail=_detail_from_response(res))
```

## Production

### Timeouts with `httpx.Timeout`

It will separate the connection, writing and reading timeout. Large documents in `quality`/`balanced` require more reading time than connection time:

```python theme={null}
timeout = httpx.Timeout(connect=10.0, write=30.0, read=300.0, pool=10.0)

async with httpx.AsyncClient(timeout=timeout) as client:
    ...
```

### Retries with exponential backoff and jitter

Retry only `429` and `5xx`; for the rest of the `4xx`, do not retry without correcting the request first:

```python theme={null}
import asyncio
import random
import httpx


async def post_with_retry(
    client: httpx.AsyncClient,
    url: str,
    max_retries: int = 4,
    **kwargs,
) -> httpx.Response:
    for attempt in range(max_retries + 1):
        try:
            res = await client.post(url, **kwargs)
        except (httpx.TimeoutException, httpx.ConnectError):
            if attempt == max_retries:
                raise
            await asyncio.sleep((2 ** attempt) + random.uniform(0, 1))
            continue

        if res.status_code == 429 or res.status_code >= 500:
            if attempt == max_retries:
                return res
            retry_after = res.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else (2 ** attempt) + random.uniform(0, 1)
            await asyncio.sleep(delay)
            continue

        return res

    return res
```

### Vavalidate size before reading from memory

`UploadFile` exposes a `SpooledTemporaryFile`; you can check `Content-Length` before reading the whole body:

```python theme={null}
from fastapi import Request, HTTPException

MAX_UPLOAD_BYTES = 25 * 1024 * 1024  # 25 MB


@app.post("/ai/read-document-safe")
async def read_document_safe(request: Request, file: UploadFile):
    content_length = request.headers.get("content-length")
    if content_length and int(content_length) > MAX_UPLOAD_BYTES:
        raise HTTPException(status_code=413, detail="File too large; usa BYOS (output_url)")

    content = await file.read()
    ...
```

### Large files with BYOS (`output_url`)

To avoid buffering heavy documents in the response of your API, generate a pre-signed URL %0006%% from your own storage (S3, R2, etc.) and pass it as `output_url`. API uploads the Markdown there and returns a small JSON instead of the full body — this automatically activates `response_format=json`, `output_encoding=zstd` and `slim=true`:

```python theme={null}
import boto3
from fastapi import FastAPI, UploadFile, HTTPException

s3 = boto3.client("s3", region_name="us-east-1")


@app.post("/ai/read-document-large")
async def read_document_large(file: UploadFile):
    content = await file.read()

    output_key = f"markdown/{file.filename}.md.zst"
    presigned_put_url = s3.generate_presigned_url(
        "put_object",
        Params={"Bucket": "mi-bucket", "Key": output_key},
        ExpiresIn=600,
    )
    presigned_head_url = s3.generate_presigned_url(
        "head_object",
        Params={"Bucket": "mi-bucket", "Key": output_key},
        ExpiresIn=600,
    )

    async with httpx.AsyncClient(timeout=httpx.Timeout(connect=10.0, read=300.0)) as client:
        res = await client.post(
            f"{FLASH_MD_URL}/convert/raw",
            params={
                "filename": file.filename,
                "mode": "fast",
                "output_url": presigned_put_url,
                "output_head_url": presigned_head_url,
            },
            headers={
                "x-api-key": FLASH_MD_KEY,
                "content-type": file.content_type or "application/octet-stream",
            },
            content=content,
        )

    if res.status_code not in (200, 202):
        raise HTTPException(status_code=res.status_code, detail=_detail_from_response(res))

    # The response is a small JSON that points to the object already uploaded to your bucket.
    return res.json()
```

<Tip>
  `output_head_url` is a presigned URL HEAD that allows API to detect a cache hit before reprocessing: if the output object already exists in your storage, it directly returns the `output_url` without converting the document again.
</Tip>