一、为什么用 Go 写 MCP Server？

上篇我们用 Java 手写了 MCP 协议，但生产环境需要更多工程能力：

Go 的 goroutine + channel 模型天然适合 MCP Server 的并发场景。本篇构建一个功能完备的生产级 MCP Server。

二、项目结构与核心接口

mcp-server-go/
├── go.mod
├── cmd/server/main.go          // 入口
├── internal/
│   ├── protocol/
│   │   ├── jsonrpc.go          // JSON-RPC 2.0 类型
│   │   └── mcp.go              // MCP 协议类型 (Tool, Resource, Prompt)
│   ├── transport/
│   │   └── stdio.go            // stdio 传输层
│   ├── server/
│   │   └── server.go           // MCP Server 核心
│   └── tools/
│       ├── registry.go         // 工具注册表
│       ├── weather.go          // 天气工具
│       └── calculator.go       // 计算器工具
└── mcp-server-go               // 编译产物

核心接口定义

// internal/protocol/mcp.go
package protocol

// Tool 代表一个 MCP 工具
type Tool struct {
    Name        string      `json:"name"`
    Description string      `json:"description"`
    InputSchema InputSchema `json:"inputSchema"`
}

type InputSchema struct {
    Type       string              `json:"type"`
    Properties map[string]Property `json:"properties"`
    Required   []string            `json:"required"`
}

type Property struct {
    Type        string `json:"type"`
    Description string `json:"description"`
}

// ToolHandler 是工具的执行函数 —— 任何工具都实现此签名
type ToolHandler func(ctx context.Context, args map[string]interface{}) (*CallToolResult, error)

// CallToolResult 工具调用返回
type CallToolResult struct {
    Content []ContentItem `json:"content"`
    IsError bool          `json:"isError,omitempty"`
}

type ContentItem struct {
    Type string `json:"type"`
    Text string `json:"text,omitempty"`
    Data string `json:"data,omitempty"`     // base64 编码的二进制数据
    MimeType string `json:"mimeType,omitempty"`
}

// Resource 代表 MCP 资源（文件、数据库记录等）
type Resource struct {
    URI         string `json:"uri"`
    Name        string `json:"name"`
    Description string `json:"description,omitempty"`
    MimeType    string `json:"mimeType,omitempty"`
}

type ReadResourceResult struct {
    Contents []ContentItem `json:"contents"`
}

三、工具注册表：并发安全的动态注册

// internal/tools/registry.go
package tools

import (
    "context"
    "fmt"
    "sync"

    "github.com/example/mcp-server-go/internal/protocol"
)

// Registry 工具注册表 —— 线程安全，支持运行时动态增减工具
type Registry struct {
    mu       sync.RWMutex
    tools    map[string]protocol.Tool
    handlers map[string]protocol.ToolHandler
}

func NewRegistry() *Registry {
    return &Registry{
        tools:    make(map[string]protocol.Tool),
        handlers: make(map[string]protocol.ToolHandler),
    }
}

// Register 注册工具（可运行时调用）
func (r *Registry) Register(tool protocol.Tool, handler protocol.ToolHandler) {
    r.mu.Lock()
    defer r.mu.Unlock()
    r.tools[tool.Name] = tool
    r.handlers[tool.Name] = handler
}

// List 返回所有已注册工具
func (r *Registry) List() []protocol.Tool {
    r.mu.RLock()
    defer r.mu.RUnlock()

    result := make([]protocol.Tool, 0, len(r.tools))
    for _, t := range r.tools {
        result = append(result, t)
    }
    return result
}

// Call 调用指定工具
func (r *Registry) Call(ctx context.Context, name string, args map[string]interface{}) (*protocol.CallToolResult, error) {
    r.mu.RLock()
    handler, ok := r.handlers[name]
    r.mu.RUnlock()

    if !ok {
        return nil, fmt.Errorf("unknown tool: %s", name)
    }
    return handler(ctx, args)
}

四、并发工具调用：goroutine 池 + 超时控制

这是生产级 Server 与玩具实现的最大区别。MCP Client 可以同时发起多个 tools/call 请求，Server 必须有并发控制。

// internal/server/server.go
package server

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"

    "github.com/example/mcp-server-go/internal/protocol"
    "github.com/example/mcp-server-go/internal/tools"
    "github.com/example/mcp-server-go/internal/transport"
)

type Config struct {
    MaxConcurrentTools int           // 最大并发工具调用数，0 = 不限制
    ToolTimeout        time.Duration // 单次工具调用超时
    ShutdownTimeout    time.Duration // 优雅关闭等待时间
}

func DefaultConfig() Config {
    return Config{
        MaxConcurrentTools: 10,
        ToolTimeout:        30 * time.Second,
        ShutdownTimeout:    10 * time.Second,
    }
}

// Server 是 MCP Server 的核心结构
type Server struct {
    cfg      Config
    transport *transport.StdioTransport
    registry *tools.Registry
    semaphore chan struct{}      // 并发控制信号量
    wg        sync.WaitGroup     // 等待所有工具调用完成
    done      chan struct{}      // 关闭信号
}

