🔥

Livebook SQLiteを扱うMCPサーバを立てる

に公開
Elixir
livebook
tech

MCPサーバの役割として、データを蓄積したいことがあるかと思います。
本記事は、LivebookでMCPサーバを立ててSQLiteを扱う習作です。

下記に少し味付けした格好です。
https://zenn.dev/ta_to/articles/48971ba6c60e53

// 以下、例としてNGワードを返すようなMCPサーバにしています（特に意味はありません）

Mix.install

hermes_mcpを使用します。

setup
Mix.install([
  {:hermes_mcp, "~> 0.14.1"},
  {:plug, "~> 1.17"},
  {:bandit, "~> 1.6"},
  {:ecto_sqlite3, "~> 0.17"}
])

Repo

migrationを実行する関数群を用意します。（余談ですがこのあたりはElixir Desktop味がありました）

defmodule MyApp.Repo do
  use Ecto.Repo, otp_app: :my_app, adapter: Ecto.Adapters.SQLite3

  def migrate do
    Ecto.Migrator.run(MyApp.Repo, migrations(), :up, all: true)
  end
  
  def down do
    Ecto.Migrator.run(MyApp.Repo, migrations(), :down, all: true)
  end

  def reset do
    Ecto.Migrator.run(MyApp.Repo, migrations(), :down, all: true)
    Ecto.Migrator.run(MyApp.Repo, migrations(), :up, all: true)
  end

  defp migrations do
    [
      {20251008000001, MyApp.Migrations.CreateNGWords},   
    ]
  end
end

migrationファイルとschemaは、特にphoenix等でやっているいつものEctoです。

# migrations
defmodule MyApp.Migrations.CreateNGWords do
  use Ecto.Migration

  def change do
    create table(:ng_words) do
      add(:word, :string, null: false)
      timestamps(type: :utc_datetime)
    end

    create unique_index :ng_words, [:word]
  end
end

# schema
defmodule NGWord do
  use Ecto.Schema
  
  schema "ng_words" do
    field :word, :string
    timestamps(type: :utc_datetime)
  end
end

Repoのサーバを起動します。ここでSQLiteのファイルパスは永続化領域にすること。下記ではノートブックと同じフォルダに置いています。

file_path = __ENV__.file |> URI.parse() |> Map.get(:path)
db_path = file_path <> ".db"

if Process.whereis(MyApp.Repo) do
  Supervisor.stop(MyApp.Repo)
end

# 複数起動するとエラーが表示されたため、pool_size: 1
MyApp.Repo.start_link(database: db_path, pool_size: 1)
# MyApp.Repo.down()
MyApp.Repo.migrate()

サンプルデータ投入

MyApp.Repo.insert(%NGWord{word: "完璧"})

コンテキスト実装

defmodule NGWords do
  import Ecto.Query
  
  alias MyApp.Repo

  def list_ng_words do
    Repo.all(NGWord)
  end
end

MCPサーバ機能実装

hermesの書き方に従って、MCPサーバの機能(tool)を作ります。

use Hermes.Serverしたモジュールにcomponentとしてtoolのモジュールを追加する形です。// 独立して定義できるのでとても嬉しい。

defmodule MyApp.Tools.NGWords.List do
  # 下記がtoolのdescriptionとなるため重要
  @moduledoc """
  Important: List of forbidden words on chats
  """

  use Hermes.Server.Component, type: :tool

  alias Hermes.Server.Response

  schema do
    %{}
  end

  def execute(_params, frame) do
    ng_words = NGWords.list_ng_words() |> Enum.map(& Map.take(&1, [:word]))

    response =
      Response.tool()
      |> Response.structured(%{ng_words: ng_words, total: length(ng_words)})

    {:reply, response, frame}
  end
end

↓のように↑のモジュールをcomponentとして記載する。nameがtoolのnameになるので大事です。

defmodule MyApp.MCPServer do
  use Hermes.Server,
    name: "My MGWord Server",
    version: "1.0.0",
    capabilities: [:tools]

  component(MyApp.Tools.NGWords.List, name: "ng_words_list")
  # # 以下は、create/deleteを追加するときの想定記述
  # component(MyApp.Tools.NGWords.Create, name: "ng_words_create")
  # component(MyApp.Tools.NGWords.Delete, name: "ng_words_delete")
end

確認

MyApp.MCPServer.__components__(:tool)

MCPサーバの起動

defmodule MyRouter do
  use Plug.Router

  plug Plug.Parsers,
    parsers: [:urlencoded, :json],
    pass: ["text/*"],
    json_decoder: JSON

  plug :match
  plug :dispatch
  
  forward "/mcp", to: Hermes.Server.Transport.StreamableHTTP.Plug, init_opts: [server: MyApp.MCPServer]

  match _ do
    send_resp(conn, 404, "Not found")
  end
end

children = [
  Hermes.Server.Registry,
  {MyApp.MCPServer, transport: :streamable_http},
  {Bandit, plug: MyRouter, port: 4002}
]

opts = [strategy: :one_for_one, name: MyApplication.Supervisor]
Supervisor.start_link(children, opts)

簡単なMCPサーバならLivebook１つで補えそうな感じですね。

Discussion