MCP over SSE

SSE 経由の MCP

このライブラリは、Server-Sent Events（SSE）経由の Model Context Protocol（MCP）のシンプルな実装を提供します。

Model Context Protocol の詳細については、以下をご覧ください： Model Context Protocol のドキュメント。

目次

• 機能
• 独自の MCP サーバーを作成する
• インストール
  • Phoenix アプリケーションの場合
  • Bandit を使用する Plug アプリケーションの場合
• 使い方
  • MCP Inspector を使う
  • Cursor を使う
• 設定
  • ポートと HTTPS
  • パス
  • キープアライブ
• クイックデモ
• その他の注意事項
  • クライアントの使用例
  • セッション管理
  • MCP レスポンスの形式
• コントリビューション

機能

• MCP サーバーの完全な実装
• SSE 接続管理
• JSON-RPC メッセージ処理
• ツールの登録と実行
• セッション管理
• 自動 ping／キープアライブ
• エラー処理と検証

独自の MCP サーバーを作成する

MCPServer ビヘイビアを実装する必要があります。

必須のコールバック（handle_ping/1 と handle_initialize/2）と、サポートしたい機能に応じた任意のコールバックを実装するだけで構いません。

use MCPServer マクロは、以下を提供します：

• 組み込みのメッセージルーティング
• プロトコルバージョンの検証
• 任意のコールバックのデフォルト実装
• JSON-RPC エラー処理
• ロギング

MCPServer ビヘイビアのデフォルト実装については、DefaultServer をご覧ください。

インストール

Phoenix アプリケーションの場合：

1. config/config.exs に必要な設定を追加します：

# Configure MIME types for SSE
config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

# Configure the MCP Server
config :mcp_sse, :mcp_server, YourApp.YourMCPServer

2. mix.exs の依存関係に追加します：

def deps do
  [
    {:mcp_sse, "~> 0.1.6"}
  ]
end

3. ルーターを設定します（lib/your_app_web/router.ex）：

pipeline :sse do
  plug :accepts, ["sse"]
end

scope "/" do
  pipe_through :sse
  get "/sse", SSE.ConnectionPlug, :call
  post "/message", SSE.ConnectionPlug, :call
end

4. アプリケーションを実行します：

mix phx.server

Bandit を使用する Plug アプリケーションの場合：

1. スーパービジョン付きの新しい Plug アプリケーションを作成します：

mix new your_app --sup

2. config/config.exs に必要な設定を追加します：

import Config

# Configure MIME types for SSE
config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

# Configure the MCP Server
config :mcp_sse, :mcp_server, YourApp.YourMCPServer

3. mix.exs に依存関係を追加します：

def deps do
  [
    {:mcp_sse, "~> 0.1.6"},
    {:plug, "~> 1.14"},
    {:bandit, "~> 1.2"}
  ]
end

4. ルーターを設定します（lib/your_app/router.ex）：

defmodule YourApp.Router do
  use Plug.Router

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

  plug :match
  plug :ensure_session_id
  plug :dispatch

  # Middleware to ensure session ID exists
  def ensure_session_id(conn, _opts) do
    case get_session_id(conn) do
      nil ->
        # Generate a new session ID if none exists
        session_id = generate_session_id()
        %{conn | query_params: Map.put(conn.query_params, "sessionId", session_id)}
      _session_id ->
        conn
    end
  end

  # Helper to get session ID from query params
  defp get_session_id(conn) do
    conn.query_params["sessionId"]
  end

  # Generate a unique session ID
  defp generate_session_id do
    Base.encode16(:crypto.strong_rand_bytes(8), case: :lower)
  end

  forward "/sse", to: SSE.ConnectionPlug
  forward "/message", to: SSE.ConnectionPlug

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

5. アプリケーションのスーパービジョンを設定します（lib/your_app/application.ex）：

defmodule YourApp.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {Bandit, plug: YourApp.Router, port: 4000}
    ]

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

6. アプリケーションを実行します：

mix run --no-halt

使い方

MCP Inspector を使う

• Inspector を起動します：

MCP_SERVER_URL=localhost:4000 npx @modelcontextprotocol/inspector@latest

• http://localhost:6274/ に移動します
• サーバーが実行中であることを確認します
• Connect をクリックします
• ツールの一覧表示と呼び出しができるようになります

Cursor を使う

• Cursor の設定を開きます
• MCP タブに移動します
• Add new global MCP server をクリックします
• ~/.cursor/mcp.json に以下を入力します：

{
  "mcpServers": {
    "your-mcp-server": {
      "url": "http://localhost:4000/sse"
    }
  }
}

• サーバーが実行中であることを確認します
• Cursor にツールの実行を指示します

設定

ポートと HTTPS

Bandit サーバーは、アプリケーションモジュールで追加のオプションを設定できます：