func New(cfg Config, registry *tools.Registry) *Server {
    s := &Server{
        cfg:      cfg,
        transport: transport.NewStdio(),
        registry: registry,
        done:     make(chan struct{}),
    }
    if cfg.MaxConcurrentTools > 0 {
        s.semaphore = make(chan struct{}, cfg.MaxConcurrentTools)
    }
    return s
}

// Run 启动 MCP Server 主循环
func (s *Server) Run() error {
    s.transport.Start()
    log.SetOutput(os.Stderr) // 日志走 stderr，不污染协议通道

    // 监听系统信号，实现优雅关闭
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

    go func() {
        <-sigCh
        log.Println("[server] received shutdown signal")
        s.Shutdown()
    }()

    // 主事件循环
    log.Println("[server] MCP Server ready, waiting for client...")
    for {
        select {
        case <-s.done:
            return nil
        default:
        }

        req, err := s.transport.Receive()
        if err != nil {
            select {
            case <-s.done:
                return nil
            default:
                log.Printf("[server] receive error: %v", err)
                return err
            }
        }

        resp := s.handleRequest(req)
        if resp != nil {
            if err := s.transport.Send(resp); err != nil {
                log.Printf("[server] send error: %v", err)
            }
        }
    }
}

// handleRequest 处理单条 JSON-RPC 请求
func (s *Server) handleRequest(req *protocol.JSONRPCRequest) *protocol.JSONRPCResponse {
    switch req.Method {
    case "initialize":
        return s.handleInitialize(req)
    case "tools/list":
        return s.handleToolsList(req)
    case "tools/call":
        return s.handleToolsCall(req)
    default:
        return &protocol.JSONRPCResponse{
            JSONRPC: "2.0",
            ID:      req.ID,
            Error: &protocol.JSONRPCError{
                Code:    -32601,
                Message: fmt.Sprintf("Method not found: %s", req.Method),
            },
        }
    }
}

// handleToolsCall —— 并发控制的入口
func (s *Server) handleToolsCall(req *protocol.JSONRPCRequest) *protocol.JSONRPCResponse {
    // 解析工具名和参数
    name, _ := req.Params["name"].(string)
    args, _ := req.Params["arguments"].(map[string]interface{})

    // 1. 信号量控制并发
    if s.semaphore != nil {
        select {
        case s.semaphore <- struct{}{}:
            defer func() { <-s.semaphore }()
        default:
            // 并发数已满，返回错误
            return s.errorResponse(req.ID, -32000,
                fmt.Sprintf("too many concurrent tool calls (max=%d)", s.cfg.MaxConcurrentTools))
        }
    }

    // 2. context 超时控制
    ctx, cancel := context.WithTimeout(context.Background(), s.cfg.ToolTimeout)
    defer cancel()

    // 3. 执行工具调用（通过 channel 实现异步-同步桥接）
    type callResult struct {
        result *protocol.CallToolResult
        err    error
    }
    resultCh := make(chan callResult, 1)

    s.wg.Add(1)
    go func() {
        defer s.wg.Done()
        defer func() {
            if r := recover(); r != nil {
                resultCh <- callResult{
                    err: fmt.Errorf("tool panic: %v", r),
                }
            }
        }()
        result, err := s.registry.Call(ctx, name, args)
        resultCh <- callResult{result: result, err: err}
    }()

    // 4. 等待结果或超时
    select {
    case <-ctx.Done():
        return s.errorResponse(req.ID, -32001,
            fmt.Sprintf("tool call timed out after %v", s.cfg.ToolTimeout))
    case cr := <-resultCh:
        if cr.err != nil {
            return s.errorResponse(req.ID, -32000, cr.err.Error())
        }
        return s.successResponse(req.ID, map[string]interface{}{
            "content": cr.result.Content,
            "isError": cr.result.IsError,
        })
    }
}

func (s *Server) handleToolsList(req *protocol.JSONRPCRequest) *protocol.JSONRPCResponse {
    tools := s.registry.List()
    return s.successResponse(req.ID, map[string]interface{}{
        "tools": tools,
    })
}

// Shutdown 优雅关闭
func (s *Server) Shutdown() {
    log.Println("[server] shutting down...")

    // 1. 停止接收新请求
    close(s.done)

    // 2. 等待正在执行的工具调用完成（有超时）
    done := make(chan struct{})
    go func() {
        s.wg.Wait()
        close(done)
    }()

    select {
    case <-done:
        log.Println("[server] all tool calls completed")
    case <-time.After(s.cfg.ShutdownTimeout):
        log.Printf("[server] shutdown timeout after %v, forcing exit", s.cfg.ShutdownTimeout)
    }

    s.transport.Close()
    log.Println("[server] shutdown complete")
}

func (s *Server) successResponse(id interface{}, result interface{}) *protocol.JSONRPCResponse {
    return &protocol.JSONRPCResponse{
        JSONRPC: "2.0",
        ID:      id,
        Result:  result,
    }
}

func (s *Server) errorResponse(id interface{}, code int, message string) *protocol.JSONRPCResponse {
    return &protocol.JSONRPCResponse{
        JSONRPC: "2.0",
        ID:      id,
        Error: &protocol.JSONRPCError{
            Code:    code,
            Message: message,
        },
    }
}

