Unstructured Transform MCP server installation for NVIDIA NeMo Agent Toolkit

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.
An Unstructured API key, which the Transform MCP server uses as a bearer token. Get an API key.
An NVIDIA API key for the agent’s underlying model, which the toolkit’s nim model client uses. Get an API key.

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:

pip install "nvidia-nat[langchain,mcp]"

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 for changes.

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:

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:

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:

request_file_upload_url: Returns a pre-signed upload_url and a file_ref for a local file.
An HTTP PUT of the raw file bytes to the pre-signed upload_url. This is not an MCP call.
transform_files: Starts a transform job for one or more file_ref values (or public HTTP(S) URLs) and returns a job_id.
check_transform_status: Reports whether the job is SCHEDULED, IN_PROGRESS, or COMPLETED.
get_transform_results: Returns a pre-signed download_url for the Markdown output of each transformed file.
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.

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.

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:

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:

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:

[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:

pip install -e transform_tool

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:

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.
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: 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: 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.

​Requirements

​Install the packages

​Connect to the Transform MCP server

​Understand the transform job lifecycle

​Expose Transform as a single agent tool

​Parsing requests

​Next steps

​Questions? Need help?

Requirements

Install the packages

Connect to the Transform MCP server

Understand the transform job lifecycle

Expose Transform as a single agent tool

Parsing requests

Next steps

Questions? Need help?