# Example with custom port and HTTPS
children = [
  {Bandit,
    plug: YourApp.Router,
    port: System.get_env("PORT", "4000") |> String.to_integer(),
    scheme: :https,
    certfile: "priv/cert/selfsigned.pem",
    keyfile: "priv/cert/selfsigned_key.pem"
  }
]

パス

SSE エンドポイントとメッセージエンドポイントで使用するパスをカスタマイズできます。

config :mcp_sse,
  sse_path: "/mcp/sse",    # Default: "/sse"
  message_path: "/mcp/msg" # Default: "/message"

これにより、ルーターでカスタムパスを使用できます。

# Phoenix
scope "/mcp" do
  pipe_through :sse
  get "/sse", SSE.ConnectionPlug, :call
  post "/msg", SSE.ConnectionPlug, :call
end

# Plug
forward "/mcp/sse", to: SSE.ConnectionPlug
forward "/mcp/msg", to: SSE.ConnectionPlug

キープアライブ

SSE 接続は、接続タイムアウトを防ぐために定期的にキープアライブ ping を送信します。 config/config.exs で ping 間隔を設定したり、完全に無効にしたりできます。

# Set custom ping interval (in milliseconds)
config :mcp_sse, :sse_keepalive_timeout, 30_000  # 30 seconds

# Or disable pings entirely
config :mcp_sse, :sse_keepalive_timeout, :infinity

クイックデモ

MCP サーバーの動作を確認するには、次の手順を実行します。

1. 1 つ目のターミナルでサーバーを起動します。

# Our example server
elixir dev/example_server.exs

# Your Phoenix application
mix phx.server

# Your Plug application
mix run --no-halt

2. 別のターミナルでデモクライアントスクリプトを実行します。

elixir dev/example_client.exs

クライアントスクリプトは次の処理を行います。

• SSE エンドポイントに接続します
• 接続を初期化します
• 使用可能なツールを一覧表示します
• 入力例を使って upcase ツールを呼び出します
• 各ステップの結果を表示します

これにより、Model Context Protocol のフローとサーバーの機能を実際に確認できます。

その他の注意事項

クライアントの使用例

// Connect to SSE endpoint
const sse = new EventSource('/sse');

// Handle endpoint message
sse.addEventListener('endpoint', (e) => {
  const messageEndpoint = e.data;
  // Use messageEndpoint for subsequent JSON-RPC requests
});

// Send initialize request
fetch('/message?sessionId=YOUR_SESSION_ID', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    jsonrpc: '2.0',
    id: 1,
    method: 'initialize',
    params: {
      protocolVersion: '2024-11-05',
      capabilities: {}
    }
  })
});

セッション管理

MCP SSE サーバーでは、接続ごとにセッション ID が必要です。ルーターは自動的に次の処理を行います。

• クエリパラメーターに既存のセッション ID が指定されている場合は、それを使用します
• セッション ID がない場合は、新しいセッション ID を生成します
• /sse および /message エンドポイントへのすべてのリクエストに有効なセッション ID が含まれるようにします

MCP のレスポンス形式

MCP サーバーでツールのレスポンスを実装する場合、コンテンツは MCP 仕様に準拠したコンテンツタイプにする必要があります。 レスポンスのコンテンツは、次のいずれかのタイプでフォーマットしてください。

# Text content
{:ok,
 %{
   jsonrpc: "2.0",
   id: request_id,
   result: %{
     content: [
       %{
         type: "text",
         text: "Your text response here"
       }
     ]
   }
 }}

# Image content
{:ok,
 %{
   jsonrpc: "2.0",
   id: request_id,
   result: %{
     content: [
       %{
         type: "image",
         data: "base64_encoded_image_data",
         mimeType: "image/png"
       }
     ]
   }
 }}

# Resource reference
{:ok,
 %{
   jsonrpc: "2.0",
   id: request_id,
   result: %{
     content: [
       %{
         type: "resource",
         resource: %{
           name: "resource_name",
           description: "resource description"
         }
       }
     ]
   }
 }}

JSON などの構造化データの場合は、フォーマット済みの文字列に変換してください。

def handle_call_tool(request_id, %{"name" => "list_companies"} = _params) do
  companies = fetch_companies()  # Your data fetching logic

  {:ok,
   %{
     jsonrpc: "2.0",
     id: request_id,
     result: %{
       content: [
         %{
           type: "text",
           text: JSON.encode!(companies, pretty: true)
         }
       ]
     }
   }}
end

レスポンス形式の詳細については、MCP コンテンツタイプ仕様を参照してください。

コントリビューション

• リポジトリをフォークしてクローンします
• フォーク内に新しいブランチを作成します
• 変更を加えてコミットします
• 変更をフォークにプッシュします
• アップストリームにプルリクエストを作成します