[MCP] Add EEGPrep agent server

.notes/implementation-notes.html

@@ -721,5 +721,28 @@ <h2>Verification Notes</h2>
     option equivalence, catalog validator ownership, shared extension metadata
     helpers, and bundled plugin menu projection from registry metadata.</li>
   </ul>
+  <h1>EEGPrep MCP Agent Server Notes</h1>
+  <h2>Design Decisions</h2>
+  <ul>
+    <li>Added <code>eegprep-mcp</code> as an optional package extra and entrypoint
+    rather than a base dependency, so normal EEGPrep installs remain lightweight
+    while agent users can opt into the MCP SDK.</li>
+    <li>Kept MCP tool behavior in <code>eegprep.mcp.tools</code> without importing
+    the MCP SDK. The FastMCP server is only an adapter, which lets tests validate
+    the agent contract directly and gives users a clear missing-extra error.</li>
+    <li>Command execution routes through the installed EEGPrep Python CLI
+    dispatcher with captured stdout/stderr. It never uses shell execution, always
+    requires <code>--json</code>, and requires explicit <code>allow_write</code>
+    and <code>allow_overwrite</code> gates for mutating commands.</li>
+    <li>MCP reuses the CLI's existing capabilities, schemas, examples, validation,
+    and bundled skills instead of creating a second agent contract.</li>
+  </ul>
+  <h2>Verification Notes</h2>
+  <ul>
+    <li>Focused tests cover sample-data inspection/validation, command planning,
+    write/overwrite safety gates, in-process CLI execution, FastMCP tool/resource
+    registration, the module help/version path, and CLI discovery of the bundled
+    MCP skill.</li>
+  </ul>
 </body>
 </html>

docs/source/user_guide/agent_cli.rst

@@ -9,6 +9,12 @@ reproducible preprocessing pipelines. It complements the human-facing
 ``eegprep-gui`` and ``eegprep-console`` entry points; it does not replace the
 shared GUI plus console workspace.
 
+EEGPrep also includes an optional ``eegprep-mcp`` server for AI tools that
+support the Model Context Protocol. The MCP server exposes the same
+machine-readable command schemas, examples, dataset inspection, validation, and
+safe command execution rules through tool calls, so agents can work with EEGPrep
+without scraping terminal help text.
+
 To get started, point your AI agent at this page or at the EEGPrep repository.
 The documentation, bundled CLI skill, and repository ``AGENTS.md`` file provide
 the context agents need to use EEGPrep commands safely and reproducibly.
@@ -70,10 +76,74 @@ Discovery Commands
    eegprep examples pipeline --json
    eegprep skills list --json
    eegprep skills get eegprep-cli
+   eegprep skills get eegprep-mcp
 
 These commands let agents discover supported operations and version-matched
 usage guidance without scraping docs.
 
+Model Context Protocol Server
+=============================
+
+Install the optional MCP dependencies when you want an AI assistant or agent
+host to call EEGPrep through MCP tools:
+
+.. code-block:: bash
+
+   pip install "eegprep[mcp]"
+
+From a source checkout, the development environment already includes the MCP
+dependency group after ``uv sync --group dev``. Start the server with stdio,
+which is the transport most local agent hosts expect:
+
+.. code-block:: bash
+
+   uv run eegprep-mcp
+
+For MCP clients that use HTTP transports:
+
+.. code-block:: bash
+
+   uv run eegprep-mcp --transport streamable-http --host 127.0.0.1 --port 8000
+
+The server exposes these tools:
+
+.. list-table::
+   :header-rows: 1
+
+   * - Tool
+     - Purpose
+   * - ``eegprep_capabilities``
+     - List MCP tools, allowed CLI commands, and write/overwrite safety policy.
+   * - ``eegprep_agent_guide``
+     - Return version-matched MCP usage rules for agents.
+   * - ``eegprep_inspect_dataset``
+     - Inspect dataset summary, events, channels, or ICA fields.
+   * - ``eegprep_validate_dataset``
+     - Validate a dataset and return stable warning/error codes.
+   * - ``eegprep_command_schema``
+     - Return a command schema before generating command arguments.
+   * - ``eegprep_command_examples``
+     - Return copy-pasteable examples for a command.
+   * - ``eegprep_plan_cli_command``
+     - Plan an allowlisted CLI command without running it.
+   * - ``eegprep_run_cli_command``
+     - Run an allowlisted JSON CLI command in-process with explicit write gates.
+
+Agents should use this sequence for data-changing work:
+
+.. code-block:: text
+
+   eegprep_inspect_dataset(path, section="summary")
+   eegprep_validate_dataset(path)
+   eegprep_command_schema("resample")
+   eegprep_plan_cli_command(["resample", path, "--freq", "128", "--output", out, "--json"])
+   eegprep_run_cli_command([...], allow_write=True)
+
+``eegprep_run_cli_command`` never invokes a shell. It routes through EEGPrep's
+Python CLI dispatcher, captures stdout and stderr separately, requires
+``--json``, and blocks write or overwrite operations unless the agent passes the
+matching explicit flag after user approval.
+
 Dataset Inspection And Validation
 =================================
 

