⚡🧠 eeg-mcp

Real-time EEG for AI agents — stream, replay, visualize, record, stimulate

Documentation · Tutorial · Hardware · Safety · Tool Reference

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

mermaid

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

📡 Stream

66 BrainFlow board identifiers — OpenBCI, Muse, ANT Neuro, g.tec, Mentalab and more — plus a synthetic board that needs no hardware. Samples land in a ring buffer filled by a background thread, so tool calls read a live view instead of blocking on a device.

⏪ Replay

Play an EDF/BDF/GDF/SET/FIF or BrainFlow CSV at the rate it was recorded, re-emitting annotations as events at their original timings. Adds speed, seek, pause and looping. A pipeline developed against a file runs unchanged against hardware.

👁 Visualize

A loopback-bound, token-gated live browser view: rolling traces, event markers, band power, per-electrode quality, and transport controls. Plus self-contained HTML reports — no CDN, no external assets, opens on an air-gapped machine.

💾 Record

Write continuously to MNE-native .fif with the event log attached as annotations, plus a metadata row in a store schema-compatible with neuro-mcp. Crash-safe: an interrupted session is recoverable.

🧩 Extend

Plug in your own real-time processor — feature extractor, classifier, artifact gate, or EEG tokenizer for sequence models — and it runs on the same footing as the built-ins, inside the acquisition loop. → Extending

⚡ Stimulate

One send_stim_event contract over pluggable backends: LSL for software, BrainFlow's marker channel for sample-aligned embedding, serial/TTL and templated ASCII for hardware including TMS and tES — behind three safety gates.

[!NOTE] One event log. Board markers, replayed annotations, dispatched stimulations and manual notes all land in the same table on the same clock, with absolute sample indices. A closed-loop run reconstructs afterwards with no clock join.

Install

bash

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

Optional extras and MCP client registration

Terminal

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

config.json

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

Or with the Claude Code CLI:

Terminal

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

→ Full guide: Installation

Quick start

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

python

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:

python

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.")

mermaid

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

Guide
🚀 Installation	Environment, client registration, troubleshooting
📘 Tutorial	End to end, no hardware needed
⏪ Replay-Driven Development	Build against a recording, deploy live
🔁 Closed-Loop Neurofeedback	Feature → trigger, with a measured latency budget
🩺 Live Clinical Review	Visual review, annotation, reporting
⚡ Stimulation Protocols	TMS and tES through the safety gates
🧩 Extending	Write a custom processor or EEG tokenizer
🔌 Supported Hardware	All 66 boards, formats, stimulation transports
⚠️ Safety	Read before connecting a stimulator
🛠 Tool Reference	All 47 tools

The one design decision worth knowing

[!IMPORTANT] Filtering happens in the producer thread, not at query time.

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.

mermaid

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 &amp; 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.

Consequence
Filters are causal	No zero-phase option — that needs future samples. `stream_status` reports `group_delay_sec`
Both buffers are kept	`read_window(filtered=false)` always gets raw signal, to check whether a feature is real or an artifact
Indices are shared	An event's `sample_index` means the same thing in either buffer

[!TIP] Budget a closed loop as group delay + poll interval + dispatch latency — measured at ~71 ms in the reference configuration. Good for neurofeedback; not adequate for phase-locked stimulation.

Stimulation safety

[!CAUTION] This software is not a medical device and has not been validated for clinical use. TMS and tES can cause harm, including seizure. Use only under a protocol approved by your ethics board, on a rig whose device-level interlocks are intact, with a trained operator present.

Three gates apply to every hardware backend:

#	Gate	Effect
1	Config	Hardware backends refuse to open unless the server was started with `EEG_MCP_ALLOW_HARDWARE_STIM=1`. An agent cannot set this.
2	Arming	`arm_stim` permits dispatch for a window that expires, so a stalled agent cannot resume and fire later
3	Limits	Intensity, duration and interval are clamped; violations raise rather than silently saturate

None of this replaces the interlocks on the device itself.

[!WARNING] The hardware backends are generic transports driven by command templates you supply from your device's manual — not vendor drivers, and none has been tested against a physical stimulator. A plausible-looking untested driver would be worse than none: it would fail silently while connected to something pointed at a person's head.

Start every protocol on backend="log", which accepts everything and emits nothing.

Verify

bash

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.

⬆ back to top · Part of the AImplifier neuro toolchain · sibling project neuro-mcp