Model Context Protocol (MCP) Hands-On Walkthrough for the Unstructured Workflow 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-demo
cd mcp-unstructured-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

from typing import Any, AsyncIterator, Optional
from mcp.server.fastmcp import FastMCP, Context
from dotenv import load_dotenv
import os
from dataclasses import dataclass
from contextlib import asynccontextmanager
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import (
    ListSourcesRequest,
    GetSourceRequest
)
from unstructured_client.models.shared import SourceConnectorType

def load_environment_variables() -> None:
    """
    Loads environment variables from a .env file and checks for required variables. 
    Raises an error if any required variable is missing.

    Args:
        None

    Returns:
        None
    """
    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}")

@dataclass
class AppContext:
    client: UnstructuredClient

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """
    Manages application lifecycle with type-safe context
    for the FastMCP server. Initializes the UnstructuredClient.
    
    Args:
        server (FastMCP): The FastMCP server instance.
    
    Returns:
        AsyncIterator[AppContext]: An asynchronous context manager providing the 
        application context.
    """
    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:
        pass

# Initialize the FastMCP server.
mcp = FastMCP("workflow", lifespan=app_lifespan)

@mcp.tool()
async def list_sources(ctx: Context, source_type: Optional[str] = None) -> str:
    """
    Lists available source connectors. Optionally filters by source type.
    
    Args:
        ctx (Context): The context for the request.
        source_type (Optional[str]): The type of source connector to filter by. 
            If None, all sources are listed.
    
    Returns:
        str: A formatted string listing available source connector names and IDs.
    """
    client = ctx.request_context.lifespan_context.client
    request = ListSourcesRequest()

    if source_type:
        try:
            request.source_type = SourceConnectorType[source_type]
        except KeyError:
            return f"Invalid source type: {source_type}"

    response = await client.sources.list_sources_async(
        request=request
    )
    sorted_sources = sorted(
        response.response_list_sources, 
        key=lambda source: source.name.lower()
    )
    result = ["Available source connectors:"]

    for source in sorted_sources:
        result.append(f"- {source.name} (ID: {source.id})")

    return "\n".join(result)

@mcp.tool()
async def get_source(ctx: Context, source_id: str) -> str:
    """
    Gets detailed information about a specific source connector by its ID.
    
    Args:
        ctx (Context): The context for the request.
        source_id (str): The ID of the source connector to retrieve.
        
    Returns: 
        str: A formatted string containing information about the source connector.
    """
    client = ctx.request_context.lifespan_context.client
    
    response = await client.sources.get_source_async(
        request=GetSourceRequest(source_id=source_id)
    )
    info = response.source_connector_information

    result = ["Source connector information:"]
    result.append(f"- name: {info.name}")
    
    for key, value in info.config:
        result.append(f"{key}: {value}")

    return "\n".join(result)
    
if __name__ == "__main__":
    load_environment_variables()
    # Initialize and run the MCP 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": {
        "workflow": {
            "command": "/ABSOLUTE/PATH/TO/uv",
            "args": [
                "--directory",
                "/ABSOLUTE/PATH/TO/PROJECT/DIRECTORY/mcp-unstructured-demo/",
                "run",
                "workflow.py"
            ]
        }
    }
}

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

List all available source connectors of type S3.

Provide information about the source connector with ID <source-connector-id>.

git clone https://github.com/Unstructured-IO/UNS-MCP.git
cd UNS-MCP

UNSTRUCTURED_API_KEY="<key-here>"

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

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

deactivate

uv sync

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

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

{
    "mcpServers": {
        "UNS-MCP": {
            "command": "/ABSOLUTE/PATH/TO/uv",
            "args": [
                "--directory",
                "/ABSOLUTE/PATH/TO/PARENT/FOLDER/UNS-MCP",
                "run",
                "server.py"
            ]
        }
    }
}

{
    "mcpServers": {
        "workflow": {
            "command": "C:\\ABSOLUTE\\PATH\\TO\\uv",
            "args": [
                "--directory",
                "C:\\ABSOLUTE\\PATH\\TO\\PARENT\\FOLDER\\UNS-MCP",
                "run",
                "server.py"
            ]
        }
    }
}