# Legacy-to-FHIR Mapping MCP Server

![Demo: querying the legacy database in natural language and getting back validated FHIR](docs/demo.gif)

<i>Note: For this demo, a simplified local REPL (`scripts/demo.py`) that calls the
tool functions directly, with a keyword-based stand-in for tool selection —
the real natural-language routing is AI-driven via an MCP client, documented in [Usage](#usage) below.</i>

A Model Context Protocol (MCP) server that maps fragmented, inconsistently-formatted
legacy healthcare records into valid, schema-conformant FHIR resources that are queryable
by an AI agent using natural language.

## Problem

Legacy healthcare systems store patient data in fragmented, non-standardized
formats that predate modern interoperability standards like FHIR. Hospitals
and payers still run production systems on these decades-old schemas, and
migrating them wholesale is slow, expensive, and high-risk.

With the increasing adoption of AI agents, plugging them in directly to
legacy systems risks two failure modes: the LLM either can't reach the data
at all, or has to write raw SQL against an undocumented, inconsistent
schema, with no guarantee the result is even valid healthcare data.
Therefore, in order to *use* legacy healthcare data safely, we need a
translation layer in front of it. Someone has to do the hard work of
defining what "correct" means for that data and enforcing it. This is a
product problem as much as an engineering one: deciding when the system is
allowed to guess, and if it should be guessing at all.

This project builds a working version of that translation layer end-to-end:
a synthetic legacy database with realistic messiness (inconsistent date
formats, demographic data split across unlinked tables, units embedded in
free-text values, un-coded diagnosis notes), and an MCP server that maps it
cleanly into FHIR. The system never guesses when a mapping is ambiguous, and
never returns a resource that hasn't independently passed schema validation.

## Architecture

The system is a three-stage pipeline, not a single black box — each stage is a
separate, independently testable unit with one job:

```mermaid
flowchart TD
    Q["Natural-language query"] --> S["MCP server\nmcp_server/server.py"]
    S -->|"patient / condition query"| T1["query_legacy_patient_records\n(Patient + Condition)"]
    S -->|"vitals / labs query"| T2["translate_vitals_log\n(Observation)"]

    T1 --> R["legacy_repo.py\nresolves query against mock legacy\nSQLite database (data/legacy.db)"]
    T2 --> R

    R --> B["fhir_build.py\nmaps raw legacy fields to FHIR elements\nvia codes.py (LOINC/SNOMED-CT) + datetimes.py"]
    B --> V{"validation.py\nindependent FHIR R4B\nschema check"}

    V -->|"pass"| OK["Validated FHIR Bundle\nreturned to caller"]
    V -->|"fail"| BAD["Excluded from bundle,\nreported in validation_failures"]
```

**1. Retrieve** — the MCP server exposes two narrowly-scoped tools
(`query_legacy_patient_records`, `translate_vitals_log`) rather than one generic
"query anything" tool. `legacy_repo.py` resolves the natural-language query (a
name, an MRN, or both) against the mock legacy SQLite database.

**2. Translate** — `fhir_build.py` maps raw legacy fields to FHIR elements,
resolving clinical codes via `codes.py` (LOINC for vitals, SNOMED-CT for
conditions) and normalizing four different legacy date formats via
`datetimes.py`.

**3. Validate** — `validation.py` independently re-checks every resource against
the FHIR R4B schema before it's allowed to leave the system, as a step that does
not trust the mapping code that produced the resource. Anything that fails is
excluded and reported, never silently returned as if it were valid.

The mock legacy data itself comes from [Synthea](https://github.com/synthetichealth/synthea)
(MITRE's open-source synthetic patient generator), then deliberately degraded
by `legacy_data/mangle.py` into a fragmented, legacy-system-shaped schema — see
[Legacy data design](#legacy-data-design) in the technical reference below.

## Why this approach

The interesting decisions in this project weren't really about FHIR — they were
about how much a system should be allowed to assume on its own, and where to
draw tool boundaries for an AI agent. Five decisions worth walking through:

| Decision | Alternative considered | Why this one won |
|---|---|---|
| Two narrowly-scoped tools (`query_legacy_patient_records`, `translate_vitals_log`) instead of one generic query tool | A single flexible `query(sql_or_natural_language)` tool | Tool *descriptions* are the interface contract an LLM uses to route a request. A generic tool produces ambiguous routing and untestable behavior; narrow tools with precise docstrings make routing reliable — verified directly by testing natural-language queries against the live MCP connection and confirming correct tool selection every time |
| Deterministic, rule-based mapping (lookup tables + parsers) instead of having an LLM interpret/map the legacy data | Prompting an LLM to extract and code each record on the fly | Reliability and auditability. LLM-based extraction of clinical codes is non-deterministic and expensive to verify at scale; a healthcare data pipeline that can't guarantee the same input always produces the same output isn't one you can trust in a regulated domain — deterministic logic plus an independent validation gate can |
| "Flag, never guess" as the core product principle for every ambiguous case | Best-effort inference (e.g. assume a missing unit from the vital's name) | A wrong guess in healthcare data is worse than a visible gap — a flagged gap can be caught and reviewed downstream; a silently wrong value usually can't. This is a risk decision as much as an engineering one |
| Validation as a fully independent second pass, not trusted to the mapping code that built the resource | Trusting `fhir_build.py`'s own correctness | Defense in depth — the mapping layer is treated as an untrusted component, the same posture you'd want for any pipeline writing into a compliance-sensitive format |
| Scoped to exactly 3 FHIR resource types (`Patient`, `Observation`, `Condition`) instead of the ~150-resource FHIR spec | Attempt broader (shallower) resource coverage | Proving the pattern deeply on a representative slice is more convincing than covering more resource types shallowly — a scope cut made deliberately, not by running out of time |

## Tradeoffs

Being upfront about what this project deliberately does *not* solve:

- **Not a general legacy-schema translator.** The mapping logic is built against
  the specific mess patterns this dataset contains. A new legacy field format
  outside those patterns wouldn't be handled — a production system ingesting
  arbitrary unknown schemas would need much broader lookup coverage, or a
  human-in-the-loop review step for genuinely novel free text.
- **Not complete interoperability coverage.** Three resource types proves the
  pattern; a commercially viable version would need to cover far more of the
  FHIR resource surface (`MedicationRequest`, `Encounter`, `AllergyIntolerance`,
  and more) before it replaces anything a hospital actually depends on.
- **Not tested at real-world scale or messiness.** 42 synthetic patients from
  one generator's mangling logic is not representative of the volume or edge
  cases in real legacy data — the fuzzy-matching join logic in particular would
  need real stress-testing before this pattern could be trusted further.
- **No performance engineering.** SQLite plus in-process Python is adequate for
  a demo; it says nothing about throughput at hospital-system scale.
- **No auth/access-control layer.** Deliberately out of scope for a read-only
  portfolio project against mock data — but a real product handling any patient
  data, de-identified or not, would need this before anything else. Worth
  stating plainly rather than leaving implicit.

## Demo

The GIF at the top is a real recording of two live natural-language queries
answered by the system end-to-end, including one query with a name typo
(`ben thorp` instead of `Ben Torp`) that the fuzzy patient-matching still
resolved correctly.

Try it yourself without any setup beyond [Installation](#installation):

```bash
python3 scripts/demo.py
```

That's a simplified local REPL — real usage is through an actual MCP client
(Claude Code or otherwise), where an AI model does the natural-language tool
routing itself rather than the REPL's keyword stand-in. See
[Usage](#usage) in the technical reference for both paths, plus the direct
Python-call option with no AI involved at all.

## What I would improve

Roughly in priority order, if this moved past portfolio scope:

1. **Expand resource coverage to `MedicationRequest` and `Encounter`.**
   (High impact, moderate effort.) Medication and visit history are the
   next-most-requested data types after diagnoses and vitals in real clinical
   workflows — the natural next slice, not a random expansion.
2. **Add a review queue for unresolved free-text conditions**, instead of only
   flagging and excluding them. (High impact, higher effort.) Right now an
   unmapped condition note is surfaced as a gap; closing that loop with a
   human-in-the-loop coding step would make the system actually usable in a
   real clinical workflow, not just honest about its limits.
3. **Add authentication, authorization, and audit logging.** (Medium impact,
   required before touching anything beyond mock data.) Not a "nice to have" —
   a hard prerequisite for handling any real patient data at all.
4. **Stress-test against a larger, noisier synthetic dataset.** (Medium impact.)
   42 patients from one generator run is enough to prove the pattern, not
   enough to trust the fuzzy-matching logic at scale.

---

## Technical reference

The sections below are the full engineering detail — installation, usage,
configuration, testing, schema validation internals, and the legacy data
design. Collapsed by default to keep the page narrative-forward; expand
whichever section you need.

<details>
<summary><strong>Prerequisites</strong></summary>

- Python 3.10+
- A JDK (11+) only if you intend to regenerate the mock database from
  scratch — not needed to run the server against the checked-in
  `data/legacy.db`

</details>

<a id="installation"></a>
<details>
<summary><strong>Installation</strong></summary>

Three steps — clone, install, verify. Dependency versions are pinned
(`pyproject.toml`) to a combination that's actually tested working; an
unpinned `fhir.resources` install can pull in an incompatible
`annotated-types` release and fail with an `ImportError`, so match these
versions rather than installing latest:

```bash
# 1. Clone
git clone https://github.com/gdanse/legacy-to-fhir-mcp-server.git
cd legacy-to-fhir-mcp-server

# 2. Install (pinned versions -- see pyproject.toml)
python3 -m venv .venv && source .venv/bin/activate
pip install "mcp==1.28.1" "fhir.resources==8.3.0" "annotated-types<0.8.0"

# 3. Verify
python3 scripts/test_query_legacy_patient_records.py
```

All three steps together run in well under two minutes on a normal
connection — the mock database (`data/legacy.db`) is already checked in,
so there's no seeding step.

</details>

<a id="usage"></a>
<details>
<summary><strong>Usage</strong></summary>

Start the server (stdio transport):

```bash
python3 mcp_server/server.py
```

Claude Code isn't required — the server speaks the standard Model Context
Protocol over stdio, so anything that can act as an MCP client can use it.
A few ways to actually make queries:

**Via Claude Code (or any MCP client)**

Copy `.mcp.json.example` to `.mcp.json` and replace `<path-to-this-repo>`
with your local checkout path (`.mcp.json` is gitignored since that path is
machine-specific):

```bash
cp .mcp.json.example .mcp.json
```

Then ask natural-language questions — the model picks the right tool
automatically:

| Example query | Tool selected | Returns |
|---|---|---|
| `"What conditions does Ben Torp have?"` | `query_legacy_patient_records` | `Patient` + `Condition` resources |
| `"Adelle Raynor's weight"` | `translate_vitals_log` | `Observation` resources, filtered to weight readings |
| `"MRN720072 heart rate"` | `translate_vitals_log` | `Observation` resources, resolved by MRN and filtered to heart rate |

Both tools accept a name, a legacy MRN, or both, and return:

```json
{
  "bundle": "<FHIR searchset Bundle>",
  "warnings": ["<flags for anything left deliberately unresolved>"],
  "validation_failures": ["<resources excluded for failing schema validation>"]
}
```

Any other MCP client works the same way — e.g. the standalone
[MCP Inspector](https://github.com/modelcontextprotocol/inspector), a
browser-based dev tool for exercising an MCP server without wiring it into
an AI assistant at all:

```bash
npx @modelcontextprotocol/inspector python3 mcp_server/server.py
```

**Without any MCP client — direct Python calls**

Natural-language tool *selection* is what an MCP client provides. The data
retrieval, FHIR mapping, and validation underneath don't depend on it — the
tool functions are plain Python and can be called directly with a query
string, no AI involved:

```python
from mcp_server.server import query_legacy_patient_records, translate_vitals_log

query_legacy_patient_records("Ben Torp")
translate_vitals_log("Adelle Raynor's weight")
```

This is exactly what the smoke tests in `scripts/` do — see
[Testing](#testing) below.

</details>

<details>
<summary><strong>Configuration</strong></summary>

None needed. This project requires no API keys or secrets — all data is
local, synthetic, and self-contained (`data/legacy.db`), and no code path
calls any external service at runtime.

</details>

<a id="testing"></a>
<details>
<summary><strong>Testing</strong></summary>

Smoke tests bypass the MCP transport, call the tool functions directly, and
validate every returned resource against `fhir.resources`' FHIR R4B models:

```bash
python3 scripts/test_query_legacy_patient_records.py
python3 scripts/test_translate_vitals_log.py
python3 scripts/test_validation_layer.py
```

`scripts/verify_legacy_mess.py` prints direct SQL evidence of each legacy
mess pattern described below, straight from `data/legacy.db`.

`scripts/demo.py` is an interactive REPL for trying queries yourself
without any MCP client (this is what the demo GIF at the top was recorded
from) — run it and type a natural-language query at the `query>` prompt.

</details>

<details>
<summary><strong>Project structure</strong></summary>

```
mcp_server/          MCP server + FHIR mapping/validation logic
  server.py            tool entry points
  legacy_repo.py       resolves natural-language queries against the DB
  fhir_build.py        maps legacy rows -> FHIR resource dicts
  codes.py             LOINC / SNOMED-CT lookup tables
  datetimes.py         legacy date-format parsing
  validation.py        independent FHIR R4B schema validation gate
legacy_data/          builds the mock legacy database from Synthea output
  extract.py            reads clean FHIR bundles
  mangle.py             deliberately degrades them into legacy-shaped rows
  schema.py             SQLite schema for the four legacy tables
mapping/
  fhir-mapping-schema.md  field-by-field mapping spec + verified code tables
scripts/               smoke tests + database build/verification scripts
  demo.py                interactive REPL for trying queries (used for the demo GIF)
docs/demo.gif          the README demo recording
data/legacy.db         the mock legacy database (checked in, regeneratable)
```

</details>

<a id="legacy-data-design"></a>
<details>
<summary><strong>Legacy data design</strong></summary>

`data/legacy.db` is generated from Synthea's clean, FHIR-conformant output,
then deliberately degraded into four legacy-system-shaped SQLite tables:

| Table | Legacy mess pattern |
|---|---|
| `patient_identity` | DOB as `MM/DD/YYYY` string |
| `patient_contact` | Same patient's *other* half of the demographic record, in a separate table with no shared key — joins to `patient_identity` only by fuzzy name+DOB match. DOB stored as a Unix epoch here (a different format than `patient_identity`). |
| `vitals_log` | Date as bare `YYYYMMDD` string. Units are frequently `NULL`, and sometimes ride along inside `value_raw` instead (e.g. `"55.8 kg"`) rather than in a `unit` column. |
| `conditions_log` | Date as `DD-Mon-YYYY` string. Conditions are stored as denormalized free text (e.g. `"high blood pressure, on meds"`) instead of a coded value, and same-day conditions are sometimes merged into one comma-separated note. |

To regenerate from scratch:

```bash
# 1. Requires a JDK (11+) on PATH.
git clone --depth 1 https://github.com/synthetichealth/synthea.git
cd synthea
./run_synthea -p 40   # generates ~40 synthetic patients into output/fhir/

# 2. Copy the bundles into this project
cp -R output/fhir "<this-repo>/synthea_output/fhir"

# 3. Build the mangled legacy SQLite DB
cd "<this-repo>"
python3 scripts/build_legacy_db.py
```

</details>

<details>
<summary><strong>Validation & fallback philosophy</strong></summary>

Every resource returned by either tool passes through `mcp_server/validation.py`
— an independent FHIR R4B schema check that doesn't trust the mapping logic
in `fhir_build.py`. Anything that fails is excluded from the bundle and
reported in a sibling `validation_failures` list instead of being returned
as if it were valid.

The same never-guess principle governs every ambiguous mapping case:

| Flag | Resource | Trigger |
|---|---|---|
| `identity_join_unresolved` | Patient | Zero or multiple name+DOB matches across the split identity/contact tables |
| `loinc_code_unresolved` | Observation | `vital_label` not found in the LOINC lookup table |
| `unit_missing` | Observation | `unit` column is `NULL` (and no unit was embedded in the raw value) |
| `condition_code_unresolved` | Condition | `note_text` not found in the SNOMED-CT lookup table, or is multi-concept free text |

See `mapping/fhir-mapping-schema.md` for the full field-by-field mapping
spec, including the verified LOINC/SNOMED-CT lookup tables and the
corrections made when initial code guesses turned out to be wrong.

</details>

<details>
<summary><strong>Schema validation internals (pydantic)</strong></summary>

Every FHIR resource this server returns is a **pydantic model, not a plain
dict** — `fhir.resources` (the FHIR model library used here) is built on
[`fhir_core`](https://pypi.org/project/fhir-core/), which is itself built on
[pydantic](https://docs.pydantic.dev/) v2. Concretely:

```python
>>> from fhir.resources.R4B.patient import Patient
>>> import pydantic
>>> issubclass(Patient, pydantic.BaseModel)
True
```

`mcp_server/validation.py` instantiates the matching pydantic model
(`Patient`, `Observation`, or `Condition`) for every resource `fhir_build.py`
produces, **as an independent step that doesn't trust the mapping code that
built the resource**. Pydantic's own validation raises on missing required
fields, wrong types, or malformed structure — the exact same enforcement
mechanism a hand-written Zod schema would provide in a TypeScript service,
applied here to the FHIR R4B spec. A resource that fails is excluded from
the bundle and reported in `validation_failures`; there is no code path
that returns a resource without going through this check.

</details>

<details>
<summary><strong>Security boundaries</strong></summary>

| Boundary | How it's enforced |
|---|---|
| Read-only data access | The SQLite connection is opened in `mode=ro` (`legacy_repo.py`) — no write path exists even at the driver level, independent of any application-level checks |
| No secrets / API keys | Nothing in the codebase reads `os.environ` or any credential store; no external service is called at runtime (see Configuration above) |
| No real PHI | All patient data is synthetic, generated by Synthea — every SSN carries Synthea's `999-`-prefix synthetic-data marker, and no real patient data is ingested anywhere in the pipeline |
| Local-only execution | The MCP server communicates over stdio only; it does not open a network port or accept remote connections |
| Untrusted-input handling | Every ambiguous or unparseable legacy value is flagged and excluded rather than guessed — see Validation & fallback philosophy above |

</details>

## License

MIT — see [LICENSE](LICENSE).