eeg-mcp

一个 Model Context Protocol 服务，为 AI Agent 提供统一接口，覆盖 EEG 的实时工作流：从约 66 种 BrainFlow 设备采集数据、按墙上时钟回放现有录制内容、有状态的在线 DSP、实时浏览器监视器、崩溃安全录制，以及受门控保护的刺激输出。

离线配套项目是 neuro-mcp（MNE 处理、源成像、BIDS/EHR 存储）。本项目负责实时部分——即必须在信号仍持续到达时完成的一切工作。

概念

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

无需手动调用工具——只需用日常语言与 Agent 对话，它便会驱动底层的 47 种工具。 教程 将端到端展示具体操作过程，无需任何硬件。

功能

说明

一份事件日志。 板卡标记、回放注释、已发送的刺激和手动备注都使用同一时钟，写入同一张表，并带有绝对样本索引。闭环运行结束后即可重建，无需关联时钟。

安装

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

快速开始

告诉 Agent 你想要的结果，由它选择调用方式。无需硬件：

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

像实时数据一样回放真实录制内容，然后继续录制：

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

值得了解的一项设计决策

重要

滤波在生产者线程中进行，而不是在查询时进行。

有状态 IIR 滤波器必须按顺序恰好处理每个样本一次。常见的捷径——对每个查询窗口独立滤波——会在每个窗口边界重新启动滤波器，并在每次启动时引入瞬态。在频带功率图中看不出来，但对任何相位敏感的用途都致命。

因此，生产者会在每个数据块到达时对其滤波一次，沿用 sosfilt 的延迟线状态，并写入第二个环形缓冲区。查询只需读取数据。

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

测试套件断言，分块滤波与整段信号滤波的结果相差不超过 1e-9，并且断言作为对照的朴素方法确实无法满足这一点。

提示

闭环的预算应为群延迟 + 轮询间隔 + 分发延迟—— 在参考配置中测得约 71 ms。适用于神经反馈； 不适用于相位锁定刺激。

刺激安全

注意

本软件不是医疗器械，且未经临床用途验证。 TMS 和 tES 可能造成伤害，包括癫痫发作。仅可在经伦理委员会批准的方案下使用；设备的硬件级联锁必须完好，并且必须有经过培训的操作员在场。

每个硬件后端都适用以下三道安全门控：

这些措施都不能替代设备本身的联锁。

警告

硬件后端是通用传输层，通过你根据设备手册提供的命令模板驱动——它们不是厂商驱动程序，也没有任何一个经过实体刺激器测试。 一个看似可信但未经测试的驱动程序比没有驱动程序更糟：连接到某个对准他人头部的设备时，它可能悄无声息地失效。

每个协议都应从 backend="log" 开始；它会接受所有内容，但不会发出任何内容。

验证

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

这 5 项测试都通过 FastMCP 的内存客户端驱动真实服务，并根据预设的真实值进行断言：

• 在标注起始时刻预设的尖峰，在提取的历元中落在 t = 0 ± 0 ms——证明标注时间、回放时钟、环形缓冲区索引和历元提取彼此一致；
• 故意损坏的插件无法中断采集——吞吐量保持在 1.0；
• 未设置配置门控时，拒绝硬件刺激。

→ 已覆盖和未覆盖的内容 — 其中包括一份坦诚的清单，列出哪些内容从未在真实硬件上测试过。

许可证

BSD-3-Clause。请参阅 LICENSE 和 NOTICE。