在 《万字长文，带你读懂 Anthropic MCP》中我们介绍了MCP的基本框架和组件，并初步说了在golang中的框架metoro-io/mcp-golang和mark3labs/mcp-go。本文将通过实践和源码的方式先解读mark3labs/mcp-go。

MCP Server一般为轻量的服务端程序，通过一种标准的协议(MCP)暴露出特定资源的一些特定的能力。

初始化后，支持以下模式：

enum ErrorCode {
  // Standard JSON-RPC error codes
  ParseError = -32700,
  InvalidRequest = -32600,
  MethodNotFound = -32601,
  InvalidParams = -32602,
  InternalError = -32603
}

任何一方都可以终止连接：

常见错误码：

 

Request Method	发起方	响应方	描述
initialized	Client	Server	初始化会话
tools-list	Client	Server	发现可用的工具
tools/call	Client	Server	调用工具
resources/list	Client	Server	发现可用的资源
resources/read	Client	Server	获取资源内容
resources/templates	Client	Server	发现可用的参数化资源
resources/subscribe	Client	Server	订阅特定资源，监听其变化事件
prompts/list	Client	Server	发现可用的提示词
prompts/get	Client	Server	获取特定提示词
roots/list	Server	Client	列出服务器有权访问的客户端文件系统根节点（暴露目录和文件）
sampling/create	Server	Client	启用服务器的AI生成能力（ sampling creation ）

我们就以框架中的官方示例代码为引子一步步解读流程和框架代码。

package main

import (
    "context"
    "errors"
    "fmt"

    "github.com/mark3labs/mcp-go/mcp"
    "github.com/mark3labs/mcp-go/server"
)

func main() {
    // 创建MCP服务器
    s := server.NewMCPServer(
       "Demo 🚀", // 服务器名称
       "1.0.0",  // 服务器版本
    )

    // 添加工具
    tool := mcp.NewTool("hello_world", // 工具名称
       mcp.WithDescription("Say hello to someone"), // 工具描述
       mcp.WithString("name", // 参数名称
          mcp.Required(), // 参数是必需的
          mcp.Description("Name of the person to greet"), // 参数描述
       ),
    )

    // 为工具添加处理器
    s.AddTool(tool, helloHandler)

    // 启动标准输入输出服务器
    if err := server.ServeStdio(s); err != nil {
       fmt.Printf("Server error: %v\n", err) // 打印服务器错误
    }
}

func helloHandler(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
    // 从请求参数中获取名字参数，并断言为字符串类型
    name, ok := request.Params.Arguments["name"].(string)
    if !ok {
       // 如果断言失败，返回错误
       return nil, errors.New("name must be a string")
    }

    // 返回包含问候语的结果
    return mcp.NewToolResultText(fmt.Sprintf("Hello, %s!", name)), nil
}

 

就以上面的代码为例，将上面的代码编译为二进制命令：

$ go build -v -o server

再启动mcp inspetor：

$ npx -y @modelcontextprotocol/inspector ./server

这样一个简单的MCP Client和MCP Server就搭建好了，后续也为我们开发测试构建好了环境。

在上面的代码中main函数中的第一个代码就是server.NewMCPServer，那我们就从MCPServer这个结构体入手。

代码地址：https://github.com/mark3labs/mcp-go/blob/e183dd17cfec07072a188f6169033bf61f7bf37d/server/server.go#L135

type MCPServer struct {
    mu                   sync.RWMutex                       // 用于保护共享资源，确保并发访问时的数据一致性
    name                 string                             // 服务器的名称，用于标识服务器
    version              string                             // 服务器的版本，用于跟踪和管理服务器的不同版本
    instructions         string                             // 服务器的指令，通常在初始化响应中返回给客户端，提供使用指南或帮助信息
    resources            map[string]resourceEntry           // 存储服务器支持的资源及其处理函数
    resourceTemplates    map[string]resourceTemplateEntry   // 存储资源模板及其处理函数，支持URI模板匹配多类似资源
    prompts              map[string]mcp.Prompt              // 存储服务器支持的提示，用于与用户交互
    promptHandlers       map[string]PromptHandlerFunc       // 存储处理提示请求的函数，每个提示对应一个处理函数
    tools                map[string]ServerTool              // 存储服务器支持的工具及其处理函数
    notificationHandlers map[string]NotificationHandlerFunc // 存储处理传入通知的函数，接收客户端通知并处理
    capabilities         serverCapabilities                 // 定义服务器支持的功能特性，包括资源、提示、工具和日志记录等
    sessions             sync.Map                           // 存储当前活跃的客户端会话，用于跟踪用户交互
    initialized          atomic.Bool                        // 使用原子操作标记服务器是否已初始化，确保线程安全
    hooks                *Hooks                             // 存储服务器钩子，允许在请求处理前后或返回错误前执行自定义逻辑
}

MCPServer 是 Model Control Protocol (MCP) 服务器的实现，用于处理包括资源、提示和工具在内的各种类型的请求。

看看如何创建一个MCPServer对象。

func NewMCPServer(
    name, version string,
    opts ...ServerOption,
) *MCPServer {
    s := &MCPServer{
       resources:            make(map[string]resourceEntry),
       resourceTemplates:    make(map[string]resourceTemplateEntry),
       prompts:              make(map[string]mcp.Prompt),
       promptHandlers:       make(map[string]PromptHandlerFunc),
       tools:                make(map[string]ServerTool),
       name:                 name,
       version:              version,
       notificationHandlers: make(map[string]NotificationHandlerFunc),
       capabilities: serverCapabilities{
          tools:     nil,
          resources: nil,
          prompts:   nil,
          logging:   false,
       },
    }

    for _, opt := range opts {
       opt(s)
    }

    return s
}

