← Back to DevBytes

Building a Data Extraction Pipeline with MCP (Model Context Protocol): Complete Guide

Introduction to MCP for Data Extraction

The Model Context Protocol (MCP) is an open standard introduced by Anthropic that standardizes how AI models connect to external data sources and tools. Think of it as USB-C for AI applications: a universal interface that lets any model talk to any data source through a consistent protocol. For developers building data extraction pipelines, MCP offers a clean, composable way to expose documents, databases, APIs, and file systems to language models without writing bespoke integrations for every new source.

In this tutorial, you'll learn how to build a production-grade data extraction pipeline using MCP. We'll cover the protocol's architecture, build a custom MCP server that exposes extraction tools, connect an MCP client, and walk through best practices for reliability, security, and performance.

What Is the Model Context Protocol?

MCP defines a client-server architecture where an MCP Server exposes resources, tools, and prompts, and an MCP Client (often embedded inside an AI application or IDE) consumes them. Communication happens over JSON-RPC 2.0, typically transported via stdio or HTTP with Server-Sent Events.

Core Primitives

Why MCP Matters for Data Extraction

Traditional extraction pipelines require hardcoded connectors for every source: a PDF parser, a SQL adapter, an HTTP client for each SaaS API, and custom prompt engineering for each document type. MCP decouples the model from the sources. You write one server per source, and any MCP-compatible client — Claude Desktop, a LangChain agent, or your own application — can immediately use it.

Key benefits include:

Prerequisites and Project Setup

You'll need Python 3.10 or later and the official MCP SDK. Create a new project directory and install dependencies:

mkdir mcp-extraction-pipeline && cd mcp-extraction-pipeline
python -m venv .venv
source .venv/bin/activate
pip install "mcp[cli]" pydantic httpx pypdf

Create the following project structure:

mcp-extraction-pipeline/
├── server/
│   ├── __init__.py
│   ├── extraction_server.py
│   └── extractors/
│       ├── __init__.py
│       ├── pdf_extractor.py
│       └── api_extractor.py
├── client/
│   └── pipeline_client.py
├── data/
│   └── sample_invoice.pdf
└── requirements.txt

Building the PDF Extractor Module

Before defining MCP tools, let's build the underlying extraction logic. We'll create a PDF extractor that pulls structured fields from invoices.

# server/extractors/pdf_extractor.py
from pathlib import Path
import re
from pypdf import PdfReader
from typing import Optional

INVOICE_PATTERNS = {
    "invoice_number": r"Invoice\s*(?:No\.?|Number|#)\s*[:\-]?\s*([A-Z0-9\-]+)",
    "date": r"Date\s*[:\-]?\s*(\d{4}-\d{2}-\d{2})",
    "total": r"Total\s*(?:Amount)?\s*[:\-]?\s*\$?([\d,]+\.\d{2})",
    "vendor": r"(?:From|Vendor)\s*[:\-]?\s*(.+)",
}

def extract_text_from_pdf(file_path: str) -> str:
    reader = PdfReader(file_path)
    pages = []
    for page in reader.pages:
        text = page.extract_text() or ""
        pages.append(text)
    return "\n".join(pages)

def parse_invoice_fields(text: str) -> dict:
    fields = {}
    for key, pattern in INVOICE_PATTERNS.items():
        match = re.search(pattern, text, re.IGNORECASE)
        fields[key] = match.group(1).strip() if match else None
    return fields

def extract_invoice(file_path: str) -> dict:
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(f"PDF not found: {file_path}")
    text = extract_text_from_pdf(str(path))
    fields = parse_invoice_fields(text)
    return {
        "source_file": str(path),
        "page_count": len(PdfReader(str(path)).pages),
        "extracted_fields": fields,
        "raw_text_preview": text[:500],
    }

Building the API Extractor Module

Next, let's add an extractor that pulls structured data from a REST API endpoint — useful for enriching extracted records with external metadata.

# server/extractors/api_extractor.py
import httpx
from typing import Optional