pyproject.toml

@@ -54,6 +54,9 @@ console = [
   "eegprep[gui]",
   "ipython>=8.0",
 ]
+mcp = [
+  "mcp>=1.22",
+]
 docs = [
   "sphinx>=7.0",
   "pydata-sphinx-theme>=0.14.0",
@@ -69,20 +72,23 @@ all = [
   "eegprep[torch]",
   "eegprep[gui]",
   "eegprep[console]",
+  "eegprep[mcp]",
   "eegprep[docs]",
 ]
 
 [project.scripts]
 eegprep = "eegprep.cli.main:main"
 eegprep-gui = "eegprep.functions.adminfunc.eeglab:main"
 eegprep-console = "eegprep.functions.adminfunc.console:main"
+eegprep-mcp = "eegprep.mcp.server:main"
 eegprep-validate-extension-catalog = "eegprep.extension_catalog_validation:main"
 
 [dependency-groups]
 dev = [
   # Keep `uv run eegprep-console --full` working from a fresh source checkout.
   # Published installs still keep GUI/console dependencies behind extras.
   "ipython>=8.0",
+  "mcp>=1.22",
   "pytest>=8.0",
   "pyqtgraph>=0.13.7",
   "PySide6>=6.6",

src/eegprep/cli/discovery.py

@@ -9,6 +9,7 @@
 
 
 CLI_SKILL_NAME = "eegprep-cli"
+MCP_SKILL_NAME = "eegprep-mcp"
 
 
 def capabilities() -> dict[str, Any]:
@@ -324,7 +325,12 @@ def skills_list() -> dict[str, Any]:
                 "name": CLI_SKILL_NAME,
                 "description": "Core EEGPrep CLI usage guide for AI agents.",
                 "path": str(_skill_path(CLI_SKILL_NAME)),
-            }
+            },
+            {
+                "name": MCP_SKILL_NAME,
+                "description": "MCP server usage guide for AI agents working with EEGPrep.",
+                "path": str(_skill_path(MCP_SKILL_NAME)),
+            },
         ],
     )
 

src/eegprep/mcp/__init__.py

@@ -0,0 +1,25 @@
+"""Agent-facing Model Context Protocol integration for EEGPrep."""
+
+from __future__ import annotations
+
+from .tools import (
+    agent_guide,
+    capabilities,
+    command_examples,
+    command_schema,
+    inspect_eeg_dataset,
+    plan_cli_command,
+    run_cli_command,
+    validate_eeg_dataset,
+)
+
+__all__ = [
+    "agent_guide",
+    "capabilities",
+    "command_examples",
+    "command_schema",
+    "inspect_eeg_dataset",
+    "plan_cli_command",
+    "run_cli_command",
+    "validate_eeg_dataset",
+]

src/eegprep/mcp/server.py

