AI导哈

Elixir MCP服务器

一个轻量级的Elixir MCP（模型上下文协议）服务器，用于管理会话、工具、提示和资源。

  • by PJUllrich
  • Elixir
  • 本地部署
  • 35 星标
  • 更新于
在 GitHub 查看 PJUllrich/emcp

使用说明

电磁兼容性计划

一个最小的Elixir MCP（模型上下文协议）服务器。

设置

1.初始化会话存储

将会话存储初始化添加到应用程序的 start/2:

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    EMCP.SessionStore.ETS.init()

    children = [
      # ...
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

您可以通过实现以下功能来使用自定义会话存储 EMCP.SessionStore 行为。

2.定义服务器

defmodule MyApp.MCPServer do
  use EMCP.Server,
    name: "my-app",
    version: "1.0.0",
    tools: [MyApp.Tools.Echo],
    prompts: [MyApp.Prompts.CodeReview],
    resources: [MyApp.Resources.Readme],
    resource_templates: [MyApp.ResourceTemplates.UserProfile],
    # You can optionally add these meta fields as well:
    title: "My App",
    description: "MCP server for My App",
    instructions: "Use the tools to interact with My App.",
end

3.安装运输工具

将StreamableHTTP传输添加到您的Phoenix路由器。由于EMCP本身处理内容协商，因此将其安装在任何管道之外：

scope "/mcp" do
  forward "/", EMCP.Transport.StreamableHTTP, server: MyApp.MCPServer
end

4.向服务器添加工具

defmodule MyApp.Tools.Echo do
  @behaviour EMCP.Tool

  @impl EMCP.Tool
  def name, do: "echo"

  @impl EMCP.Tool
  def description, do: "Echoes back the provided message"

  @impl EMCP.Tool
  def input_schema do
    %{
      type: :object,
      properties: %{
        message: %{type: :string}
      },
      required: [:message]
    }
  end

  @impl EMCP.Tool
  def call(_conn, %{"message" => message}) do
    EMCP.Tool.response([%{"type" => "text", "text" => message}])
  end

  # Return errors with:
  # EMCP.Tool.error("something went wrong")
end

提示

提示是返回结构化消息的可重用模板：

defmodule MyApp.Prompts.CodeReview do
  @behaviour EMCP.Prompt

  @impl EMCP.Prompt
  def name, do: "code_review"

  @impl EMCP.Prompt
  def description, do: "Reviews code with optional focus area"

  @impl EMCP.Prompt
  def arguments do
    [
      %{name: "code", description: "The code to review", required: true},
      %{name: "focus", description: "Optional area to focus on"}
    ]
  end

  @impl EMCP.Prompt
  def template(_conn, %{"code" => code} = args) do
    focus = args["focus"]

    user_text =
      if focus,
        do: "Review this code, focusing on #{focus}:\n\n#{code}",
        else: "Review this code:\n\n#{code}"

    %{
      "description" => "Code review prompt",
      "messages" => [
        %{"role" => "user", "content" => %{"type" => "text", "text" => user_text}},
        %{"role" => "assistant", "content" => %{"type" => "text", "text" => "I'll review the code you've provided."}}
      ]
    }
  end
end

资源

资源公开了客户端可以读取的数据。静态资源具有固定的URI：

defmodule MyApp.Resources.Readme do
  @behaviour EMCP.Resource

  @impl EMCP.Resource
  def uri, do: "file:///project/readme"

  @impl EMCP.Resource
  def name, do: "readme"

  @impl EMCP.Resource
  def description, do: "The project README"

  @impl EMCP.Resource
  def mime_type, do: "text/plain"

  @impl EMCP.Resource
  def read(_conn), do: File.read!("README.md")
end

资源模板使用URI模式，因此客户端可以请求动态内容：

defmodule MyApp.ResourceTemplates.UserProfile do
  @behaviour EMCP.ResourceTemplate

  @impl EMCP.ResourceTemplate
  def uri_template, do: "db:///users/{user_id}/profile"

  @impl EMCP.ResourceTemplate
  def name, do: "user_profile"

  @impl EMCP.ResourceTemplate
  def description, do: "A user profile by ID"

  @impl EMCP.ResourceTemplate
  def mime_type, do: "application/json"

  @impl EMCP.ResourceTemplate
  def read(_conn, "db:///users/" <> rest) do
    case String.split(rest, "/") do
      [user_id, "profile"] ->
        user = MyApp.Repo.get!(MyApp.User, user_id)
        {:ok, JSON.encode!(user)}

      _ ->
        {:error, "Resource not found"}
    end
  end

  def read(_conn, _uri), do: {:error, "Resource not found"}
end

当客户来电时 resources/read，服务器首先尝试与静态资源进行精确的URI匹配。如果没有匹配，它会按顺序尝试每个资源模板，直到有一个处理URI。

原产地验证

为了防止DNS重新绑定攻击，您可以在传输上启用源验证。启用后，只有具有以下条件的请求 Origin 将接受与允许列表匹配的标头。无请求 Origin 始终允许使用标头（例如来自CLI工具）。

forward "/", EMCP.Transport.StreamableHTTP,
  server: MyApp.MCPServer,
  validate_origin: Mix.env() == :prod,
  allowed_origins: ["example.com", "staging.example.com"]

允许的来源可以指定为：

  • 空列表： [] - 阻止所有带有Origin标头的传入请求
  • 完整网址： "https://example.com" --匹配确切的方案和主机，以及任何端口
  • 裸域名： "example.com" --匹配任何方案(http 或 https)以及任何端口

原点匹配不区分大小写。如果你想允许 全部 起源，刚刚设定 validate_origin: false 或者完全删除此配置。

会话管理

默认情况下，StreamableHTTP会自动重新创建过期或未知的会话。

您的EMCP服务器可能会收到未知会话，例如，如果您在多个服务器上运行应用程序，并且负载平衡器没有根据以下情况路由客户端HTTP请求 mcp-session-id 头球如果负载平衡器将HTTP请求路由到未初始化MCP会话的服务器，则服务器将返回 404 - Session not found.

对于无状态HTTP请求（几乎所有MCP请求都是无状态的），会话是否初始化并不重要。这就是为什么StreamableHTTP默认情况下会重新创建此类未知会话，这种权衡偏离了MCP规范，以提供更好的用户体验。

如果您想遵循MCP规范并返回 404 对于丢失或过期的会话，设置 recreate_missing_session: false.

forward "/", EMCP.Transport.StreamableHTTP,
  server: MyApp.MCPServer,
  recreate_missing_session: false

但请注意，如果没有适当的HTTP请求路由，这可能会激怒您的用户，因为大多数LLM客户端（如Claude Code）在正在进行的对话中不会重新创建过期或未知的MCP会话，并要求用户开始新的对话以重新连接。这对失去对话上下文的用户来说很不方便，但是嘿！-符合MCP规范！

STDIO传输

对于本地开发或CLI工具，您可以使用STDIO传输。它从stdin读取JSON-RPC消息，并将响应写入stdout。

将其添加到您的监督树中：

children = [
  {EMCP.Transport.STDIO, server: MyApp.MCPServer}
]

通过以下方式在Claude代码中配置它 .claude/settings.json:

{
  "mcpServers": {
    "my-app": {
      "command": "mix",
      "args": ["run", "--no-halt"],
      "cwd": "/path/to/your/elixir/project"
    }
  }
}

致谢

根据官方 Ruby MCP SDK 参考实施。

受这些更完整的Elixir实现的启发：

发展

e2e测试使用 MCP检查员 CLI。在运行测试之前安装它：

cd test/inspector && bun install

然后运行测试：

mix test

相关 MCP