模型上下文协议库

McpServer是一个Elixir库,用于定义模型上下文协议(MCP)工具、提示和路由器,支持输入/输出验证和交互式提示。

  • by GuillaumeMilan
  • Elixir
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 GuillaumeMilan/mcp_server

使用说明

McpServer

McpServer是一个Elixir库,它构建了一个DSL,用于在Elixir中定义模型上下文协议(MCP)工具、提示和路由器。它允许您轻松地公开具有输入/输出模式和验证的工具端点,以及定义具有参数完成的交互式提示。

主要特点

  • 类型安全结构:所有MCP协议结构现在都是具有编译时验证的Elixir结构类型
  • 连接上下文:所有控制器功能接收 conn 参数作为其第一个参数,通过以下方式提供对会话信息、用户数据和其他特定连接上下文的访问 conn.session_id 和 conn.private
  • 已验证的工具:定义具有自动输入验证和输出模式的工具
  • 互动提示:创建具有参数完成支持的提示
  • 资源管理:使用URI模板定义和提供资源
  • 图标元数据:通过将图标附加到工具、提示和资源 icon DSL宏
  • 自动JSON编码:所有结构都自动编码为正确的MCP JSON格式

安装和设置

  1. 将依赖项添加到您的 mix.exs:
def deps do
  [
    {:mcp_server, "~> 0.10.0"},
    {:bandit, "~> 1.0"} # HTTP server
  ]
end
  1. 定义您的MCP路由器:

创建一个使用 McpServer.Router 并定义您的工具和提示。例子:

defmodule MyApp.MyController do
  import McpServer.Controller, only: [message: 3, completion: 2, content: 3]
  alias McpServer.Tool.Content, as: ToolContent
  alias McpServer.Tool.CallResult

  # Tool functions - all receive conn as first parameter
  # Return {:ok, CallResult.new(content: [...])} or {:error, reason}
  def echo(_conn, args) do
    {:ok, CallResult.new(content: [ToolContent.text(Map.get(args, "message", "default"))])}
  end

  def greet(conn, args) do
    name = Map.get(args, "name", "World")
    {:ok, CallResult.new(content: [ToolContent.text("Hello, #{name}, you are connected with session #{conn.session_id}!")])}
  end

  def calculate(_conn, args) do
    result = Map.get(args, "a", 0) + Map.get(args, "b", 0)
    {:ok, CallResult.new(content: [ToolContent.text("#{result}")])}
  end

  # Prompt functions - all receive conn as first parameter
  def get_greet_prompt(_conn, %{"user_name" => user_name}) do
    [
      message("user", "text", "Hello #{user_name}! Welcome to our MCP server. How can I assist you today?"),
      message("assistant", "text", "I'm here to help you with any questions or tasks you might have.")
    ]
  end

  def complete_greet_prompt(_conn, "user_name", user_name_prefix) do
    names = ["Alice", "Bob", "Charlie", "David"]
    filtered_names = Enum.filter(names, &String.starts_with?(&1, user_name_prefix))
    completion(filtered_names, total: 100, has_more: true)
  end

  # Resource reader example - receives conn as first parameter, returns ReadResult struct
  def read_user(_conn, %{"id" => id}) do
    McpServer.Resource.ReadResult.new(
      contents: [
        content(
          "User #{id}",
          "https://example.com/users/#{id}",
          mimeType: "application/json",
          text: "{\"id\": \"#{id}\", \"name\": \"User #{id}\"}",
          title: "User title #{id}"
        )
      ]
    )
  end
end

defmodule MyApp.Router do
  use McpServer.Router

  # Define tools
  tool "greet", "Greets a person", MyApp.MyController, :greet do
    input_field("name", "The name to greet", :string, required: false)
    output_field("greeting", "The greeting message", :string)
  end

  tool "calculate", "Adds two numbers", MyApp.MyController, :calculate do
    input_field("a", "First number", :integer, required: true)
    input_field("b", "Second number", :integer, required: true)
    output_field("result", "The sum of the numbers", :integer)
  end

  tool "echo", "Echoes back the input", MyApp.MyController, :echo,
    title: "Echo",
    hints: [:read_only, :non_destructive, :idempotent, :closed_world] do
    icon "https://example.com/echo-icon.svg", mime_type: "image/svg+xml", sizes: ["48x48"]
    input_field("message", "The message to echo", :string, required: true)
    output_field("response", "The echoed message", :string)
  end

  # Define prompts
  prompt "greet", "A friendly greeting prompt that welcomes users" do
    icon "https://example.com/greet-icon.png"
    argument("user_name", "The name of the user to greet", required: true)
    get MyApp.MyController, :get_greet_prompt
    complete MyApp.MyController, :complete_greet_prompt
  end

  # Define resources
  resource "user", "https://example.com/users/{id}" do
    icon "https://example.com/user-icon.png", mime_type: "image/png"
    description "User resource"
    mimeType "application/json"
    title "User title"
    read MyApp.MyController, :read_user
    complete MyApp.MyController, :complete_user
  end
end
  1. 使用路由器启动Bandit服务器:

添加到您的应用程序监督树中:

确保尊重推荐 MCP服务器的安全选项

children = [
  {Bandit, plug: {
              McpServer.HttpPlug,
              router: MyApp.Router,
              server_info: %{name: "MyApp MCP Server", version: "1.0.0"}
            }, port: 4000, ip: {127, 0, 0, 1}}
]

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

您的MCP服务器现在将运行并提供您定义的工具和提示。

