Model Context Protocol (MCP) Hands-On Walkthrough for the Unstructured Partition Endpoint

pip install uv

# For macOS and Linux:
curl -LsSf https://astral.sh/uv/install.sh | sh

# For Windows:
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

uv init mcp-unstructured-partition-demo
cd mcp-unstructured-partition-demo

# For macOS and Linux:
uv venv
source .venv/bin/activate

# For Windows:
uv venv
.venv\Scripts\activate

deactivate

uv add "mcp[cli]" "unstructured-client>=0.30.6" dotenv

import os
from dotenv import load_dotenv
import json
from unstructured_client import UnstructuredClient
from typing import AsyncIterator
from dataclasses import dataclass
from contextlib import asynccontextmanager
from mcp.server.fastmcp import FastMCP, Context
from unstructured_client.models import operations, shared

@dataclass
class AppContext:
    client: UnstructuredClient

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manages the Unstructured API client lifecycle."""
    api_key = os.getenv("UNSTRUCTURED_API_KEY")
    if not api_key:
        raise ValueError("UNSTRUCTURED_API_KEY environment variable is required")

    client = UnstructuredClient(api_key_auth=api_key)
    try:
        yield AppContext(client=client)
    finally:
        # No cleanup needed for the API client.
        pass

# Create the MCP server instance.
mcp = FastMCP("unstructured-partition-mcp", lifespan=app_lifespan, dependencies=["unstructured-client", "python-dotenv"])
# Specify the absolute path to the local directory to store processed files.
PROCESSED_FILES_FOLDER = "<absolute-path-to-local-directory>"

def load_environment_variables() -> None:
    """
    Load environment variables from .env file.
    Raises an error if critical environment variables are missing.
    """
    load_dotenv()
    required_vars = [
        "UNSTRUCTURED_API_KEY"
    ]

    for var in required_vars:
        if not os.getenv(var):
            raise ValueError(f"Missing required environment variable: {var}")

def json_to_text(file_path) -> str:
    with open(file_path, 'r') as file:
        elements = json.load(file)

    doc_texts = []

    for element in elements:
        text = element.get("text", "").strip()
        element_type = element.get("type", "")
        metadata = element.get("metadata", {})

        if element_type == "Title":
            doc_texts.append(f"<h1> {text}</h1><br>")
        elif element_type == "Header":
            doc_texts.append(f"<h2> {text}</h2><br/>")
        elif element_type == "NarrativeText" or element_type == "UncategorizedText":
            doc_texts.append(f"<p>{text}</p>")
        elif element_type == "ListItem":
            doc_texts.append(f"<li>{text}</li>")
        elif element_type == "PageNumber":
            doc_texts.append(f"Page number: {text}")
        elif element_type == "Table":
            table_html = metadata.get("text_as_html", "")
            doc_texts.append(table_html)  # Keep the table as HTML.
        else:
            doc_texts.append(text)

    return " ".join(doc_texts)

@mcp.tool()
async def process_document(ctx: Context, filepath: str) -> str:
    """
    Sends the document to Unstructured for processing. 
    Returns the processed contents of the document
    
    Args:
        filepath: The local path to the document.
    """

    filepath = os.path.abspath(filepath)

    if not os.path.isfile(filepath):
        return "File does not exist"

    # Check whether Unstructured supports the file's extension.
    _, ext = os.path.splitext(filepath)
    supported_extensions = {".abw", ".bmp", ".csv", ".cwk", ".dbf", ".dif", ".doc", ".docm", ".docx", ".dot",
                            ".dotm", ".eml", ".epub", ".et", ".eth", ".fods", ".gif", ".heic", ".htm", ".html",
                            ".hwp", ".jpeg", ".jpg", ".md", ".mcw", ".mw", ".odt", ".org", ".p7s", ".pages",
                            ".pbd", ".pdf", ".png", ".pot", ".potm", ".ppt", ".pptm", ".pptx", ".prn", ".rst",
                            ".rtf", ".sdp", ".sgl", ".svg", ".sxg", ".tiff", ".txt", ".tsv", ".uof", ".uos1",
                            ".uos2", ".web", ".webp", ".wk2", ".xls", ".xlsb", ".xlsm", ".xlsx", ".xlw", ".xml",
                            ".zabw"}

    if ext.lower() not in supported_extensions:
        return "File extension not supported by Unstructured"

    client = ctx.request_context.lifespan_context.client
    file_basename = os.path.basename(filepath)

    req = operations.PartitionRequest(
        partition_parameters=shared.PartitionParameters(
            files=shared.Files(
                content=open(filepath, "rb"),
                file_name=filepath,
            ),
            strategy=shared.Strategy.AUTO,
        ),
    )

    try:
        res = client.general.partition(request=req)
        element_dicts = [element for element in res.elements]
        json_elements = json.dumps(element_dicts, indent=2)
        output_json_file_path = os.path.join(PROCESSED_FILES_FOLDER, f"{file_basename}.json")
        with open(output_json_file_path, "w") as file:
            file.write(json_elements)

        return json_to_text(output_json_file_path)
    except Exception as e:
        return f"The following exception happened during file processing: {e}"

if __name__ == "__main__":
    load_environment_variables()
    # Initialize and run the server.
    mcp.run(transport='stdio')

UNSTRUCTURED_API_KEY="<your-unstructured-api-key>"

# For macOS or Linux:
code ~/Library/Application\ Support/Claude/claude_desktop_config.json

# For Windows:
code $env:AppData\Claude\claude_desktop_config.json

{
    "mcpServers": {
        "unstructured-partition-mcp": {
            "command": "/ABSOLUTE/PATH/TO/uv",
            "args": [
                "--directory",
                "/ABSOLUTE/PATH/TO/PROJECT/DIRECTORY/mcp-unstructured-partition-demo/",
                "run",
                "doc_processor.py"
            ]
        }
    }
}

{
    "mcpServers": {
        "unstructured-partition-mcp": {
            "command": "C:\\ABSOLUTE\\PATH\\TO\\uv",
            "args": [
                "--directory",
                "C:\\ABSOLUTE\\PATH\\TO\\PROJECT\\DIRECTORY\\mcp-unstructured-partition-demo\\",
                "run",
                "doc_processor.py"
            ]
        }
    }
}

Process the document at <absolute-path-to-local-file>