📌Overview

Purpose: This library aims to provide a straightforward implementation of the Model Context Protocol (MCP) using Server-Sent Events (SSE), facilitating real-time communications and updates.

Overview: The MCP over SSE library enables seamless integration of the Model Context Protocol within Elixir applications, allowing developers to manage connections, handle JSON-RPC messages, and maintain session integrity through configurable server settings.

Key Features:

Full MCP Server Implementation: Provides a complete setup for enabling the Model Context Protocol, allowing robust interaction through standardized methods.
SSE Connection Management: Handles the management of Server-Sent Event connections, ensuring reliable real-time communication and connection persistence.
JSON-RPC Message Handling: Automatically processes JSON-RPC requests, simplifying communication with a structured request-response mechanism.
Session Management: Ensures each connection includes a valid session ID, facilitating easy tracking and maintaining connection states.
Automatic Ping/Keepalive: Implements periodic keepalive pings to prevent disconnections, optimizing session reliability.
Error Handling and Validation: Offers built-in mechanisms to manage protocol errors and validate requests efficiently.

MCP over SSE

This library provides a simple implementation of the Model Context Protocol (MCP) over Server-Sent Events (SSE).

For more information about the Model Context Protocol, visit the Model Context Protocol Documentation.

Installation

For Phoenix Applications

Configure MIME types for SSE in config/config.exs:

config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

config :mcp_sse, :mcp_server, MCP.DefaultServer

Add the dependency in mix.exs:

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

Update your router (lib/your_app_web/router.ex):

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

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

  pipe_through :api
  post "/message", SSE.ConnectionPlug, :call
end

For Plug Applications with Bandit

Create a new Plug application with supervision:
```
mix new your_app --sup
```

Configure MIME types for SSE in config/config.exs:

config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

config :mcp_sse, :mcp_server, YourApp.MCPServer

Add dependencies to mix.exs:

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

Update your router (lib/your_app/router.ex):

defmodule YourApp.Router do
  use Plug.Router

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

  plug :match
  plug :ensure_session_id
  plug :dispatch

  def ensure_session_id(conn, _opts) do
    case get_session_id(conn) do
      nil ->
        session_id = generate_session_id()
        %{conn | query_params: Map.put(conn.query_params, "sessionId", session_id)}
      _session_id ->
        conn
    end
  end

  defp get_session_id(conn) do
    conn.query_params["sessionId"]
  end

  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

Set up your application supervision (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

Session Management

The MCP SSE server requires a session ID for each connection. The router:

Uses an existing session ID from query parameters if provided
Generates a new session ID if none exists
Ensures requests to /sse and /message endpoints have a valid session ID

Configuration Options

To configure the Bandit server with additional options:

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"
  }
]

The use MCPServer macro provides built-in message routing, protocol version validation, and logging. You need to implement handle_ping/1 and handle_initialize/2 callbacks.

Features

Full MCP server implementation
SSE connection management
JSON-RPC message handling
Tool registration and execution
Session management
Automatic ping/keepalive
Error handling and validation

Contributing

(Contributing guidelines go here)

Quick Demo

To see the MCP server in action:

Start the Phoenix server:
```
mix phx.server
```
In another terminal, run the demo client script:
```
elixir examples/mcp_client.exs
```

SSE Keepalive

The SSE connection sends periodic keepalive pings to prevent timeouts. You can configure the ping interval:

config :mcp_sse, :sse_keepalive_timeout, 30_000  # 30 seconds

Or disable pings:

config :mcp_sse, :sse_keepalive_timeout, :infinity

MCP Response Formatting

When implementing tool responses, the response must follow the MCP specification for content types. Here's how to format text responses:

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

For structured data, convert it to a formatted string:

def handle_call_tool(request_id, %{"name" => "list_companies"} = _params) do
  companies = fetch_companies()

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

For more details on response formatting, see the MCP Content Types Specification.