@@ -0,0 +1,186 @@
+"""MCP server entrypoint for EEGPrep."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from typing import Any
+
+import eegprep
+
+from . import tools
+
+
+INSTALL_HINT = "Install the optional MCP dependencies with: pip install 'eegprep[mcp]' or uv sync --extra mcp"
+
+
+def build_server(*, host: str = "127.0.0.1", port: int = 8000, log_level: str = "INFO") -> Any:
+    """Build the EEGPrep FastMCP server."""
+    FastMCP = _load_fastmcp()
+    server = FastMCP(
+        name="EEGPrep",
+        instructions=(
+            "Use EEGPrep MCP for EEGLAB-compatible EEG dataset inspection, validation, "
+            "agent guidance, CLI schema discovery, and explicit JSON-safe CLI execution. "
+            "Prefer inspect/validate/plan before mutating data."
+        ),
+        host=host,
+        port=port,
+        log_level=log_level,
+    )
+
+    @server.tool(
+        name="eegprep_capabilities",
+        description="List EEGPrep MCP tools, allowed CLI commands, and agent safety policy.",
+    )
+    def eegprep_capabilities() -> dict[str, Any]:
+        return _call(tools.capabilities)
+
+    @server.tool(
+        name="eegprep_agent_guide",
+        description="Return bundled version-matched guidance for agents using EEGPrep MCP and CLI.",
+    )
+    def eegprep_agent_guide(full: bool = False) -> dict[str, Any]:
+        return _call(tools.agent_guide, full=full)
+
+    @server.tool(
+        name="eegprep_inspect_dataset",
+        description="Inspect an EEGLAB .set dataset. section: summary, events, channels, or ica.",
+    )
+    def eegprep_inspect_dataset(path: str, section: str = "summary", limit: int = 50) -> dict[str, Any]:
+        return _call(tools.inspect_eeg_dataset, path, section=section, limit=limit)
+
+    @server.tool(name="eegprep_validate_dataset", description="Validate an EEGLAB .set dataset.")
+    def eegprep_validate_dataset(path: str) -> dict[str, Any]:
+        return _call(tools.validate_eeg_dataset, path)
+
+    @server.tool(name="eegprep_command_schema", description="Return a machine-readable EEGPrep CLI command schema.")
+    def eegprep_command_schema(command: str) -> dict[str, Any]:
+        return _call(tools.command_schema, command)
+
+    @server.tool(name="eegprep_command_examples", description="Return EEGPrep CLI examples for a command.")
+    def eegprep_command_examples(command: str) -> dict[str, Any]:
+        return _call(tools.command_examples, command)
+
+    @server.tool(
+        name="eegprep_plan_cli_command",
+        description="Plan an allowlisted EEGPrep CLI command without executing it.",
+    )
+    def eegprep_plan_cli_command(arguments: list[str]) -> dict[str, Any]:
+        return _call(tools.plan_cli_command, arguments)
+
+    @server.tool(
+        name="eegprep_run_cli_command",
+        description=(
+            "Execute an allowlisted EEGPrep CLI command through Python, not a shell. "
+            "Arguments must include --json. File-writing commands require allow_write=True; "
+            "--overwrite also requires allow_overwrite=True."
+        ),
+    )
+    def eegprep_run_cli_command(
+        arguments: list[str],
+        allow_write: bool = False,
+        allow_overwrite: bool = False,
+    ) -> dict[str, Any]:
+        return _call(
+            tools.run_cli_command,
+            arguments,
+            allow_write=allow_write,
+            allow_overwrite=allow_overwrite,
+        )
+
+    @server.resource(
+        "eegprep://capabilities",
+        name="EEGPrep MCP capabilities",
+        description="Machine-readable EEGPrep MCP and CLI capabilities.",
+        mime_type="application/json",
+    )
+    def capabilities_resource() -> str:
+        return json.dumps(tools.capabilities(), sort_keys=True)
+
+    @server.resource(
+        "eegprep://agent-guide",
+        name="EEGPrep MCP agent guide",
+        description="Version-matched EEGPrep MCP guidance for AI agents.",
+        mime_type="text/markdown",
+    )
+    def agent_guide_resource() -> str:
+        return tools.agent_guide(full=True)["content"]
+
+    @server.prompt(
+        name="eegprep_preprocess_plan",
+        description="Prompt template for planning a safe EEGPrep preprocessing workflow.",
+    )
+    def eegprep_preprocess_plan(dataset_path: str) -> str:
+        return (
+            "Use EEGPrep MCP to inspect and validate this dataset before proposing changes: "
+            f"{dataset_path}\n"
+            "Call eegprep_inspect_dataset(section='summary'), eegprep_validate_dataset, then "
+            "eegprep_command_schema for each needed operation. Prefer non-destructive outputs, "
+            "pipeline plan/dry-run before expensive work, and report stable error codes."
+        )
+
+    return server
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Run the EEGPrep MCP server."""
+    parser = argparse.ArgumentParser(
+        prog="eegprep-mcp",
+        description="Run the EEGPrep Model Context Protocol server for AI agents.",
+        epilog=f"Agent start: configure your MCP client to run `eegprep-mcp`. {INSTALL_HINT}.",
+    )
+    parser.add_argument("--transport", choices=["stdio", "streamable-http", "sse"], default="stdio")
+    parser.add_argument("--host", default="127.0.0.1", help="Host for HTTP transports.")
+    parser.add_argument("--port", type=int, default=8000, help="Port for HTTP transports.")
+    parser.add_argument(
+        "--log-level",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+        default="INFO",
+    )
+    parser.add_argument("--version", action="store_true", help="Show EEGPrep version and exit.")
+    args = parser.parse_args(argv)
+    if args.version:
+        print(f"eegprep-mcp {eegprep.__version__}")
+        return 0
+    try:
+        server = build_server(host=args.host, port=args.port, log_level=args.log_level)
+    except RuntimeError as exc:
+        print(str(exc), file=sys.stderr)
+        return 1
+    server.run(transport=args.transport)
+    return 0
+
+
+def _load_fastmcp() -> Any:
+    try:
+        from mcp.server.fastmcp import FastMCP
+    except ImportError as exc:
+        raise RuntimeError(f"The EEGPrep MCP server requires the optional 'mcp' extra. {INSTALL_HINT}.") from exc
+    return FastMCP
+
+
+def _call(func: Any, *args: Any, **kwargs: Any) -> dict[str, Any]:
+    try:
+        return func(*args, **kwargs)
+    except Exception as exc:
+        code = getattr(exc, "code", "UNEXPECTED_ERROR")
+        message = getattr(exc, "message", str(exc))
+        payload: dict[str, Any] = {
+            "status": "error",
+            "schema_version": "eegprep.error.v1",
+            "code": code,
+            "message": message,
+        }
+        if getattr(exc, "path", None) is not None:
+            payload["path"] = str(getattr(exc, "path"))
+        if getattr(exc, "suggestion", None) is not None:
+            payload["suggestion"] = getattr(exc, "suggestion")
+        if getattr(exc, "details", None) is not None:
+            payload["details"] = getattr(exc, "details")
+        return payload
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))