在NewMCPServer()方法中，我们需要关注的opts ...ServerOption，进一步看看ServerOption有哪些选项：

选项	功能	使用方式
WithResourceCapabilities	配置资源相关的服务器功能，如订阅和资源列表变化通知	WithResourceCapabilities(subscribe, listChanged bool)
WithHooks	添加钩子函数，用于在请求处理前后执行特定逻辑	WithHooks(hooks *Hooks)
WithPromptCapabilities	配置提示相关的服务器功能，如提示列表变化通知	WithPromptCapabilities(listChanged bool)
WithToolCapabilities	配置工具相关的服务器功能，如工具列表变化通知	WithToolCapabilities(listChanged bool)
WithLogging	启用服务器日志记录功能	WithLogging()
WithInstructions	设置服务器指令，用于在初始化响应中返回给客户端	WithInstructions(instructions string)

它接受一个 Hooks 类型的指针作为参数。允许在创建 MCPServer 实例时，为服务器添加自定义的钩子函数，这些钩子函数可以在请求处理前后或返回错误给客户端之前执行。

hooks机制对开发和流程是非常有效的。框架中给的hooks能力有：

字段名	类型	描述
OnBeforeAny	[]BeforeAnyHookFunc	在请求被解析后但方法调用前执行的钩子函数。
OnSuccess	[]OnSuccessHookFunc	在请求成功生成结果但结果尚未发送给客户端之前执行的钩子函数。
OnError	[]OnErrorHookFunc	在请求解析或方法执行过程中发生错误时执行的钩子函数。
OnBeforeInitialize	[]OnBeforeInitializeFunc	在处理初始化请求前执行的钩子函数。
OnAfterInitialize	[]OnAfterInitializeFunc	在处理初始化请求后执行的钩子函数。
OnBeforePing	[]OnBeforePingFunc	在处理 Ping 请求前执行的钩子函数。
OnAfterPing	[]OnAfterPingFunc	在处理 Ping 请求后执行的钩子函数。
OnBeforeListResources	[]OnBeforeListResourcesFunc	在处理列出资源请求前执行的钩子函数。
OnAfterListResources	[]OnAfterListResourcesFunc	在处理列出资源请求后执行的钩子函数。
OnBeforeListResourceTemplates	[]OnBeforeListResourceTemplatesFunc	在处理列出资源模板请求前执行的钩子函数。
OnAfterListResourceTemplates	[]OnAfterListResourceTemplatesFunc	在处理列出资源模板请求后执行的钩子函数。
OnBeforeReadResource	[]OnBeforeReadResourceFunc	在处理读取资源请求前执行的钩子函数。
OnAfterReadResource	[]OnAfterReadResourceFunc	在处理读取资源请求后执行的钩子函数。
OnBeforeListPrompts	[]OnBeforeListPromptsFunc	在处理列出提示请求前执行的钩子函数。
OnAfterListPrompts	[]OnAfterListPromptsFunc	在处理列出提示请求后执行的钩子函数。
OnBeforeGetPrompt	[]OnBeforeGetPromptFunc	在处理获取提示请求前执行的钩子函数。
OnAfterGetPrompt	[]OnAfterGetPromptFunc	在处理获取提示请求后执行的钩子函数。
OnBeforeListTools	[]OnBeforeListToolsFunc	在处理列出工具请求前执行的钩子函数。
OnAfterListTools	[]OnAfterListToolsFunc	在处理列出工具请求后执行的钩子函数。
OnBeforeCallTool	[]OnBeforeCallToolFunc	在处理调用工具请求前执行的钩子函数。
OnAfterCallTool	[]OnAfterCallToolFunc	在处理调用工具请求后执行的钩子函数。

现在模拟创建一个完整的MCPServer：

hooks := &server.Hooks{}

hooks.AddBeforeAny(func(id any, method mcp.MCPMethod, message any) {
    fmt.Printf("beforeAny: %s, %v, %v\n", method, id, message)
})
hooks.AddOnSuccess(func(id any, method mcp.MCPMethod, message any, result any) {
    fmt.Printf("onSuccess: %s, %v, %v, %v\n", method, id, message, result)
})
hooks.AddOnError(func(id any, method mcp.MCPMethod, message any, err error) {
    fmt.Printf("onError: %s, %v, %v, %v\n", method, id, message, err)
})
hooks.AddBeforeInitialize(func(id any, message *mcp.InitializeRequest) {
    fmt.Printf("beforeInitialize: %v, %v\n", id, message)
})
hooks.AddAfterInitialize(func(id any, message *mcp.InitializeRequest, result *mcp.InitializeResult) {
    fmt.Printf("afterInitialize: %v, %v, %v\n", id, message, result)
})
hooks.AddAfterCallTool(func(id any, message *mcp.CallToolRequest, result *mcp.CallToolResult) {
    fmt.Printf("afterCallTool: %v, %v, %v\n", id, message, result)
})
hooks.AddBeforeCallTool(func(id any, message *mcp.CallToolRequest) {
    fmt.Printf("beforeCallTool: %v, %v\n", id, message)
})
// 创建MCP服务器
s := server.NewMCPServer(
    "Demo 🚀", // 服务器名称
    "1.0.0",  // 服务器版本
    server.WithLogging(),
    server.WithToolCapabilities(true),
    server.WithResourceCapabilities(true, true),
    server.WithPromptCapabilities(true),
    server.WithInstructions("initialized"),
    server.WithHooks(hooks),
)

 

