MCP Data Workflow Server

A small, readable Model Context Protocol server built on the Streamable HTTP transport. It lets an AI agent ingest a file, inspect it, generate a schema, run a multi-step workflow in the background, and then verify that the workflow really finished, using the same tool-calling pattern I use for production data platforms.

It is deliberately generic and uses only the Python standard library for data handling, so you can read the whole thing in 15 minutes and adapt it to your own backend.

Why this exists

Agents are good at deciding what to do and bad at knowing whether it worked. This server demonstrates three patterns that matter in practice:

1. Async workflows with status tracking. run_workflow returns a run_id immediately; the agent polls get_workflow_status.

2. A separate verification tool. verify_workflow independently checks that every step succeeded and produced its artifact, so an agent can't just claim success.

3. Safe-by-default exposure. Bearer-token auth, a sandboxed data directory, and refusal to start unauthenticated.

Related MCP server: tabulite-mcp

Architecture

flowchart LR
    A[AI agent / MCP client] -- "Streamable HTTP + Bearer token" --> B[Auth middleware]
    B --> C[FastMCP server /mcp]
    C --> D[Tools / Resources / Prompts]
    D --> E[services.py<br/>pure logic, no MCP imports]
    E --> F[(Store: datasets + runs<br/>in-memory)]
    E --> G[/Sandboxed data dir<br/>CSV/TSV/]

Tools

Also exposed: a resource (workflow://datasets/{dataset_id}/ddl) and a prompt (onboard_new_file).

Quickstart

git clone https://github.com/Jaysharma2/mcp-data-workflow-server.git
cd mcp-data-workflow-server
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

export MCP_API_KEY=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
python -m mcp_data_workflow
# -> serving on http://127.0.0.1:8000/mcp

In another terminal:

MCP_API_KEY=<same key> python examples/client.py

Or explore interactively with the MCP Inspector:

npx -y @modelcontextprotocol/inspector
# connect to http://127.0.0.1:8000/mcp (Streamable HTTP) and add the header
#   Authorization: Bearer <your key>

Connect Claude Code

claude mcp add --transport http data-workflow http://127.0.0.1:8000/mcp \
  --header "Authorization: Bearer $MCP_API_KEY"

Then try: "Use the onboard_new_file prompt for orders.csv."

Docker

docker build -t mcp-data-workflow-server .
docker run --rm -p 8000:8000 -e MCP_API_KEY=your-secret \
  -v "$PWD/my_data:/data:ro" mcp-data-workflow-server

Configuration

Example session

agent -> ingest_file(filename="orders.csv")
      <- {"dataset_id": "a1b2c3d4e5f6", "row_count": 80, "column_count": 7, ...}
agent -> run_workflow(dataset_id="a1b2c3d4e5f6")
      <- {"run_id": "9f8e7d6c5b4a", "status": "queued", ...}
agent -> verify_workflow(run_id="9f8e7d6c5b4a")
      <- {"verified": true, "checks": [{"step": "profile", "ok": true}, ...]}
agent -> get_workflow_artifact(run_id="9f8e7d6c5b4a", artifact="ddl")
      <- CREATE TABLE orders (
           order_id BIGINT NOT NULL,
           customer_id VARCHAR NOT NULL,
           order_date DATE NOT NULL,
           ...
         );

Security notes

• Auth is mandatory by default. The server exits at startup if MCP_API_KEY is not set.

• Path sandboxing. Filenames are resolved inside MCP_DATA_DIR; ../ traversal and non-CSV/TSV files are rejected (tested).

• Constant-time key comparison via hmac.compare_digest.

• Use TLS in front of it (a reverse proxy or load balancer) for anything beyond localhost.

• A single shared API key is fine for a demo. For multi-user production use, switch to OAuth 2.1 as described in the MCP authorization spec.

• The server runs in stateless mode, so any replica can handle any request. Workflow state lives in the in-memory Store; with more than one replica you must replace it with a shared backend (see below).

Testing

pytest -q          # unit tests + end-to-end tests over real HTTP with the official MCP client
ruff check .

The end-to-end test starts the real server in a thread, connects with the official mcp client over Streamable HTTP, and drives the whole ingest → run → verify flow, including the 401 and path-traversal cases.

Design decisions

• services.py has no MCP imports. Business logic is testable on its own and portable to another framework.

• Standard library only for data handling (csv, re). No pandas, so the image stays small. Swap in pandas/Polars/DuckDB for larger files.

• Heuristic categorization returns reasons. Every classification is explainable, which makes it easy for an agent (or a human) to sanity-check.

• Validation failures are data, not exceptions. A failed step is recorded on the run, so verify_workflow reports it honestly.

SDK version note

This project targets the v1.x line of the official Python SDK, pinned as mcp>=1.28,<2. The SDK maintainers now document a v2 release; check their migration notes before relaxing the pin.

Roadmap

• Persistent Store (Redis or Postgres) for multi-replica deployments

• Real backend steps (e.g., submit a Spark/Trino job, trigger an Airflow DAG)

• OAuth 2.1 authorization

• Structured logging and OpenTelemetry traces

• Parquet/JSON input

License

MIT, see LICENSE.