async def fetch_record(base_url: str, record_id: str, api_key: Optional[str] = None) -> dict:
    headers = {}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(f"{base_url.rstrip('/')}/records/{record_id}", headers=headers)
        response.raise_for_status()
        return response.json()

def transform_record(raw: dict) -> dict:
    return {
        "external_id": raw.get("id"),
        "name": raw.get("name"),
        "status": raw.get("status"),
        "metadata": raw.get("metadata", {}),
    }

Creating the MCP Server

Now we wire the extractors into an MCP server. The server exposes two tools — extract_invoice_pdf and fetch_api_record — plus a resource for listing available local PDFs.

# server/extraction_server.py
import os
import glob
from mcp.server.fastmcp import FastMCP
from .extractors.pdf_extractor import extract_invoice
from .extractors.api_extractor import fetch_record, transform_record

mcp = FastMCP("data-extraction-pipeline")

@mcp.tool()
def extract_invoice_pdf(file_path: str) -> dict:
    """Extract structured fields from an invoice PDF.

    Args:
        file_path: Absolute or relative path to the PDF file.

    Returns:
        A dictionary with extracted fields, page count, and a text preview.
    """
    return extract_invoice(file_path)

@mcp.tool()
async def fetch_api_record(
    base_url: str,
    record_id: str,
    api_key: str = ""
) -> dict:
    """Fetch and transform a record from an external REST API.

    Args:
        base_url: The API base URL, e.g. https://api.example.com.
        record_id: The unique identifier of the record.
        api_key: Optional bearer token for authentication.

    Returns:
        A transformed record dictionary.
    """
    key = api_key if api_key else None
    raw = await fetch_record(base_url, record_id, key)
    return transform_record(raw)

@mcp.resource("files://invoices")
def list_invoice_files() -> str:
    """List all PDF files in the configured data directory."""
    data_dir = os.environ.get("EXTRACTION_DATA_DIR", "./data")
    files = glob.glob(os.path.join(data_dir, "*.pdf"))
    return "\n".join(sorted(files))

if __name__ == "__main__":
    mcp.run(transport="stdio")

The FastMCP class handles JSON-RPC wiring automatically. Type hints and docstrings are converted into JSON Schema for the model, so clear documentation directly improves extraction quality.

Running the Server

You can run the server directly for testing:

python -m server.extraction_server

For integration with Claude Desktop, add the server to its configuration file (claude_desktop_config.json):

{
  "mcpServers": {
    "data-extraction-pipeline": {
      "command": "python",
      "args": ["-m", "server.extraction_server"],
      "env": {
        "EXTRACTION_DATA_DIR": "/absolute/path/to/data"
      }
    }
  }
}

Building a Programmatic MCP Client

For production pipelines, you'll want a programmatic client rather than a GUI. The MCP SDK provides a ClientSession for this purpose. Below is a complete client that connects to the server, lists available tools, and runs an extraction.

# client/pipeline_client.py
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_pipeline(pdf_path: str) -> dict:
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "server.extraction_server"],
        env={"EXTRACTION_DATA_DIR": "./data"},
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # List available tools
            tools_result = await session.list_tools()
            print("Available tools:")
            for tool in tools_result.tools:
                print(f"  - {tool.name}: {tool.description}")

            # Call the extraction tool
            result = await session.call_tool(
                "extract_invoice_pdf",
                arguments={"file_path": pdf_path},
            )

            # Parse the structured content returned by the tool
            for content_block in result.content:
                if hasattr(content_block, "text"):
                    return json.loads(content_block.text)

            return {}

if __name__ == "__main__":
    pdf = "./data/sample_invoice.pdf"
    extracted = asyncio.run(run_pipeline(pdf))
    print(json.dumps(extracted, indent=2))

When you run this client against a sample invoice PDF, you'll see output like:

{
  "source_file": "./data/sample_invoice.pdf",
  "page_count": 2,
  "extracted_fields": {
    "invoice_number": "INV-2024-0042",
    "date": "2024-09-15",
    "total": "1,250.00",
    "vendor": "Acme Supplies Inc."
  },
  "raw_text_preview": "INVOICE\nInvoice No: INV-2024-0042\nDate: 2024-09-15\nFrom: Acme Supplies Inc.\n..."
}

Adding an LLM Orchestration Layer

The real power of MCP emerges when you let a language model decide which tools to call. Below is an orchestration layer that uses an OpenAI-compatible API to drive the extraction pipeline autonomously.

# client/orchestrator.py
import asyncio
import json
import os
from openai import AsyncOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

SYSTEM_PROMPT = """You are a data extraction assistant. Use the available MCP tools
to extract structured data from sources. Always validate that required fields
are present before returning results. If a field is missing, note it explicitly."""

async def orchestrate(user_instruction: str) -> str:
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "server.extraction_server"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools_result = await session.list_tools()

            # Convert MCP tools to OpenAI function-calling format
            openai_tools = []
            tool_map = {}
            for tool in tools_result.tools:
                openai_tools.append({
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description or "",
                        "parameters": tool.inputSchema,
                    },
                })
                tool_map[tool.name] = tool

            messages = [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_instruction},
            ]

            for _ in range(5):  # Limit tool-call rounds
                response = await client.chat.completions.create(
                    model="gpt-4o",
                    messages=messages,
                    tools=openai_tools,
                )
                choice = response.choices[0]
                messages.append(choice.message)

                if not choice.message.tool_calls:
                    return choice.message.content or ""

                for call in choice.message.tool_calls:
                    args = json.loads(call.function.arguments)
                    result = await session.call_tool(call.function.name, args)
                    text_parts = [
                        block.text for block in result.content
                        if hasattr(block, "text")
                    ]
                    messages.append({
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": "\n".join(text_parts),
                    })

            return "Maximum tool-call rounds reached."

if __name__ == "__main__":
    instruction = "Extract all fields from ./data/sample_invoice.pdf and summarize the vendor and total."
    print(asyncio.run(orchestrate(instruction)))

Error Handling and Validation

Production pipelines must handle malformed inputs gracefully. Wrap tool logic with validation and return structured error objects rather than raising exceptions that crash the server.

# server/extractors/pdf_extractor.py (updated)
from pydantic import BaseModel, ValidationError, Field
from typing import Optional

class InvoiceExtractionResult(BaseModel):
    source_file: str
    page_count: int = Field(ge=0)
    extracted_fields: dict
    raw_text_preview: str
    errors: list[str] = []

def extract_invoice_safe(file_path: str) -> dict:
    errors = []
    try:
        path = Path(file_path)
        if not path.suffix.lower() == ".pdf":
            errors.append("File is not a PDF")
        if not path.exists():
            errors.append(f"File not found: {file_path}")
            return InvoiceExtractionResult(
                source_file=file_path,
                page_count=0,
                extracted_fields={},
                raw_text_preview="",
                errors=errors,
            ).model_dump()
        text = extract_text_from_pdf(str(path))
        fields = parse_invoice_fields(text)
        return InvoiceExtractionResult(
            source_file=str(path),
            page_count=len(PdfReader(str(path)).pages),
            extracted_fields=fields,
            raw_text_preview=text[:500],
            errors=errors,
        ).model_dump()
    except Exception as exc:
        errors.append(f"Unexpected error: {str(exc)}")
        return InvoiceExtractionResult(
            source_file=file_path,
            page_count=0,
            extracted_fields={},
            raw_text_preview="",
            errors=errors,
        ).model_dump()

Update the tool to call extract_invoice_safe instead of extract_invoice. This guarantees the model always receives a parseable JSON response, even when the input is invalid.

Best Practices

Design Tool Schemas Carefully

The model relies entirely on tool names, descriptions, and parameter schemas to decide what to call. Use descriptive names like extract_invoice_pdf rather than extract. Write docstrings that explain expected inputs, outputs, and failure modes. Avoid optional parameters unless truly optional — every ambiguous parameter increases the chance of a malformed call.

Keep Tools Atomic

Each tool should do one thing well. Instead of a single extract_everything tool, expose extract_invoice_pdf, fetch_api_record, and list_invoice_files separately. The model can compose them, and you can test each in isolation.

Enforce Security Boundaries

MCP servers run with the permissions of their host process. Never expose arbitrary file system access. Validate all paths against an allowlist of directories. For API extractors, store credentials in environment variables or a secrets manager — never hardcode them in tool arguments or source files.

ALLOWED_DIRS = [
    os.path.abspath("./data"),
    os.path.abspath("./uploads"),
]

def validate_path(file_path: str) -> str:
    abs_path = os.path.abspath(file_path)
    if not any(abs_path.startswith(d) for d in ALLOWED_DIRS):
        raise PermissionError(f"Access denied: {file_path}")
    return abs_path

Log Every Tool Invocation

For observability and debugging, log tool calls with timestamps, arguments, and result summaries. This is invaluable when diagnosing why a model made unexpected extraction decisions.

import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("extraction-pipeline")

@mcp.tool()
def extract_invoice_pdf(file_path: str) -> dict:
    logger.info("extract_invoice_pdf called: %s", file_path)
    result = extract_invoice_safe(file_path)
    logger.info("extraction complete, errors=%d", len(result.get("errors", [])))
    return result

Cache Expensive Operations

If your extractors call rate-limited APIs or parse large PDFs repeatedly, add caching. A simple in-memory LRU cache works for single-process servers; Redis is better for distributed deployments.

from functools import lru_cache

@lru_cache(maxsize=128)
def extract_invoice_cached(file_path: str) -> dict:
    return extract_invoice_safe(file_path)

Test Tools Independently

MCP tools are just Python functions. Write unit tests that call them directly, bypassing the protocol layer. This keeps tests fast and isolates extraction logic from transport concerns.

# tests/test_pdf_extractor.py
import pytest
from server.extractors.pdf_extractor import extract_invoice_safe

def test_missing_file_returns_error():
    result = extract_invoice_safe("/nonexistent/file.pdf")
    assert result["errors"]
    assert any("not found" in e for e in result["errors"])

def test_valid_pdf_extracts_fields(tmp_path):
    # Create a minimal PDF or use a fixture
    result = extract_invoice_safe("data/sample_invoice.pdf")
    assert result["page_count"] > 0
    assert result["extracted_fields"]["invoice_number"] is not None

Deploying the Pipeline

For production, package the server as a Docker container. This ensures consistent runtime environments and simplifies deployment.

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
ENV EXTRACTION_DATA_DIR=/app/data

CMD ["python", "-m", "server.extraction_server"]

Build and run:

docker build -t mcp-extraction-pipeline .
docker run -i --rm \
  -v $(pwd)/data:/app/data \
  -e EXTRACTION_DATA_DIR=/app/data \
  mcp-extraction-pipeline

The -i flag keeps stdin open, which is required for stdio transport. For HTTP-based deployments, switch the transport in your server entry point:

if __name__ == "__main__":
    mcp.run(transport="sse", port=8080)

Conclusion

The Model Context Protocol transforms data extraction from a tangle of bespoke connectors into a clean, standardized architecture. By exposing extractors as MCP tools, you give any compatible client — whether Claude Desktop, a custom orchestrator, or an agentic framework — immediate access to your pipeline. The separation of concerns between extraction logic (the server) and orchestration logic (the client) makes each component independently testable, reusable, and deployable. Start with atomic, well-documented tools, enforce strict security boundaries, and add observability from day one. As your pipeline grows, you can compose multiple MCP servers — one per data source — and let the model route between them automatically. This composable, protocol-driven approach is the foundation of robust, maintainable extraction systems in the age of large language models.

— Ad —

Google AdSense will appear here after approval

← Back to all articles