mark3labs/mcp-go框架中创建Tool有两个方式：

下面通过一段代码进行对比：

方式一：mcp.NewTool()
tool := mcp.NewTool("hello_world", // 工具名称
    mcp.WithDescription("Say hello to someone"), // 工具描述
    mcp.WithString("name", // 参数名称
       mcp.Required(), // 参数是必需的
       mcp.Description("Name of the person to greet"), // 参数描述
    ),
)

方式二：mcp.NewToolWithRawSchema()
rawSchema := json.RawMessage(`{
    "type": "object",
    "properties": {
       "name": {"type": "string", "description": "Name of the person to greet"}
    },
    "required": ["name"]
}`)

// Create a tool with raw schema
toolRS := mcp.NewToolWithRawSchema("hello_world_1", "Say hello to someone", rawSchema)

其中rawSchema的结构需要符合ToolInputSchema结构体：

type ToolInputSchema struct {
    Type       string                 `json:"type"`
    Properties map[string]interface{} `json:"properties,omitempty"`
    Required   []string               `json:"required,omitempty"`
}

需要注意的是Properties来源于jsonSchema，所以具备的需要校验属性，比如default、maximum、minimum、maxLength、minLength、enum等等。其中key为请求传入的参数字段，interface{}为对key的各种属性校验。

更建议使用方式一：mcp.NewTool()更加符合编码方式，也更好控制器生成的 jsonSchema 。

创建好Tool值之后，调用server的方法AddTool()将Tool添加进入，相当于web框架中的添加路由与相关handle的关系。相关代码如下：

// AddTool registers a new tool and its handler
func (s *MCPServer) AddTool(tool mcp.Tool, handler ToolHandlerFunc) {
    s.AddTools(ServerTool{Tool: tool, Handler: handler})
}

// AddTools registers multiple tools at once
func (s *MCPServer) AddTools(tools ...ServerTool) {
    // 检查工具
    if s.capabilities.tools == nil {
       s.capabilities.tools = &toolCapabilities{}
    }

    // 加锁
    s.mu.Lock()
    // 遍历工具
    for _, entry := range tools {
       s.tools[entry.Tool.Name] = entry
    }
    // 获取初始化状态
    initialized := s.initialized.Load()
    s.mu.Unlock()

    // 发送通知
    if initialized {
       s.sendNotificationToAllClients("notifications/tools/list_changed", nil)
    }
}

可以看出Tool是放入到MCPServer的tools字段中，使用name作为key，ServerTool作为value，其中ServerTool结构如下：

type ServerTool struct {
    Tool    mcp.Tool
    Handler ToolHandlerFunc
}

框架还提供了：

老规矩先来一个demo，再看器源码实现：

// Static resource example - exposing a README file
resource := mcp.NewResource(
    "docs://readme",
    "Project README",
    mcp.WithResourceDescription("The project's README file"),
    mcp.WithMIMEType("text/markdown"),
)

// Add resource with its handler
s.AddResource(resource, func(ctx context.Context, request mcp.ReadResourceRequest) ([]mcp.ResourceContents, error) {
    content, err := os.ReadFile("main.go")
    if err != nil {
       return nil, err
    }

    return []mcp.ResourceContents{
       mcp.TextResourceContents{
          URI:      "docs://readme",
          MIMEType: "text/markdown",
          Text:     string(content),
       },
    }, nil
})

 

在MCP中Resource是指允许Server公开可供客户端读取并用作交互上下文的数据和内容。有很多类型的Resource：

框架中Resource的结构体如下：

type Resource struct {
    Annotated // 包含可选注解，用于告知客户端如何使用或显示对象

    // 资源的URI
    URI string `json:"uri"` // 资源的唯一标识符，用于定位和访问资源

    // 资源的可读名称
    //
    // 客户端可以使用此名称来填充UI元素
    Name string `json:"name"` // 资源的显示名称，便于用户理解和界面展示

    // 对此资源所代表内容的描述
    //
    // 客户端可以使用此描述来帮助大型语言模型（LLM）理解可用资源
    // 这可以看作是对模型的“提示”
    Description string `json:"description,omitempty"` // 资源的详细描述，为模型提供上下文信息

    // 如果已知，此资源的MIME类型
    MIMEType string `json:"mimeType,omitempty"` // 资源的媒体类型，如text/plain、image/jpeg等，用于指示资源的内容格式
}

type Annotated struct {
    Annotations *struct {
       // 描述此对象或数据的预期客户是谁
       //
       // 它可以包含多个条目，以指示对多个受众有用的内容（例如，`["user", "assistant"]`）
       Audience []Role `json:"audience,omitempty"` // 受众群体，指示数据对哪些角色或用户群体有用

       // 描述此数据对服务器操作的重要性
       //
       // 值为1表示“最重要”，并指示数据实际上是必需的，而0表示“最不重要”，并指示数据完全是可选的
       Priority float64 `json:"priority,omitempty"` // 优先级，表示数据对服务器操作的重要程度，范围从0（最不重要）到1（最重要）
    } `json:"annotations,omitempty"`
}

其中：

func (s *MCPServer) AddResource(
    resource mcp.Resource,
    handler ResourceHandlerFunc,
) {
    // 检查资源
    if s.capabilities.resources == nil {
       s.capabilities.resources = &resourceCapabilities{}
    }

    // 加锁
    s.mu.Lock()
    // 解锁（defer）
    defer s.mu.Unlock()

    // 存储资源
    s.resources[resource.URI] = resourceEntry{
       resource: resource,
       handler:  handler,
    }
}

