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

# Unstructured Transform MCP server installation for NVIDIA NeMo Agent Toolkit

> Learn how to connect the Unstructured Transform MCP server to an NVIDIA NeMo Agent Toolkit workflow. Your agents can then point to your files and have Transform start producing partitioned, enriched, chunked, and embedded data based on your files in minutes.

[NVIDIA NeMo Agent Toolkit](https://github.com/NVIDIA/NeMo-Agent-Toolkit) is NVIDIA's open-source framework for
building agent workflows. Unlike end-user AI tools that add the Transform MCP server through a settings screen, the
toolkit connects to it *from a workflow configuration*: you declare the Transform MCP server as a tool source and hand
its tools to an agent. Because Transform is a hosted *remote* MCP server, there is nothing to install or run locally.
You point the client at the server URL and authenticate with your Unstructured API key.

The toolkit ships two building blocks that make this work out of the box, so you do not need a custom transport:

* The `mcp_client` connector, which connects to a remote MCP server over the streamable HTTP transport.
* The `api_key` authentication provider, which sends your Unstructured API key as an `Authorization: Bearer` header on
  every request, including the initial `initialize` handshake.

## Requirements

Before you begin, you must have the following:

* Python 3.11, 3.12, or 3.13 on your local development machine. To check, in your terminal, run `python --version` or `python3 --version`. [Install Python](https://www.python.org/downloads/).
* An Unstructured API key, which the Transform MCP server uses as a bearer token. [Get an API key](/transform/code#get-your-unstructured-api-key-and-url).
* An NVIDIA API key for the agent's underlying model, which the toolkit's `nim` model client uses. [Get an API key](https://build.nvidia.com/).

## Install the packages

The `mcp_client` connector and the `api_key` provider are included in the released toolkit (version 1.8.0 or later). In
your terminal, install the toolkit with the `langchain` and `mcp` extras:

```bash theme={null}
pip install "nvidia-nat[langchain,mcp]"
```

<Note>
  The toolkit is in active development and its APIs change between releases. This guide was verified against version
  1.8.0. If you install a different version, check the
  [NeMo Agent Toolkit release notes](https://github.com/NVIDIA/NeMo-Agent-Toolkit/releases) for changes.
</Note>

## Connect to the Transform MCP server

Store your Unstructured API key in an environment variable rather than hard-coding it. The toolkit reads it when it
loads the workflow configuration:

```bash theme={null}
export UNSTRUCTURED_API_KEY="<your-unstructured-api-key>"
export NVIDIA_API_KEY="<your-nvidia-api-key>"
```

In a workflow configuration file, declare the Transform MCP server as an `mcp_client` function group and reference an
`api_key` authentication provider. The provider attaches the `Authorization: Bearer` header to every request, so the
handshake and the tool calls are both authenticated:

```yaml theme={null}
function_groups:
  unstructured_transform:
    _type: mcp_client
    server:
      transport: streamable-http
      url: https://mcp.transform.unstructured.io
      auth_provider: unstructured_auth
    include:
      - request_file_upload_url
      - transform_files
      - check_transform_status
      - get_transform_results

authentication:
  unstructured_auth:
    _type: api_key
    raw_key: ${UNSTRUCTURED_API_KEY}
    auth_scheme: Bearer
```

The four tools in the `include` list are the tools the transform job lifecycle uses. Listing them makes the
configuration fail fast if the server ever stops exposing one of them.

## Understand the transform job lifecycle

Transforming a document is an asynchronous, multi-step flow. Four of the steps are MCP tool calls, and two are plain
HTTP transfers that are *not* MCP calls:

1. `request_file_upload_url`: Returns a pre-signed `upload_url` and a `file_ref` for a local file.
2. An HTTP `PUT` of the raw file bytes to the pre-signed `upload_url`. This is not an MCP call.
3. `transform_files`: Starts a transform job for one or more `file_ref` values (or public HTTP(S) URLs) and returns a `job_id`.
4. `check_transform_status`: Reports whether the job is `SCHEDULED`, `IN_PROGRESS`, or `COMPLETED`.
5. `get_transform_results`: Returns a pre-signed `download_url` for the Markdown output of each transformed file.
6. An HTTP `GET` of the Markdown from the pre-signed `download_url`. This is not an MCP call.

An agent cannot perform the two raw HTTP transfers through MCP tools alone, and driving the polling loop tool by tool
is slow and unreliable. The recommended approach is to wrap the whole flow in a single function that the agent calls
once. The next section shows how.

<Note>
  The pre-signed `upload_url` and `download_url` carry their own credentials in the URL itself. The `PUT` and `GET`
  must not send the `Authorization` header, or the storage service rejects them.
</Note>

## Expose Transform as a single agent tool

Register a `transform_document` function that runs the full lifecycle deterministically, then give that one tool to an
agent. Create a small package with the following layout:

```text theme={null}
transform_tool/
├── pyproject.toml
└── src/nat_transform_tool/
    ├── __init__.py
    └── register.py
```

In `register.py`, resolve the four Transform tools from the function group and orchestrate the flow:

```python theme={null}
import asyncio
import json
import mimetypes
import typing
from pathlib import Path

import httpx
from pydantic import Field

from nat.builder.builder import Builder
from nat.builder.function import Function
from nat.builder.function import FunctionGroup
from nat.builder.function_info import FunctionInfo
from nat.cli.register_workflow import register_function
from nat.data_models.component_ref import FunctionGroupRef
from nat.data_models.function import FunctionBaseConfig

MAX_FILE_SIZE_BYTES = 52_428_800  # 50 MB
_PENDING = {"SCHEDULED", "IN_PROGRESS"}


class TransformDocumentConfig(FunctionBaseConfig, name="transform_document"):
    mcp_group: FunctionGroupRef = Field(default=FunctionGroupRef("unstructured_transform"))
    poll_interval_seconds: float = Field(default=5.0, gt=0)
    transform_timeout_seconds: float = Field(default=900.0, gt=0)


async def _call(tool: Function, **args: typing.Any) -> dict:
    payload = json.loads(await tool.ainvoke(tool.input_schema(**args)))
    if not isinstance(payload, dict):
        raise RuntimeError(f"Unexpected response from {tool.instance_name}: {str(payload)[:200]!r}")
    if isinstance(payload.get("error"), dict):
        raise RuntimeError(f"{tool.instance_name} error: {payload['error']}")
    return payload


@register_function(config_type=TransformDocumentConfig)
async def transform_document(config: TransformDocumentConfig, builder: Builder):
    group = await builder.get_function_group(config.mcp_group)
    functions = await group.get_accessible_functions()
    tools = {FunctionGroup.decompose(name)[1]: fn for name, fn in functions.items()}

    async def _transform(source: str) -> str:
        """Convert a local file path or a public HTTP(S) URL into Markdown."""
        try:
            async with httpx.AsyncClient(timeout=config.transform_timeout_seconds) as client:
                if source.startswith(("http://", "https://")):
                    file_ref = source
                else:
                    path = Path(source).expanduser()
                    size = (await asyncio.to_thread(path.stat)).st_size
                    if size > MAX_FILE_SIZE_BYTES:
                        return f"ERROR: {source} is {size} bytes, over the 50 MB limit."
                    content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
                    upload = await _call(tools["request_file_upload_url"],
                                         filename=path.name,
                                         content_type=content_type,
                                         size_bytes=size)
                    # The pre-signed URL carries its own credentials: do not send the bearer token here.
                    put = await client.request(upload.get("method", "PUT"),
                                               upload["upload_url"],
                                               content=await asyncio.to_thread(path.read_bytes),
                                               headers=upload.get("headers") or {})
                    put.raise_for_status()
                    file_ref = upload["file_ref"]

                job = await _call(tools["transform_files"], file_refs=[file_ref])
                deadline = asyncio.get_running_loop().time() + config.transform_timeout_seconds
                while True:
                    status = str((await _call(tools["check_transform_status"],
                                              job_id=job["job_id"])).get("status", "")).upper()
                    if status == "COMPLETED":
                        break
                    if status not in _PENDING:
                        return f"ERROR: transform job ended in state {status}."
                    if asyncio.get_running_loop().time() >= deadline:
                        return "ERROR: transform timed out. Raise transform_timeout_seconds for large documents."
                    await asyncio.sleep(config.poll_interval_seconds)

                results = await _call(tools["get_transform_results"], job_id=job["job_id"])
                files = results.get("files") or []
                if not (files and files[0].get("download_url")):
                    return "ERROR: transform completed but returned no downloadable result."
                # The download URL is pre-signed as well: no bearer token here either.
                download = await client.get(files[0]["download_url"])
                download.raise_for_status()
                return download.text
        except (OSError, RuntimeError, httpx.HTTPError) as error:
            return f"ERROR: failed to transform {source}: {error}"

    yield FunctionInfo.from_fn(
        _transform,
        description=("Convert a document (PDF, DOCX, PPTX, XLSX, HTML, images, and 40+ other formats) into "
                     "Markdown. The input is a local file path or a public HTTP(S) URL."))
```

In `pyproject.toml`, register the function so the toolkit can discover it:

```toml theme={null}
[project]
name = "nat_transform_tool"
version = "0.1.0"
requires-python = ">=3.11,<3.14"
dependencies = ["nvidia-nat[langchain,mcp]>=1.8.0", "httpx~=0.27"]

[project.entry-points.'nat.components']
nat_transform_tool = "nat_transform_tool.register"

[tool.setuptools.packages.find]
where = ["src"]
```

Install the package, then add the `transform_document` function and an agent to the same workflow configuration used
above:

```bash theme={null}
pip install -e transform_tool
```

```yaml theme={null}
functions:
  transform_document:
    _type: transform_document
    mcp_group: unstructured_transform

llms:
  nim_llm:
    _type: nim
    model_name: nvidia/nemotron-3-nano-30b-a3b
    temperature: 0.0

workflow:
  _type: react_agent
  tool_names:
    - transform_document
  llm_name: nim_llm
```

Run the workflow, pointing the agent at a local file or a public URL:

```bash theme={null}
nat run --config_file workflow.yml \
  --input "Transform https://arxiv.org/pdf/1706.03762 to Markdown and list the section headings."
```

The agent calls `transform_document` once, and the function handles the upload, transform, polling, and download, then
returns the Markdown for the agent to work with.

## Parsing requests

Parsing requests have the following limits:

* Each file must be of a [supported file type](/transform/supported-file-types).
* Each file must be 50 MB or less in size.
* Each request must have 10 files or fewer.
* Only 5 requests can be running at a time.

## Next steps

* [Control Transform file parsing output](/transform/output): Control how the Unstructured Transform MCP server instructs Transform to partition, enrich, chunk, and embed the data based on your files.
* [Control Transform generated sample code](/transform/code): Control how the Unstructured Transform MCP server generates sample curl or Python code that demonstrates how to use Transform to partition, enrich, chunk, and embed the data based on your files.

## Questions? Need help?

* For technical support, [request support](/support/request).