# MCP Data Workflow Server

[![CI](https://github.com/Jaysharma2/mcp-data-workflow-server/actions/workflows/ci.yml/badge.svg)](https://github.com/Jaysharma2/mcp-data-workflow-server/actions/workflows/ci.yml)
![Python](https://img.shields.io/badge/python-3.10%2B-blue)
![License](https://img.shields.io/badge/license-MIT-green)

A small, readable [Model Context Protocol](https://modelcontextprotocol.io) 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.

## Architecture

```mermaid
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/]
```

| File | Responsibility |
|---|---|
| `src/mcp_data_workflow/server.py` | MCP tool/resource/prompt definitions, ASGI app, `/healthz` |
| `src/mcp_data_workflow/services.py` | Ingestion, type inference, categorization, DDL, workflow engine |
| `src/mcp_data_workflow/auth.py` | Constant-time bearer-token ASGI middleware |
| `src/mcp_data_workflow/config.py` | Environment-based settings |

## Tools

| Tool | Description |
|---|---|
| `list_files` | List CSV/TSV files available in the data directory |
| `ingest_file` | Ingest a file, infer column types, register a dataset |
| `list_datasets` | List datasets ingested so far |
| `profile_table` | Per-column type, null count, distinct count, samples |
| `categorize_table` | Label a table `fact` / `dimension` / `lookup` / `reference`, with reasons |
| `generate_schema` | `CREATE TABLE` DDL for `trino`, `spark`, or `postgres` |
| `run_workflow` | Start a background run of `profile`, `categorize`, `schema`, `validate` |
| `get_workflow_status` | Per-step status and timestamps for a run |
| `verify_workflow` | Wait for completion, then verify all expected artifacts exist |
| `get_workflow_artifact` | Fetch one artifact (`profile`, `category`, `ddl`, `validation`) |

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

## Quickstart

```bash
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:

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

Or explore interactively with the [MCP Inspector](https://github.com/modelcontextprotocol/inspector):

```bash
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

```bash
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

```bash
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

| Variable | Default | Meaning |
|---|---|---|
| `MCP_API_KEY` | *(required)* | Bearer token clients must send |
| `MCP_ALLOW_ANONYMOUS` | unset | `1` to run without auth, **local development only** |
| `MCP_DATA_DIR` | `./examples/sample_data` | Directory files are read from (sandboxed) |
| `MCP_HOST` / `MCP_PORT` | `127.0.0.1` / `8000` | Bind address |
| `MCP_MAX_ROWS_SCANNED` | `100000` | Row cap per file for profiling |

## Example session

```text
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

```bash
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](LICENSE).