resource是放入到MCPServer的resources字段中，使用URI作为key，resourceEntry作为value，其中resourceEntry结构如下：

type resourceEntry struct {
    resource mcp.Resource
    handler  ResourceHandlerFunc
}

框架还提供了添加Resource templates的功能，主要是针对动态资源，服务器可以公开 URI 模板 ，客户端可以使用它来构建有效的资源 URI。

// Dynamic resource example - user profiles by ID
template := mcp.NewResourceTemplate(
    "users://{id}/profile",
    "User Profile",
    mcp.WithTemplateDescription("Returns user profile information"),
    mcp.WithTemplateMIMEType("application/json"),
)

// Add template with its handler
s.AddResourceTemplate(template, func(ctx context.Context, request mcp.ReadResourceRequest) ([]mcp.ResourceContents, error) {
    // Extract ID from the URI using regex matching
    // The server automatically matches URIs to templates
    userID := extractIDFromURI(request.Params.URI)

    profile := fmt.Sprintf("Hello %s", userID) // Your DB/API call here

    return []mcp.ResourceContents{
       mcp.TextResourceContents{
          URI:      request.Params.URI,
          MIMEType: "application/json",
          Text:     profile,
       },
    }, nil
})

// extractIDFromURI 从给定的 URI 中提取用户 ID。
// 假设 URI 的格式为 "users://{id}/profile"。
func extractIDFromURI(uri string) string {
    // 定义正则表达式来匹配 URI 中的 ID
    re := regexp.MustCompile(`users://([^/]+)/profile`)

    // 使用正则表达式查找匹配项
    matches := re.FindStringSubmatch(uri)

    // 如果找到了匹配项，并且匹配项的数量正确，则返回 ID
    if len(matches) == 2 {
       return matches[1]
    }

    // 如果没有找到匹配项，或者匹配项的数量不正确，则返回空字符串
    return ""
}

 

Prompts创建可重复使用的提示模板和工作流程，提示使服务器能够定义可重复使用的提示模板和工作流程。

// Simple greeting prompt
s.AddPrompt(mcp.NewPrompt("greeting",
    mcp.WithPromptDescription("A friendly greeting prompt"),
    mcp.WithArgument("name",
       mcp.ArgumentDescription("Name of the person to greet"),
    ),
), func(ctx context.Context, request mcp.GetPromptRequest) (*mcp.GetPromptResult, error) {
    name := request.Params.Arguments["name"]
    if name == "" {
       name = "friend"
    }

    return mcp.NewGetPromptResult(
       "A friendly greeting",
       []mcp.PromptMessage{
          mcp.NewPromptMessage(
             mcp.RoleAssistant,
             mcp.NewTextContent(fmt.Sprintf("Hello, %s! How can I help you today?", name)),
          ),
       },
    ), nil
})

 

Prompt的结构体如下：

// Prompt 表示服务器提供的提示或提示模板。
// 如果 Arguments 非空且包含元素，则表示该提示是一个模板，
// 在调用 prompts/get 时需要提供参数值。
// 如果 Arguments 为空或为 nil，则这是一个不需要参数的静态提示。
type Prompt struct {
    // 提示或提示模板的名称。
    Name string `json:"name"`
    // 提示提供内容的可选描述。
    Description string `json:"description,omitempty"`
    // 用于模板化提示的参数列表。
    // 参数的存在表明这是一个模板提示。
    Arguments []PromptArgument `json:"arguments,omitempty"`
}

// PromptArgument 描述提示模板可以接受的参数。
// 当提示包含参数时，客户端在发出 prompts/get 请求时
// 必须为所有必需参数提供值。
type PromptArgument struct {
    // 参数的名称。
    Name string `json:"name"`
    // 参数的可读描述。
    Description string `json:"description,omitempty"`
    // 此参数是否必须提供。
    // 如果为 true，则客户端在调用 prompts/get 时必须包含此参数。
    Required bool `json:"required,omitempty"`
}

可以通过mcp.NewPrompt方法来生成Prompt对象：

func NewPrompt(name string, opts ...PromptOption) Prompt {
    prompt := Prompt{
       Name: name,
    }

    for _, opt := range opts {
       opt(&prompt)
    }

    return prompt
}

又见到了opts ...PromptOption，看看有哪些选项：

现在代码已经来到了：

// 启动标准输入输出服务器
if err := server.ServeStdio(s); err != nil {
    fmt.Printf("Server error: %v\n", err) // 打印服务器错误
}

MCP当前主要提供两类Stdio transport和HTTP with SSE transport 。

StdioServer的结构体：

type StdioServer struct {
    server      *MCPServer
    errLogger   *log.Logger
    contextFunc StdioContextFunc
}

其中在，重要的字段有server和contextFunc，给MCPServer进来就是为了具备MCP的能力，只是使用stdio的传输方式。contextFunc是为了让外部自定义的context可以进入到StdioServer，可以用于结束服务和控制超时.

// 为MCP Server 启用stdio
func ServeStdio(server *MCPServer, opts ...StdioOption) error {
    // 创建Stdio服务器
    s := NewStdioServer(server)
    // 设置错误日志
    s.SetErrorLogger(log.New(os.Stderr, "", log.LstdFlags))

    // 应用选项
    for _, opt := range opts {
       opt(s)
    }

    // 创建上下文
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // 设置信号通道
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGTERM, syscall.SIGINT)

    // 监听信号
    go func() {
       <-sigChan
       cancel()
    }()

    // 开始监听
    return s.Listen(ctx, os.Stdin, os.Stdout)
}

