MCP ex

MCP Elixir SDK

Elixir SDK 模型上下文协议 （MCP）——一种用于将LLM应用程序与外部数据源和工具集成的开放协议。

提供两者 客户端 和 服务器 使用可插拔传输（stdio、Streamable HTTP）的实现。

100%一致性 使用官方MCP测试套件（Tier 1）。

特性

• MCP客户端 --连接到任何MCP服务器，发现和调用工具，阅读资源，使用提示
• MCP服务器 --通过处理程序行为向MCP客户端公开工具、资源和提示
• 运输 --stdio（子进程）和流式HTTP（POST+SSE）
• 完全协议支持 --初始化握手、能力协商、通知、分页
• 异步工具执行 --工具可以发送日志消息、进度更新，并在执行过程中发出双向请求（采样、启发）
• 一致性测试 --30/30场景，40/40检查官方MCP一致性套件

协议版本

实施MCP规范 2025-11-25.

安装

添加 mcp_elixir_sdk 您的依赖关系 mix.exs:

def deps do
  [
    {:mcp_elixir_sdk, "~> 1.0"}
  ]
end

对于 可流式传输的HTTP 传输支持，还添加了以下可选依赖项：

def deps do
  [
    {:mcp_elixir_sdk, "~> 1.0"},
    {:req, "~> 0.5"},        # HTTP client (for MCP client over HTTP)
    {:plug, "~> 1.16"},      # HTTP framework (for MCP server over HTTP)
    {:bandit, "~> 1.5"}      # HTTP server (for MCP server over HTTP)
  ]
end

stdio传输无需额外依赖。

客户示例

示例1：连接到stdio MCP服务器

连接到作为子进程运行的MCP服务器。客户端启动服务器 通过stdin/stdout进行处理和通信。

# Start the client with a stdio transport
{:ok, client} = MCP.Client.start_link(
  transport: {MCP.Transport.Stdio, command: "npx", args: ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]},
  client_info: %{name: "my_app", version: "1.0.0"}
)

# Perform the initialization handshake
{:ok, info} = MCP.Client.connect(client)
IO.puts("Connected to #{info.server_info.name} #{info.server_info.version}")

# List available tools
{:ok, result} = MCP.Client.list_tools(client)
for tool  "/tmp/hello.txt"})
IO.puts("Result: #{hd(result["content"])["text"]}")

# List and read resources
{:ok, result} = MCP.Client.list_resources(client)
for resource 
    # Forward to your LLM and return the result
    {:ok, %{
      "role" => "assistant",
      "content" => %{"type" => "text", "text" => "Sample response"},
      "model" => "my-model",
      "stopReason" => "endTurn"
    }}
  end,
  # Report filesystem roots to the server
  on_roots_list: fn _params ->
    {:ok, %{"roots" => [
      %{"uri" => "file:///home/user/project", "name" => "Project"}
    ]}}
  end,
  # Receive server notifications
  notification_handler: fn method, params ->
    IO.puts("Notification: #{method} #{inspect(params)}")
  end
)

# Connect and use the server
{:ok, _info} = MCP.Client.connect(client)

# Use pagination helpers to list all tools across pages
{:ok, all_tools} = MCP.Client.list_all_tools(client)
IO.puts("Found #{length(all_tools)} tools")

# Get a prompt template and use it
{:ok, result} = MCP.Client.get_prompt(client, "code_review", %{"language" => "elixir"})
IO.inspect(result["messages"])

MCP.Client.close(client)

服务器示例

示例1：带有工具和资源的Stdio服务器

定义一个实现 MCP.Server.Handler 行为和 在stdio上运行它。服务器根据以下内容自动检测功能 你实现的回调。

defmodule MyHandler do
  @behaviour MCP.Server.Handler

  @impl true
  def init(_opts), do: {:ok, %{}}

  @impl true
  def handle_list_tools(_cursor, state) do
    tools = [
      %{
        "name" => "get_weather",
        "description" => "Get current weather for a city",
        "inputSchema" => %{
          "type" => "object",
          "properties" => %{
            "city" => %{"type" => "string", "description" => "City name"}
          },
          "required" => ["city"]
        }
      },
      %{
        "name" => "calculate",
        "description" => "Evaluate a math expression",
        "inputSchema" => %{
          "type" => "object",
          "properties" => %{
            "expression" => %{"type" => "string"}
          },
          "required" => ["expression"]
        }
      }
    ]

    {:ok, tools, nil, state}
  end

  @impl true
  def handle_call_tool("get_weather", %{"city" => city}, state) do
    # Your weather API logic here
    {:ok, [%{"type" => "text", "text" => "Weather in #{city}: 72F, sunny"}], state}
  end

  def handle_call_tool("calculate", %{"expression" => expr}, state) do
    case Code.eval_string(expr) do
      {result, _} ->
        {:ok, [%{"type" => "text", "text" => "#{result}"}], state}
    end
  rescue
    _ -> {:error, -32_602, "Invalid expression", state}
  end

  @impl true
  def handle_list_resources(_cursor, state) do
    resources = [
      %{"uri" => "config://app", "name" => "App Config", "mimeType" => "application/json"}
    ]

    {:ok, resources, nil, state}
  end

  @impl true
  def handle_read_resource("config://app", state) do
    config = Jason.encode!(%{debug: false, version: "1.0.0"})
    {:ok, [%{"uri" => "config://app", "text" => config}], state}
  end

  def handle_read_resource(uri, state) do
    {:error, -32_002, "Resource not found: #{uri}", state}
  end
