eeg-mcp

AI エージェントに、ライブ EEG ワークフロー全体を操作する単一のインターフェースを提供する Model Context Protocol サーバーです。約 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

ツールを手動で呼び出す必要はありません。平易な英語でエージェントに話しかけると、エージェントが内部の 47 個のツールを操作します。 **チュートリアル**では、ハードウェアを必要とせず、最初から最後までの流れを確認できます。

できること

メモ

イベントログは 1 つです。 ボードのマーカー、再生された注釈、送信された刺激、手動のメモはすべて、同じクロックを使用する同じテーブルに、絶対サンプルインデックスとともに記録されます。クローズドループ実行は、クロックを結合せずに後から再構築できます。

インストール

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

クイックスタート

エージェントに実現したい結果を伝えると、必要な呼び出しを選択します。ハードウェアは不要です。

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の遅延線状態を引き継いで、2つ目のリングバッファに書き込みます。クエリは読み取りのみを行います。

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は、発作を含む危害を引き起こす可能性があります。倫理審査委員会が承認したプロトコルに従い、デバイスレベルのインターロックが正常に機能している装置を使用し、訓練を受けたオペレーターの立ち会いのもとでのみ使用してください。

すべてのハードウェアバックエンドには、次の3つのゲートが適用されます。

これらは、デバイス自体のインターロックの代わりにはなりません。

警告

**ハードウェアバックエンドは、デバイスのマニュアルをもとにユーザーが用意するコマンドテンプレートで動作する汎用トランスポートであり、ベンダードライバーではありません。また、実際の刺激装置でテストされたものもありません。**一見もっともらしく見える未テストのドライバーは、ないよりも危険です。人の頭部に向けられた装置に接続していても、何も通知せずに失敗する可能性があります。

すべてのプロトコルは、すべてのコマンドを受け付けて何も出力しない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 をご覧ください。