其中最关键的是Listen方法，它是 StdioServer 类型的一个关键方法，用于监听标准输入输出的 JSON-RPC 消息。

// Listen 启动监听，从提供的输入读取 JSON-RPC 消息，并将响应写入提供的输出。
// 它将持续运行，直到上下文被取消或发生错误。
// 如果在读取输入或写入输出时遇到问题，将返回错误。
func (s *StdioServer) Listen(
    ctx context.Context, // 监听过程的上下文，用于控制生命周期和传递请求范围的信息
    stdin io.Reader, // 标准输入流，用于读取客户端发送的 JSON-RPC 消息
    stdout io.Writer, // 标准输出流，用于将响应写回客户端
) error {
    // 由于标准输入输出只有一个客户端，因此设置一个静态客户端上下文，SessionId为stdio
    if err := s.server.RegisterSession(&stdioSessionInstance); err != nil {
       // 如果会话注册失败，返回错误
       return fmt.Errorf("register session: %w", err)
    }
    // 确保在函数结束时注销会话
    defer s.server.UnregisterSession(stdioSessionInstance.SessionID())
    // 更新上下文，将会话信息加入
    ctx = s.server.WithContext(ctx, &stdioSessionInstance)

    // 如果存在自定义上下文函数，则应用该函数修改上下文
    if s.contextFunc != nil {
       ctx = s.contextFunc(ctx)
    }

    // 创建一个带缓冲的读取器，用于从标准输入流高效读取数据
    reader := bufio.NewReader(stdin)

    // 启动一个协程专门处理通知
    go func() {
       for {
          select {
          case notification := <-stdioSessionInstance.notifications:
             // 收到通知时，调用 writeResponse 方法将通知写入标准输出
             err := s.writeResponse(notification, stdout)
             if err != nil {
                // 如果写入通知时出错，记录错误日志
                s.errLogger.Printf("Error writing notification: %v", err)
             }
          case <-ctx.Done():
             // 如果上下文完成，退出协程
             return
          }
       }
    }()

    // 主循环，用于处理输入消息
    for {
       select {
       case <-ctx.Done():
          // 如果上下文完成，返回上下文错误
          return ctx.Err()
       default:
          // 使用协程使读取操作可取消
          readChan := make(chan string, 1) // 用于接收读取到的行
          errChan := make(chan error, 1)   // 用于接收读取错误

          go func() {
             line, err := reader.ReadString('\n') // 读取一行输入
             if err != nil {
                errChan <- err // 发送读取错误
                return
             }
             readChan <- line // 发送读取到的行
          }()

          select {
          case <-ctx.Done():
             // 如果上下文完成，返回上下文错误
             return ctx.Err()
          case err := <-errChan:
             // 处理读取错误
             if err == io.EOF {
                // 如果是文件结束符，表示输入结束，返回 nil
                return nil
             }
             // 其他错误则记录日志并返回
             s.errLogger.Printf("Error reading input: %v", err)
             return err
          case line := <-readChan:
             // 处理读取到的行
             if err := s.processMessage(ctx, line, stdout); err != nil {
                // 如果处理消息时出错
                if err == io.EOF {
                   // 如果是文件结束符，返回 nil
                   return nil
                }
                // 其他错误则记录日志并返回
                s.errLogger.Printf("Error handling message: %v", err)
                return err
             }
          }
       }
    }
}

 

SSE模式凭借其分布式能力、实时性和架构灵活性，成为MCP在 企业级应用、云端协作、动态数据流处理等场景的首选。所以对SSE的支持程度和易用程度，对于一个MCP框架而言是非常重要的。

mark3labs/mcp-go如何支持SSE呢？看看如下代码：

mcpServer := NewMCPServer()
sseServer := server.NewSSEServer(mcpServer, server.WithBaseURL("http://localhost:8080"))
log.Printf("SSE server listening on :8080")
if err := sseServer.Start(":8080"); err != nil {
    log.Fatalf("Server error: %v", err)
}

现在已经知道如何在mark3labs/mcp-go中启用SSE了，现在来分析一下，它是如何实现的。

SSEServer的结构体如下：

// SSEServer implements a Server-Sent Events (SSE) based MCP server.
// It provides real-time communication capabilities over HTTP using the SSE protocol.
type SSEServer struct {
    server          *MCPServer       // MCPServer 实例，用于处理实际的消息传递和通信逻辑
    baseURL         string           // SSE 服务器的基础 URL，用于构建完整的端点路径
    basePath        string           // SSE 服务器的基础路径，通常用于区分不同的服务或版本
    messageEndpoint string           // 消息端点的路径，客户端通过此端点发送 JSON-RPC 消息
    sseEndpoint     string           // SSE 端点的路径，客户端通过此端点建立 SSE 连接
    sessions        sync.Map         // 存储活动 SSE 会话的同步映射，用于跟踪和管理客户端连接
    srv             *http.Server     // 内部的 HTTP 服务器实例，用于处理 HTTP 请求和响应
    contextFunc     SSEContextFunc   // 可选的上下文函数，用于根据请求内容自定义上下文
}

与之前的StdioServer相比，SSEServer多了用于http中使用的URI、path，还有srv这是一个httpServer的类型，用于支持HTTP请求和响应。继续查看如何构建SSEServer：

