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

# Async client

> Process documents concurrently with AsyncNdiClient

`AsyncNdiClient` exposes the same namespaces and methods as `NdiClient`, using
`await` for network calls. Use it for web services, concurrent document
processing, or any application that already runs an asyncio event loop.

## Basic usage

```python theme={"dark"}
import asyncio

from ndi_sdk import AsyncNdiClient


async def main() -> None:
    async with AsyncNdiClient() as client:
        upload = await client.documents.create_upload("report.pdf")
        queued = await client.documents.parse(upload)
        job = await client.jobs.wait(queued.job_id, timeout=600)
        print(job.result.markdown)


asyncio.run(main())
```

The async client mirrors the sync client. For example,
`client.documents.parse(...)` becomes
`await client.documents.parse(...)`, and `client.jobs.iter_all()` is consumed
with `async for`.

## Process documents concurrently

Share one client across tasks. Each operation creates its own job, and each
`jobs.wait` polls independently.

```python theme={"dark"}
import asyncio

from ndi_sdk import AsyncNdiClient


async def parse_one(client: AsyncNdiClient, path: str) -> str:
    upload = await client.documents.create_upload(path)
    queued = await client.documents.parse(upload)
    job = await client.jobs.wait(queued.job_id, timeout=600)
    return job.result.markdown


async def main() -> None:
    files = ["invoice.pdf", "contract.pdf", "report.pdf"]
    async with AsyncNdiClient() as client:
        results = await asyncio.gather(
            *(parse_one(client, path) for path in files),
        )
    print([len(result) for result in results])


asyncio.run(main())
```

## Limit concurrency

Use an `asyncio.Semaphore` when processing a large batch. This bounds the
number of jobs submitted at once and helps your application stay within its
NDI concurrent-job limit.

```python theme={"dark"}
import asyncio

from ndi_sdk import AsyncNdiClient


async def parse_batch(paths: list[str], max_concurrent: int = 5) -> list[str]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async with AsyncNdiClient() as client:
        async def parse_one(path: str) -> str:
            async with semaphore:
                upload = await client.documents.create_upload(path)
                queued = await client.documents.parse(upload)
                job = await client.jobs.wait(queued.job_id, timeout=600)
                return job.result.markdown

        return await asyncio.gather(*(parse_one(path) for path in paths))
```

The semaphore limits the complete upload, submission, and wait cycle. If you
only want to limit submission, release it before `jobs.wait`.

## Handle partial batch failures

By default, `asyncio.gather` raises when one task fails. Use
`return_exceptions=True` when each document should succeed or fail
independently.

```python theme={"dark"}
async with AsyncNdiClient() as client:
    results = await asyncio.gather(
        *(parse_one(client, path) for path in files),
        return_exceptions=True,
    )

for path, result in zip(files, results, strict=True):
    if isinstance(result, Exception):
        print(f"{path}: failed: {result}")
    else:
        print(f"{path}: {len(result)} characters")
```

See [Error handling](/sdks/python/error-handling) for the typed exceptions
returned by failed HTTP requests and jobs.

## Close the client

Prefer `async with`, which closes the SDK-owned `httpx.AsyncClient`. If you
cannot use a context manager, call `await client.aclose()`.

```python theme={"dark"}
client = AsyncNdiClient()
try:
    job = await client.jobs.get(job_id)
finally:
    await client.aclose()
```

If you pass your own `httpx.AsyncClient`, you own its lifecycle.

## When to use async

<CardGroup cols={2}>
  <Card title="Use AsyncNdiClient" icon="check">
    Concurrent batches, FastAPI or other async services, and applications that
    already use asyncio.
  </Card>

  <Card title="Use NdiClient" icon="terminal">
    One-off scripts, notebooks, and sequential processing where async would add
    unnecessary complexity.
  </Card>
</CardGroup>

## Next steps

* [Job management](/sdks/python/job-management)
* [Error handling](/sdks/python/error-handling)
* [Parse documents](/sdks/python/parse)