MCP Data Integration Demo

1. Refined understanding of the requirement

This project demonstrates a small MCP-based data integration system. The MCP client starts and connects to a FastMCP server, discovers tools, and orchestrates a workflow that reads employee data from Excel, CSV, JSON, and SQLite sources. The server owns all data operations: source reading, normalization, validation, merging, and export.

Every source can use different column names, but all records are converted to this fixed schema:

employee_id, employee_name, department, salary, joining_date, source_system

Related MCP server: MySQL MCP Server

2. Proposed MCP architecture

User
  -> MCP Client
  -> FastMCP Server
  -> Source-specific tools
  -> Data parsers
  -> Schema normalizer
  -> Validator
  -> Data merger
  -> Final unified dataset

MCP is useful here because it separates orchestration from implementation. The client only decides which tools to call and in what order. The server exposes stable data-operation tools that can be used by this client, another MCP client, or an AI application.

3. Project folder structure

.
|-- server.py
|-- client.py
|-- schema.py
|-- normalizer.py
|-- validators.py
|-- merger.py
|-- exporter.py
|-- generate_sample_data.py
|-- requirements.txt
|-- data_sources/
|   |-- excel_reader.py
|   |-- csv_reader.py
|   |-- sql_reader.py
|   |-- json_reader.py
|   |-- powerbi_reader.py
|   `-- common.py
|-- sample_data/
`-- output/

4. Complete code for each file

The complete runnable code is in the project files above. The key roles are:

• server.py: FastMCP tool provider.

• client.py: MCP workflow orchestrator.

• data_sources/*: source-specific parsers.

• normalizer.py: maps source columns into the fixed schema.

• schema.py and validators.py: Pydantic schema and validation.

• merger.py: combines valid records and removes duplicate employee_id values.

• exporter.py: writes CSV or Excel output.

5. Sample data generation code

generate_sample_data.py creates:

• sample_data/employees.xlsx

• sample_data/employees.csv

• sample_data/employees.json

• sample_data/employees.db

The JSON sample intentionally includes one invalid salary so validation output is visible.

6. Installation steps

python -m venv .venv
.\.venv\Scripts\activate
python -m pip install -r requirements.txt

7. Execution steps

Generate demo input files:

python generate_sample_data.py

Run the MCP client workflow:

python client.py

Run only selected sources:

python client.py --no-json
python client.py --excel sample_data/employees.xlsx --csv sample_data/employees.csv --no-sql --no-json

Export to Excel instead of CSV:

python client.py --output output/final_employees.xlsx

8. Step-by-step explanation of how the code works

1. client.py starts server.py as a FastMCP stdio server.

2. The client calls list_tools() and prints the available MCP tools.

3. For each configured source, the client calls the matching read tool.

4. The client sends returned records to normalize_data.

5. The client sends normalized records to validate_data.

6. Valid records from each source are passed to merge_data_sources.

7. The merged records are passed to export_final_dataset.

8. The client prints a structured final status object.

9. Example output

{
  "status": "success",
  "sources_processed": ["excel", "csv", "json", "sql"],
  "total_records_read": 11,
  "total_records_valid": 10,
  "total_records_invalid": 1,
  "duplicates_removed": 1,
  "final_records_exported": 9,
  "output_file": "C:\\MCP_prac\\output\\final_employees.csv"
}

10. How to extend the project for Power BI, APIs, and more databases

For Power BI, the demo already includes read_powerbi_data, which reads CSV or Excel exports. A direct Power BI API version would add an authenticated reader that calls the Power BI REST API, converts the JSON response to records, then reuses the existing normalizer and validator.

For REST APIs, add a new reader such as data_sources/api_reader.py, register a read_api_data tool in server.py, and add source aliases in normalizer.py.

For more databases, replace or extend sql_reader.py with SQLAlchemy connection support. The rest of the pipeline can remain unchanged as long as the reader returns list[dict] records.

11. Best practices and improvements for production

• Move source mappings to configuration files or a metadata database.

• Add authentication and secret management for external systems.

• Add richer duplicate rules, such as source priority and record timestamps.

• Store invalid records in a rejection file for review.

• Add tests for every reader, mapping, validation rule, and merge rule.

• Add structured logging and run IDs for auditability.

• Add batch processing for large files.