// NewSSEServer creates a new SSE server instance with the given MCP server and options.
func NewSSEServer(server *MCPServer, opts ...SSEOption) *SSEServer {
    s := &SSEServer{
       server:          server,
       sseEndpoint:     "/sse",
       messageEndpoint: "/message",
    }

    // Apply all options
    for _, opt := range opts {
       opt(s)
    }

    return s
}

还是采用的才是经典Option模式，继续看看SSEOption有哪些选项：

名称	功能	使用方式
WithBaseURL	设置 SSE 服务器的基础 URL	WithBaseURL("https://example.com")
WithBasePath	设置 SSE 服务器的基础路径	WithBasePath("/v1")
WithMessageEndpoint	设置消息端点的路径	WithMessageEndpoint("/custom-message")
WithSSEEndpoint	设置 SSE 端点的路径	WithSSEEndpoint("/custom-sse")
WithHTTPServer	设置 HTTP 服务器实例（通常用于测试或自定义服务器配置）	WithHTTPServer(customHttpServer)
WithContextFunc	设置一个函数，用于根据请求内容自定义上下文	WithContextFunc(func(ctx context.Context, r *http.Request) context.Context { ... })

WithHTTPServer适用于需要自定义服务器配置的场景，为后续替换更好性能的http服务实例打下基础，也体现了mark3labs/mcp-go扩展性。

剩下的Start()就是常见的启动http服务实例的功能了。

func (s *SSEServer) Start(addr string) error {
    s.srv = &http.Server{
       Addr:    addr,
       Handler: s,
    }

    return s.srv.ListenAndServe()
}

还可以调用Shutdown()实现对服务器的关闭。

在实际开发中，很多公司内部的业务有自己的框架，集成了许许多多的独特功能。总不能为了使用MCP重写一套Web框架，此时就需要使用到mark3labs/mcp-go集成到Web框架的能力了。下面以gin框架为例：

// 创建一个新的 Gin 引擎
r := gin.Default()

// 创建一个新的 MCPServer 实例（假设这是 SSEServer 所需的）
mcpServer := server.NewMCPServer("gin-mcp-server", "1.0.0") // 根据你的实际代码调整
// mcpServer 新加Tool、Resource、Prompt
// ... ...
// 创建一个新的 SSEServer 实例，并传入 MCPServer
sseServer := server.NewSSEServer(mcpServer)

// 将 SSEServer 的 SSE 端点和处理函数集成到 Gin 路由中
r.GET(sseServer.CompleteSsePath(), func(c *gin.Context) {
    sseServer.ServeHTTP(c.Writer, c.Request)
})

// 将 SSEServer 的消息端点和处理函数集成到 Gin 路由中
r.POST(sseServer.CompleteMessagePath(), func(c *gin.Context) {
    sseServer.ServeHTTP(c.Writer, c.Request)
})

// 启动 Gin 服务器
if err := r.Run("localhost:8081"); err != nil {
    log.Fatalf("Gin server startup failed: %v", err)
}

上述的代码会生成两个路由：

