⚡🧠 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

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.

Related MCP server: SupaUI MCP Server

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.

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

Register with an MCP client using an absolute path to the env's interpreter:

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

Or with the Claude Code CLI:

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:

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.

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