end

# Run as a stdio server (for use as a subprocess)
{:ok, _server} = MCP.Server.start_link(
  transport: {MCP.Transport.Stdio, mode: :server},
  handler: {MyHandler, []},
  server_info: %{name: "my-server", version: "1.0.0"}
)

示例2：带有异步工具的HTTP服务器

使用Plug+Bandit通过流式HTTP提供服务。此示例演示 异步工具执行 handle_call_tool/4，接收a ToolContext 用于发送日志消息、进度更新和制作 工具执行期间服务器到客户端的请求。

defmodule MyAsyncHandler do
  @behaviour MCP.Server.Handler

  alias MCP.Server.ToolContext

  @impl true
  def init(_opts), do: {:ok, %{}}

  @impl true
  def handle_list_tools(_cursor, state) do
    tools = [
      %{
        "name" => "analyze_code",
        "description" => "Analyze code with LLM assistance",
        "inputSchema" => %{
          "type" => "object",
          "properties" => %{
            "code" => %{"type" => "string"},
            "language" => %{"type" => "string"}
          },
          "required" => ["code"]
        }
      }
    ]

    {:ok, tools, nil, state}
  end

  # 4-arity handle_call_tool enables async execution with ToolContext
  @impl true
  def handle_call_tool("analyze_code", args, ctx, state) do
    code = args["code"]
    language = args["language"] || "unknown"

    # Send log messages to the client during execution
    ToolContext.log(ctx, "info", "Starting analysis of #{language} code")

    # Report progress
    ToolContext.send_progress(ctx, 0, 100)

    # Request LLM sampling from the client.
    # The server's request_timeout (default 30s) ensures this returns
    # even if the client can't respond (see "Sampling over HTTP" note below).
    sampling_result = ToolContext.request_sampling(ctx, %{
      "messages" => [
        %{
          "role" => "user",
          "content" => %{
            "type" => "text",
            "text" => "Analyze this #{language} code:\n\n#{code}"
          }
        }
      ],
      "maxTokens" => 1000
    })

    ToolContext.send_progress(ctx, 100, 100)
    ToolContext.log(ctx, "info", "Analysis complete")

    analysis =
      case sampling_result do
        {:ok, result} ->
          result["content"]["text"]

        {:error, _reason} ->
          # Fallback when sampling is unavailable or times out
          "Static analysis: #{language} code, #{String.length(code)} characters"
      end

    {:ok, [%{"type" => "text", "text" => analysis}], state}
  end

  @impl true
  def handle_list_prompts(_cursor, state) do
    prompts = [
      %{
        "name" => "review",
        "description" => "Code review prompt",
        "arguments" => [
          %{"name" => "code", "description" => "Code to review", "required" => true}
        ]
      }
    ]

    {:ok, prompts, nil, state}
  end

  @impl true
  def handle_get_prompt("review", %{"code" => code}, state) do
    result = %{
      "description" => "Code review",
      "messages" => [
        %{
          "role" => "user",
          "content" => %{
            "type" => "text",
            "text" => "Please review this code:\n\n#{code}"
          }
        }
      ]
    }

    {:ok, result, state}
  end
end

# Start the HTTP server
plug_config = MCP.Transport.StreamableHTTP.Plug.init(
  server_mod: MyAsyncHandler,
  server_opts: [
    server_info: %{name: "my-http-server", version: "1.0.0"}
  ]
)

{:ok, _bandit} = Bandit.start_link(
  plug: {MCP.Transport.StreamableHTTP.Plug, plug_config},
  port: 8080,
  ip: {127, 0, 0, 1}
)

IO.puts("MCP server running at http://localhost:8080/mcp")

通过HTTP采样

当使用 ToolContext.request_sampling/2 通过流式HTTP传输， 请注意，客户的 Req.post 是同步的——它会一直阻塞，直到 整个SSE响应流完成。这意味着客户端无法处理或 响应服务器的采样请求，同时 tools/call POST仍然 在飞行中，采样请求将始终超时。

服务器的 request_timeout 选项（默认值：30秒）起到安全作用 net：超时后， request_sampling 回报 {:error, :timeout} 和那个 工具处理程序可以继续回退。始终处理您的错误情况 工具处理程序，如上例所示。

随着 stdio传输，由于消息流，采样工作是双向的 独立于stdin/stdout——客户端可以响应采样请求 同时仍在等待工具结果。

处理程序行为参考

这 MCP.Server.Handler 行为有一个必需的回调(init/1)以及 每个MCP功能的可选回调。服务器自动播发 处理程序实现回调所基于的功能。

示例

看 mcp_ex_示例 对于完整的、可运行的示例项目：

文档

• MCP规范（2025-11-25）
• 建筑 --模块图、数据流、传输设计
• 入职 --贡献者的完整背景

许可证

麻省理工学院