路由	方法	作用	实例
/sse	GET	获取SessionID 接受Server响应	请求： curl 'http://localhost:3000/sse?transportType=sse&url=http%3A%2F%2Flocalhost%3A8081%2Fsse' \ -H 'Accept: */*' \ -H 'Accept-Language: zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7' \ -H 'Cache-Control: no-cache' \ -H 'Connection: keep-alive' \ -H 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36' 响应： event: endpoint data: /message?sessionId=fee6d6df-d394-4b4d-a748-fddcc73fb766
/message	POST	使用SessionID保持会话 发起功能请求	请求： curl 'http://localhost:3000/message?sessionId=fee6d6df-d394-4b4d-a748-fddcc73fb766' \ -H 'Accept: */*' \ -H 'Cache-Control: no-cache' \ -H 'Connection: keep-alive' \ -H 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36' \ -H 'content-type: application/json' --data-raw '{"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{"sampling":{},"roots":{"listChanged":true}},"clientInfo":{"name":"mcp-inspector","version":"0.7.0"}},"jsonrpc":"2.0","id":0}' /message的响应： {"jsonrpc":"2.0","id":0,"result":{"protocolVersion":"2024-11-05","capabilities":{"logging":{},"prompts":{"listChanged":true},"resources":{"subscribe":true,"listChanged":true},"tools":{}},"serverInfo":{"name":"example-servers/everything","version":"1.0.0"}}} /sse收到的响应： event: message data: {"jsonrpc":"2.0","id":0,"result":{"protocolVersion":"2024-11-05","capabilities":{"logging":{},"prompts":{"listChanged":true},"resources":{"subscribe":true,"listChanged":true},"tools":{}},"serverInfo":{"name":"example-servers/everything","version":"1.0.0"}}}

之前的内容，解析了如何构建MCP Server的实践和背后的实现。下面我们还需要了解mark3labs/mcp-go如何接受请求并进行响应的。

从SSEServer结构体中已经知道使用的http.Server，所以其接受请求的入口为ServeHTTP方法，实现了 http.Handler 接口，用于处理 HTTP 请求。根据请求的路径，它会将请求分发到不同的处理方法（handleSSE 或 handleMessage），或者返回 404 未找到。

func (s *SSEServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    // 获取请求的路径
    path := r.URL.Path

    // 使用精确路径匹配，而不是模糊包含
    ssePath := s.CompleteSsePath() // 获取完整的 SSE 路径
    if ssePath != "" && path == ssePath {
       // 如果请求路径与 SSE 路径匹配，则处理 SSE 请求
       s.handleSSE(w, r)
       return // 处理完成后直接返回，不再继续后续逻辑
    }

    // 获取消息处理的完整路径
    messagePath := s.CompleteMessagePath()
    if messagePath != "" && path == messagePath {
       // 如果请求路径与消息处理路径匹配，则处理消息请求
       s.handleMessage(w, r)
       return // 处理完成后直接返回，不再继续后续逻辑
    }

    // 如果请求路径不匹配任何已知路径，则返回 404 未找到
    http.NotFound(w, r)
}

其中：

这些信息都是可以在NewSSEServer()方法中设置。

需要详细查看的是 s.handleSSE(w, r)和s.handleMessage(w, r)方法，他们分别处理/see和/message的请求。

handleSSE实现了一个处理服务器发送事件（SSE）的HTTP处理器。SSE是一种允许服务器向客户端发送自动更新的技术。主要流程：

// 它设置适当的头信息并为客户端创建一个新的会话。
func (s *SSEServer) handleSSE(w http.ResponseWriter, r *http.Request) {
    // 1. 检查请求方法是否为GET，如果不是，返回405 Method Not Allowed错误
    if r.Method != http.MethodGet {
       http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
       return
    }

    // 2. 设置SSE相关的响应头
    w.Header().Set("Content-Type", "text/event-stream") // 设置内容类型为text/event-stream
    w.Header().Set("Cache-Control", "no-cache")         // 禁用缓存
    w.Header().Set("Connection", "keep-alive")          // 保持连接活跃
    w.Header().Set("Access-Control-Allow-Origin", "*")  // 允许所有域跨域请求

    // 检查ResponseWriter是否支持Flush，如果不支持，返回500 Internal Server Error错误
    flusher, ok := w.(http.Flusher)
    if !ok {
       http.Error(w, "Streaming unsupported", http.StatusInternalServerError)
       return
    }

    // 3. 创建一个新的会话ID和会话对象
    sessionID := uuid.New().String() // 生成唯一的会话ID
    session := &sseSession{
       writer:              w,                          // 响应写入器
       flusher:             flusher,                    // 刷新器
       done:                make(chan struct{}),        // 用于通知会话结束的通道
       eventQueue:          make(chan string, 100),     // 事件队列，缓冲区大小为100
       sessionID:           sessionID,                  // 会话ID
       notificationChannel: make(chan mcp.JSONRPCNotification, 100), // 通知通道，缓冲区大小为100
    }

    // 4. 将会话存储到会话存储中，并在处理完成后删除
    s.sessions.Store(sessionID, session)
    defer s.sessions.Delete(sessionID)

    // 在服务器中注册会话，如果注册失败，返回500 Internal Server Error错误
    if err := s.server.RegisterSession(session); err != nil {
       http.Error(w, fmt.Sprintf("Session registration failed: %v", err), http.StatusInternalServerError)
       return
    }
    // 在处理完成后注销会话
    defer s.server.UnregisterSession(sessionID)

    // 5. 启动一个goroutine处理通知通道中的事件
    go func() {
       for {
          select {
          case notification := <-session.notificationChannel: // 从通知通道接收通知
             eventData, err := json.Marshal(notification) // 将通知序列化为JSON
             if err == nil {
                select {
                case session.eventQueue <- fmt.Sprintf("event: message\ndata: %s\n\n", eventData): // 将事件发送到事件队列
                   // 事件成功入队
                case <-session.done: // 如果会话结束，退出goroutine
                   return
                }
             }
          case <-session.done: // 如果会话结束，退出goroutine
             return
          case <-r.Context().Done(): // 如果请求上下文被取消，退出goroutine
             return
          }
       }
    }()

    // 生成消息端点URL并发送初始的endpoint事件
    messageEndpoint := fmt.Sprintf("%s?sessionId=%s", s.CompleteMessageEndpoint(), sessionID)
    fmt.Fprintf(w, "event: endpoint\ndata: %s\r\n\r\n", messageEndpoint) // 发送endpoint事件
    flusher.Flush() // 刷新响应，确保事件立即发送到客户端

    // 6. 主事件循环，运行在HTTP处理器goroutine中
    for {
       select {
       case event := <-session.eventQueue: // 从事件队列接收事件
          fmt.Fprint(w, event) // 将事件写入响应
          flusher.Flush()      // 刷新响应，确保事件立即发送到客户端
       case <-r.Context().Done(): // 如果请求上下文被取消，关闭会话并退出
          close(session.done)
          return
       }
    }
}

第三阶段又看到Session了，与之前的stdioSession相比，sseSession明显复杂多了，它们都实现接口ClientSession：

type ClientSession interface {
    // NotificationChannel provides a channel suitable for sending notifications to client.
    NotificationChannel() chan<- mcp.JSONRPCNotification
    // SessionID is a unique identifier used to track user session.
    SessionID() string
}

sseSession是专门用于表示一个基于服务器发送事件（Server-Sent Events, SSE）协议的活跃连接。sseSession负责管理客户端与服务器之间的单向实时数据推送，并保持会话。其结构体如下：

type sseSession struct {
    writer              http.ResponseWriter // HTTP响应写入器，用于向客户端发送数据
    flusher             http.Flusher       // HTTP刷新器，用于刷新响应缓冲区，确保数据立即发送给客户端
    done                chan struct{}      // 用于通知会话结束的通道
    eventQueue          chan string        // 用于排队事件的通道，存储待发送给客户端的事件
    sessionID           string             // 会话的唯一标识符
    notificationChannel chan mcp.JSONRPCNotification // 用于接收JSON-RPC通知的通道
}

4.7.3 handleMessage

handleMessage 方法是 SSEServer 类型的一个方法，用于处理传入的 JSON-RPC 消息，并通过 SSE 连接和 HTTP 响应返回结果。其主要流程：

尝试将事件加入会话的事件队列。如果队列已满或会话已关闭，则丢弃事件。

那么handleSSE和handleMessage的关系是怎样的呢？使用一张图来说明：

 

// handleMessage 处理来自客户端的JSON-RPC消息，并通过SSE连接和HTTP响应返回结果
func (s *SSEServer) handleMessage(w http.ResponseWriter, r *http.Request) {
    // 1. 如果请求方法不是POST，则返回JSON-RPC错误响应，表示方法不允许
    if r.Method != http.MethodPost {
       s.writeJSONRPCError(w, nil, mcp.INVALID_REQUEST, "Method not allowed")
       return
    }

    // 从请求的URL查询参数中获取sessionId
    sessionID := r.URL.Query().Get("sessionId")
    // 如果sessionId为空，则返回JSON-RPC错误响应，表示缺少sessionId参数
    if sessionID == "" {
       s.writeJSONRPCError(w, nil, mcp.INVALID_PARAMS, "Missing sessionId")
       return
    }

    // 从session存储中加载与sessionId对应的session
    sessionI, ok := s.sessions.Load(sessionID)
    // 如果session不存在，则返回JSON-RPC错误响应，表示无效的session ID
    if !ok {
       s.writeJSONRPCError(w, nil, mcp.INVALID_PARAMS, "Invalid session ID")
       return
    }
    session := sessionI.(*sseSession)

    // 在处理消息之前设置客户端上下文
    ctx := s.server.WithContext(r.Context(), session)
    // 如果提供了自定义的上下文函数，则应用它
    if s.contextFunc != nil {
       ctx = s.contextFunc(ctx, r)
    }

    // 将请求体解析为原始JSON消息
    var rawMessage json.RawMessage
    if err := json.NewDecoder(r.Body).Decode(&rawMessage); err != nil {
       // 如果解析失败，则返回JSON-RPC错误响应，表示解析错误
       s.writeJSONRPCError(w, nil, mcp.PARSE_ERROR, "Parse error")
       return
    }

    // 通过MCPServer处理消息
    response := s.server.HandleMessage(ctx, rawMessage)

    // 如果存在响应（非通知），则发送响应
    if response != nil {
       // 将响应编码为JSON格式
       eventData, _ := json.Marshal(response)

       // 将事件排队以通过SSE发送
       select {
       case session.eventQueue <- fmt.Sprintf("event: message\ndata: %s\n\n", eventData):
          // 事件成功排队
       case <-session.done:
          // 会话已关闭，不尝试排队
       default:
          // 队列已满，可以记录此情况
       }

       // 发送HTTP响应
       w.Header().Set("Content-Type", "application/json")
       w.WriteHeader(http.StatusAccepted)
       json.NewEncoder(w).Encode(response)
    } else {
       // 对于通知，只发送202 Accepted状态码，无响应体
       w.WriteHeader(http.StatusAccepted)
    }
}

 

需要注意的是HandleMessage方法，这是通过server/internal/gen/request_handler.go.tmpl生成的，也根据MCP协议实现的模版，其流程如下：

以initialize请求为例：

// 根据消息方法进行分情况处理
switch baseMessage.Method {
// 初始化请求处理
case mcp.MethodInitialize:
    var request mcp.InitializeRequest
    var result *mcp.InitializeResult
    // 尝试将原始消息解析为初始化请求类型
    if unmarshalErr := json.Unmarshal(message, &request); unmarshalErr != nil {
       // 如果解析失败，记录错误信息
       err = &requestError{
          id:   baseMessage.ID,
          code: mcp.INVALID_REQUEST,
          err:  &UnparseableMessageError{message: message, err: unmarshalErr, method: baseMessage.Method},
       }
    } else {
       // 执行初始化请求前的钩子函数
       s.hooks.beforeInitialize(baseMessage.ID, &request)
       // 处理初始化请求
       result, err = s.handleInitialize(ctx, baseMessage.ID, request)
    }
    // 如果处理过程中发生错误
    if err != nil {
       // 执行错误钩子函数
       s.hooks.onError(baseMessage.ID, baseMessage.Method, &request, err)
       // 返回错误响应
       return err.ToJSONRPCError()
    }
    // 执行初始化请求后的钩子函数
    s.hooks.afterInitialize(baseMessage.ID, &request, result)
    // 返回成功响应
    return createResponse(baseMessage.ID, *result)

// 其他方法处理逻辑类似，省略...

// 如果方法不匹配任何已知方法，返回方法未找到错误响应
default:
    return createErrorResponse(
       baseMessage.ID,
       mcp.METHOD_NOT_FOUND,
       fmt.Sprintf("Method %s not found", baseMessage.Method),
    )
}

代码比较多，可以自行查看：https://github.com/mark3labs/mcp-go/blob/e183dd17cfec07072a188f6169033bf61f7bf37d/server/request_handler.go#L12

mark3labs/mcp-go框架是一个简单易用的MCP框架，基本上实现了MCP协议，提供对 MCP 核心规范的完整支持，包括资源（Resources）、工具（Tools）、提示（Prompts）等核心组件，确保与主流 LLM 客户端（如 Claude、Cline）的兼容性。尤其是mark3labs/mcp-go提供的hooks机制，可以让开发者更好的使用类似gin空间一样的中间件能力，比如实现统一鉴权等能力。除此之外，还可以与主流的Web框架，如gin框架进行集成，进一步扩展了mark3labs/mcp-go的适用性。