五、传输层：缓冲与流控

// internal/transport/stdio.go
package transport

import (
    "bufio"
    "encoding/json"
    "fmt"
    "io"
    "os"
    "sync"
)

type StdioTransport struct {
    reader  *bufio.Reader
    writer  *bufio.Writer
    mu      sync.Mutex // 保护 writer
    closing bool
}

func NewStdio() *StdioTransport {
    return &StdioTransport{
        reader: bufio.NewReaderSize(os.Stdin, 64*1024),  // 64KB 读缓冲
        writer: bufio.NewWriterSize(os.Stdout, 64*1024), // 64KB 写缓冲
    }
}

func (t *StdioTransport) Start() {
    fmt.Fprintln(os.Stderr, "[transport] stdio transport ready")
}

func (t *StdioTransport) Send(resp *JSONRPCResponse) error {
    t.mu.Lock()
    defer t.mu.Unlock()

    data, err := json.Marshal(resp)
    if err != nil {
        return fmt.Errorf("marshal response: %w", err)
    }

    if _, err := t.writer.Write(data); err != nil {
        return fmt.Errorf("write: %w", err)
    }
    if err := t.writer.WriteByte('\n'); err != nil {
        return fmt.Errorf("write delimiter: %w", err)
    }
    return t.writer.Flush()
}

func (t *StdioTransport) Receive() (*JSONRPCRequest, error) {
    line, err := t.reader.ReadBytes('\n')
    if err != nil {
        if err == io.EOF {
            return nil, fmt.Errorf("client closed connection")
        }
        return nil, fmt.Errorf("read: %w", err)
    }

    var req JSONRPCRequest
    if err := json.Unmarshal(line, &req); err != nil {
        return nil, fmt.Errorf("unmarshal request: %w", err)
    }
    return &req, nil
}

func (t *StdioTransport) Close() {
    t.mu.Lock()
    defer t.mu.Unlock()
    t.closing = true
}

// JSONRPCRequest / JSONRPCResponse
type JSONRPCRequest struct {
    JSONRPC string                 `json:"jsonrpc"`
    ID      interface{}            `json:"id"`
    Method  string                 `json:"method"`
    Params  map[string]interface{} `json:"params,omitempty"`
}

type JSONRPCResponse struct {
    JSONRPC string                 `json:"jsonrpc"`
    ID      interface{}            `json:"id"`
    Result  interface{}            `json:"result,omitempty"`
    Error   *JSONRPCError          `json:"error,omitempty"`
}

type JSONRPCError struct {
    Code    int    `json:"code"`
    Message string `json:"message"`
}

六、注册工具并启动

// cmd/server/main.go
package main

import (
    "context"
    "fmt"
    "time"

    "github.com/example/mcp-server-go/internal/protocol"
    "github.com/example/mcp-server-go/internal/server"
    "github.com/example/mcp-server-go/internal/tools"
)

func main() {
    registry := tools.NewRegistry()

    // 注册天气工具
    registry.Register(protocol.Tool{
        Name:        "get_weather",
        Description: "查询指定城市的天气",
        InputSchema: protocol.InputSchema{
            Type: "object",
            Properties: map[string]protocol.Property{
                "city": {Type: "string", Description: "城市名称"},
            },
            Required: []string{"city"},
        },
    }, func(ctx context.Context, args map[string]interface{}) (*protocol.CallToolResult, error) {
        city, _ := args["city"].(string)
        // 模拟 API 调用 —— 通过 ctx 支持超时取消
        select {
        case <-ctx.Done():
            return nil, ctx.Err()
        case <-time.After(200 * time.Millisecond):
        }
        weather := fmt.Sprintf("%s 当前天气：晴 28°C", city)
        return &protocol.CallToolResult{
            Content: []protocol.ContentItem{{Type: "text", Text: weather}},
        }, nil
    })

    // 注册计算器工具
    registry.Register(protocol.Tool{
        Name:        "calculator",
        Description: "执行数学计算",
        InputSchema: protocol.InputSchema{
            Type: "object",
            Properties: map[string]protocol.Property{
                "expression": {Type: "string", Description: "数学表达式，如 2+3*4"},
            },
            Required: []string{"expression"},
        },
    }, func(ctx context.Context, args map[string]interface{}) (*protocol.CallToolResult, error) {
        expr, _ := args["expression"].(string)
        // 生产环境：用 go/parser 安全解析表达式
        result := fmt.Sprintf("结果：%s = 14 (演示)", expr)
        return &protocol.CallToolResult{
            Content: []protocol.ContentItem{{Type: "text", Text: result}},
        }, nil
    })

    cfg := server.Config{
        MaxConcurrentTools: 10,
        ToolTimeout:        30 * time.Second,
        ShutdownTimeout:    10 * time.Second,
    }

    srv := server.New(cfg, registry)
    if err := srv.Run(); err != nil {
        fmt.Fprintf(os.Stderr, "server error: %v\n", err)
        os.Exit(1)
    }
}

需要导入："os"

七、工程化能力对照