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
- Resources: Read-only data the server exposes, such as files, database rows, or API responses. Identified by URIs like
file:///reports/q3.pdf. - Tools: Executable functions the model can call, such as
extract_invoiceorquery_database. Tools have typed schemas. - Prompts: Reusable prompt templates the server publishes, allowing clients to standardize extraction instructions.
- Sampling: A server-initiated request asking the client to perform LLM completions, enabling agentic workflows.
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:
- Reusability: One server serves many clients and models.
- Standardization: Consistent tool schemas mean predictable model behavior.
- Security boundary: The server controls exactly what data and actions are exposed.
- Composability: Multiple servers can run side by side, each handling a different source.
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.