A Model Context Protocol server that gives an AI agent one interface over the live EEG workflow: acquisition from ~66 BrainFlow boards, wall-clock replay of existing recordings, stateful online DSP, a live browser monitor, crash-safe recording, and gated stimulation output.

The offline counterpart is neuro-mcp (MNE processing, source imaging, BIDS/EHR storage). This is the real-time half — everything that has to happen while the signal is still arriving.

Concept

flowchart LR
    Researcher(["🔬 BCI Researcher"])
    Clinician(["🩺 Clinician"])
    Agent[["🤖 AI Agent"]]
    Server(("eeg-mcp<br/>FastMCP · 47 tools"))

    Clinician -- talks to --> Agent
    Researcher -- talks to --> Agent
    Agent -- MCP --> Server

    Server --> Acquire["Acquire<br/>66 boards · replay<br/>at true rate"]
    Server --> Process["Process<br/>stateful online DSP<br/>custom plugins"]
    Server --> Watch["Watch &amp; Record<br/>live monitor · HTML<br/>crash-safe .fif"]
    Server --> Stim["Stimulate<br/>LSL · TTL · TMS/tES<br/>3 safety gates"]

    classDef acq fill:#14b8a6,stroke:#0d9488,color:#fff
    classDef proc fill:#4f8cff,stroke:#2f5fbf,color:#fff
    classDef viz fill:#b06fe0,stroke:#7c3fae,color:#fff
    classDef stim fill:#eb5757,stroke:#b93b3b,color:#fff
    class Acquire acq
    class Process proc
    class Watch viz
    class Stim stim

Nobody calls a tool by hand — you talk to an agent in plain English and it drives the 47 tools underneath. The tutorial shows what that looks like end to end, with no hardware required.

What it does

Install

conda create -n eeg-mcp python=3.11 -y && conda activate eeg-mcp
pip install eeg-mcp

pip install "eeg-mcp[lsl]"      # LSL marker outlets (PsychoPy, OpenViBE, ...)
pip install "eeg-mcp[serial]"   # serial/TTL trigger delivery to hardware

{
  "mcpServers": {
    "eeg-realtime": {
      "command": "/path/to/envs/eeg-mcp/bin/python",
      "args": ["-m", "eeg_mcp"]
    }
  }
}

claude mcp add eeg-realtime -- /path/to/envs/eeg-mcp/bin/python -m eeg_mcp

Quick start

Ask your agent for the outcome; it picks the calls. No hardware required:

start_stream(session_id="s1", board="synthetic")
check_signal_quality(session_id="s1")          # before trusting anything
set_filters(session_id="s1", bandpass_low=1, bandpass_high=40, notch_freq=50)
get_band_power(session_id="s1", seconds=2)
start_monitor(session_id="s1")                 # → open the returned URL

Replay a real recording as if it were live, then keep the record:

inspect_recording(path="sub-04_rest.edf")
start_replay(session_id="r1", path="sub-04_rest.edf", speed=1.0)
get_events(session_id="r1", origin="annotation")
export_report(session_id="r1", notes="Routine review.")

flowchart LR
    A["start_stream<br/><i>or</i> start_replay"] --> B[check_signal_quality]
    B --> C[set_filters]
    C --> D["get_band_power<br/>get_psd"]
    C --> E[start_monitor]
    C --> P[attach_processor]
    A --> R[start_recording]
    D --> S[send_stim_event]
    P --> S
    R --> X[stop_recording]
    S --> X
    E --> X
    X --> Z[stop_stream]

    classDef hot fill:#14b8a6,stroke:#0d9488,color:#fff
    class A,X hot

Documentation

The one design decision worth knowing

A stateful IIR filter must see every sample exactly once, in order. The common shortcut — filtering each query window independently — restarts the filter at every window boundary and injects a transient each time. It is invisible in a band-power plot and fatal for anything phase-sensitive.

So the producer filters each chunk once as it arrives, carrying sosfilt delay-line state forward, and writes to a second ring buffer. Queries just read.

flowchart LR
    BF[Board / Recording] -->|poll| PR{{Producer thread}}
    PR -->|raw chunk| RB[(Raw ring buffer)]
    PR -->|stateful sosfilt| FB[(Filtered ring buffer)]
    PR -->|markers & annotations| EL[(Event log)]
    PR -->|append| DISK[(.fif on disk)]
    PR -->|streaming| PL[Your processors]
    RB & FB & EL --> Q[MCP tools]

    classDef hot fill:#14b8a6,stroke:#0d9488,color:#fff
    class PR hot

The test suite asserts chunked filtering matches whole-signal filtering to 1e-9, and asserts as a control that the naive approach does not.

Stimulation safety

Three gates apply to every hardware backend:

None of this replaces the interlocks on the device itself.

Verify

python testing/verify.py                    # core correctness
python testing/verify_recording.py          # recording + metadata store
python testing/verify_processing.py         # plugin processors
python testing/persona_bci_researcher.py    # engineer workflow
python testing/persona_clinician.py         # clinician workflow

All five drive the real server through FastMCP's in-memory client and assert against planted ground truth:

• a spike planted at an annotation onset lands at t = 0 ± 0 ms in the extracted epoch — proving annotation timing, replay clock, ring-buffer indexing and epoch extraction all agree;
• a deliberately broken plugin cannot stop acquisition — throughput holds at 1.0;
• hardware stimulation is refused while the config gate is unset.

→ What is and is not covered — including an honest list of what has never been tested against real hardware.

License

BSD-3-Clause. See LICENSE and NOTICE.