工具

工具是MCP客户端可以调用的函数。它们支持输入验证和输出模式。

工具定义

tool "tool_name", "Description", ControllerModule, :function_name do
  input_field("param", "Parameter description", :type, required: true)
  output_field("result", "Result description", :type)
end

控制器实现

工具控制器功能接收 conn 和 args,必须返回 {:ok, CallResult.new(...)} 或 {:error, reason}:

alias McpServer.Tool.Content, as: ToolContent
alias McpServer.Tool.CallResult

# Return text content
def my_tool(_conn, %{"query" => query}) do
  {:ok, CallResult.new(content: [ToolContent.text("Results for: #{query}")])}
end

# Return multiple content types
def generate_chart(_conn, %{"data" => data}) do
  chart_image = create_chart(data)
  {:ok, CallResult.new(content: [
    ToolContent.text("Chart generated successfully"),
    ToolContent.image(chart_image, "image/png")
  ])}
end

# Return audio content
def synthesize_speech(_conn, %{"text" => text}) do
  audio_data = text_to_speech(text)
  {:ok, CallResult.new(content: [
    ToolContent.text("Speech synthesized"),
    ToolContent.audio(audio_data, "audio/wav")
  ])}
end

# Return structured content for UI rendering
def get_weather(_conn, %{"location" => location}) do
  weather = fetch_weather(location)
  {:ok, CallResult.new(
    content: [ToolContent.text("Weather in #{location}: #{weather.temp}°F")],
    structured_content: %{
      "temperature" => weather.temp,
      "humidity" => weather.humidity
    }
  )}
end

# Signal an error
def risky_tool(_conn, %{"input" => input}) do
  case process(input) do
    {:ok, result} -> {:ok, CallResult.new(content: [ToolContent.text(result)])}
    {:error, reason} -> {:error, reason}
  end
end

提示

提示是具有参数完成支持的交互式消息模板。它们对于生成结构化对话很有用。

快速定义

prompt "prompt_name", "Description" do
  argument("arg_name", "Argument description", required: true)
  get ControllerModule, :get_function
  complete ControllerModule, :complete_function
end

控制器实现

提示控制器需要两个功能:

  1. 获取函数 -接收 conn 参数返回消息列表:
def get_prompt_name(conn, %{"arg_name" => value}) do
  # Access session info via conn.session_id or conn.private
  [
    message("user", "text", "User message with #{value}"),
    message("assistant", "text", "Assistant response"),
    message("system", "text", "System instructions")
  ]
end
  1. 完成函数 -接收 conn,参数名称和前缀,返回完成建议:
def complete_prompt_name(conn, "arg_name", prefix) do
  # Access session info via conn.session_id or conn.private
  suggestions = ["option1", "option2", "option3"]
  filtered = Enum.filter(suggestions, &String.starts_with?(&1, prefix))
  completion(filtered, total: 100, has_more: true)
end

Helper函数

这 McpServer.Prompt 模块提供实用功能:

  • message(role, type, content) -创建消息结构
  • completion(values, opts) -创建完成响应

结构参考

该库为所有MCP消息类型和数据结构提供类型化结构。所有结构都实现 Jason.Encoder 用于JSON序列化。

看 STRUCTURES.md 以获取所有可用结构及其字段的完整参考。

MCP应用程序(交互式UI)

McpServer支持MCP Apps扩展(io.modelcontextprotocol/ui)用于在AI对话的同时提供交互式UI。这包括:

  • UI工具 --将工具链接到UI资源 ui 和 visibility 选项
  • 用户界面资源 --通过以下方式提供HTML内容 ui:// 具有CSP和沙盒权限的URI
  • 结构化内容 --通过以下方式返回丰富的数据以进行UI渲染 McpServer.Tool.CallResult

看 MCP_APPS.md 完整的指南。

使用与测试

测试工具

defmodule MyApp.RouterTest do
  use ExUnit.Case

  setup do
    %{conn: %McpServer.Conn{session_id: "test-session"}}
  end

  test "call a tool", %{conn: conn} do
    {:ok, result} = MyApp.Router.call_tool(conn, "echo", %{"message" => "Hello World"})
    assert %McpServer.Tool.CallResult{content: [content]} = result
    assert %McpServer.Tool.Content.Text{text: "Hello World"} = content
  end

  test "list tools", %{conn: conn} do
    {:ok, tools} = MyApp.Router.list_tools(conn)
    echo = Enum.find(tools, &(&1.name == "echo"))
    assert echo.description == "Echoes back the input"
  end
end

测试提示

defmodule MyApp.PromptTest do
  use ExUnit.Case

  setup do
    %{conn: %McpServer.Conn{session_id: "test-session"}}
  end

  test "get prompt messages", %{conn: conn} do
    {:ok, messages} = MyApp.Router.get_prompt(conn, "greet", %{"user_name" => "Alice"})
    assert hd(messages).role == "user"
    assert hd(messages).content.text =~ "Hello Alice"
  end

  test "complete prompt arguments", %{conn: conn} do
    {:ok, completion} = MyApp.Router.complete_prompt(conn, "greet", "user_name", "A")
    assert "Alice" in completion.values
    assert completion.has_more == true
  end

  test "list prompts", %{conn: conn} do
    {:ok, prompts} = MyApp.Router.prompts_list(conn)
    greet = Enum.find(prompts, &(&1.name == "greet"))
    assert greet.description == "A friendly greeting prompt that welcomes users"
  end
end