MCP源码分析：SSE

mcpServer := server.NewMCPServer(
        "example-server",
        "1.0.0",
        server.WithResourceCapabilities(true, true),
        server.WithPromptCapabilities(true),
        server.WithToolCapabilities(true),
    )
    / Add echo tool
    mcpServer.AddTool(mcp.NewTool("echo",
        mcp.WithDescription("Echo back the input"),
        mcp.WithString("message",
            mcp.Required(),
            mcp.Description("Message to echo back"),
        ),
    ), echoHandler)

sseServer := s.ServeSSE("localhost:8080")

func (s *MCPServer) ServeSSE(addr string) *server.SSEServer {
    return server.NewSSEServer(s.server,
        server.WithBaseURL(fmt.Sprintf("http://%s", addr)),
    )
}

if err := sseServer.Start(":8080"); err != nil {

func NewSSEServer(server *MCPServer, opts ...SSEOption) *SSEServer {
    s := &SSEServer{
        server:                       server,
        sseEndpoint:                  "/sse",
        messageEndpoint:              "/message",
        useFullURLForMessageEndpoint: true,
        keepAlive:                    false,
        keepAliveInterval:            10 * time.Second,
    }

type SSEServer struct {
    server                       *MCPServer
    baseURL                      string
    basePath                     string
    appendQueryToMessageEndpoint bool
    useFullURLForMessageEndpoint bool
    messageEndpoint              string
    sseEndpoint                  string
    sessions                     sync.Map
    srv                          *http.Server
    contextFunc                  HTTPContextFunc
    dynamicBasePathFunc          DynamicBasePathFunc
    keepAlive         bool
    keepAliveInterval time.Duration
    mu sync.RWMutex
}

func (s *SSEServer) Start(addr string) error {
    s.mu.Lock()
    if s.srv == nil {
        s.srv = &http.Server{
            Addr:    addr,
            Handler: s,
        }
    } else {
        if s.srv.Addr == "" {
            s.srv.Addr = addr
        } else if s.srv.Addr != addr {
            return fmt.Errorf("conflicting listen address: WithHTTPServer(%q) vs Start(%q)", s.srv.Addr, addr)
        }
    }
    srv := s.srv
    s.mu.Unlock()
    return srv.ListenAndServe()
}

func (s *SSEServer) SSEHandler() http.Handler {
    return http.HandlerFunc(s.handleSSE)
}

func (s *SSEServer) MessageHandler() http.Handler {
    return http.HandlerFunc(s.handleMessage)
}

func (s *SSEServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    ssePath := s.CompleteSsePath()
    if ssePath != "" && path == ssePath {
        s.handleSSE(w, r)
    messagePath := s.CompleteMessagePath()
    if messagePath != "" && path == messagePath {
        s.handleMessage(w, r)
    http.NotFound(w, r)

func (s *SSEServer) handleSSE(w http.ResponseWriter, r *http.Request) {
    if r.Method != http.MethodGet {
        http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
        return
    }
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    w.Header().Set("Access-Control-Allow-Origin", "*")
    
    sessionID := uuid.New().String()
    
           for {
                select {
                case <-ticker.C:
                    message := mcp.JSONRPCRequest{
                        JSONRPC: "2.0",
                        ID:      session.requestID.Add(1),
                        Request: mcp.Request{
                            Method: "ping",
                        },
                    }
                    messageBytes, _ := json.Marshal(message)
                    pingMsg := fmt.Sprintf("event: message\ndata:%s\n\n", messageBytes)
                    select {
                    case session.eventQueue <- pingMsg:
                    
    endpoint := s.GetMessageEndpointForClient(r, sessionID)
    if s.appendQueryToMessageEndpoint && len(r.URL.RawQuery) > 0 {
        endpoint += "&" + r.URL.RawQuery
    }
    fmt.Fprintf(w, "event: endpoint\ndata: %s\r\n\r\n", endpoint)

func (s *SSEServer) handleMessage(w http.ResponseWriter, r *http.Request) {
    if r.Method != http.MethodPost {
        s.writeJSONRPCError(w, nil, mcp.INVALID_REQUEST, "Method not allowed")
        return
    }
    sessionID := r.URL.Query().Get("sessionId")
    if sessionID == "" {
        s.writeJSONRPCError(w, nil, mcp.INVALID_PARAMS, "Missing sessionId")
        return
    }
    sessionI, ok := s.sessions.Load(sessionID)
    if !ok {
        s.writeJSONRPCError(w, nil, mcp.INVALID_PARAMS, "Invalid session ID")
        return
    }
    session := sessionI.(*sseSession)
    
    var rawMessage json.RawMessage
    if err := json.NewDecoder(r.Body).Decode(&rawMessage); err != nil {
        s.writeJSONRPCError(w, nil, mcp.PARSE_ERROR, "Parse error")
        return
    }
    go func(ctx context.Context) {
        defer cancel()
        // Use the context that will be canceled when session is done
        // Process message through MCPServer
        response := s.server.HandleMessage(ctx, rawMessage)
        // Only send response if there is one (not for notifications)
        if response != nil {
            var message string
            if eventData, err := json.Marshal(response); err != nil {

func (s *MCPServer) HandleMessage(
    ctx context.Context,
    message json.RawMessage,
) mcp.JSONRPCMessage {
     switch baseMessage.Method {
     case mcp.MethodInitialize: