Implement HTTP transport using SSE, Protocol 2024-11-05 (Final)

Makefile

@@ -52,6 +52,6 @@ test: check-go
 	@echo "$(YELLOW)Running tests...$(NC)"
 	go test -v ./...
 
-lint: check_go
+lint: check-go
 	@echo "$(BLUE)Running linter...$(NC)"
-	golangci-lint run ./...
+	golangci-lint run ./...

README.md

@@ -61,7 +61,7 @@ go install github.com/f/mcptools/cmd/mcptools@latest
 
 ## Transport Options
 
-MCP currently supports one transport method for communicating with MCP servers:
+MCP currently supports two transport method for communicating with MCP servers:
 
 ### Stdio Transport
 
@@ -72,6 +72,20 @@ useful for command-line tools that implement the MCP protocol.
 mcp tools npx -y @modelcontextprotocol/server-filesystem ~/Code
 ```
 
+### Http SSE Transport
+
+Uses HTTP and Server-Sent Events (SSE) to communicate with an MCP server via JSON-RPC 2.0.
+This is useful for connecting to remote server that implement the MCP protocol.
+
+```
+mcp tools http://127.0.0.1:3001
+
+# As an example, you can use the everything sample server
+# docker run -p 3001:3001 --rm -it tzolov/mcp-everything-server:v1
+```
+
+_Note:_ Currently HTTP SSE supports only MCP protocol version 2024-11-05.
+
 ## Output Formats
 
 MCP supports three output formats:
@@ -202,7 +216,6 @@ mcp shell npx -y @modelcontextprotocol/server-filesystem ~/Code
 
 The following features are planned for future releases:
 
-- HTTP Transport: Add support for connecting to MCP servers over HTTP
 - Authentication: Support for secure authentication mechanisms
 
 ## License

cmd/mcptools/main.go

@@ -96,6 +96,10 @@ func createClient(args []string) (*client.Client, error) {
 		return nil, errCommandRequired
 	}
 
+	if len(args) == 1 && (strings.HasPrefix(args[0], "http://") || strings.HasPrefix(args[0], "https://")) {
+		return client.NewHTTP(args[0]), nil
+	}
+
 	return client.NewStdio(args), nil
 }
 
@@ -519,7 +523,11 @@ func newShellCmd() *cobra.Command { //nolint:gocyclo
 				os.Exit(1)
 			}
 
-			mcpClient := client.NewStdio(parsedArgs)
+			mcpClient, clientErr := createClient(parsedArgs)
+			if clientErr != nil {
+				fmt.Fprintf(os.Stderr, "Error: %v\n", clientErr)
+				os.Exit(1)
+			}
 
 			_, listErr := mcpClient.ListTools()
 			if listErr != nil {

pkg/client/client.go

@@ -4,6 +4,8 @@ Package client implements mcp client functionality.
 package client
 
 import (
+	"fmt"
+	"os"
 	"strings"
 
 	"github.com/f/mcptools/pkg/transport"
@@ -32,6 +34,19 @@ func NewStdio(command []string) *Client {
 	}
 }
 
+// NewHTTP creates a MCP client that communicates with a server via HTTP using JSON-RPC.
+func NewHTTP(address string) *Client {
+	transport, err := transport.NewHTTP(address)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error creating HTTP transport: %s\n", err)
+		os.Exit(1)
+	}
+
+	return &Client{
+		transport: transport,
+	}
+}
+
 // ListTools retrieves the list of available tools from the MCP server.
 func (c *Client) ListTools() (map[string]any, error) {
 	return c.transport.Execute("tools/list", nil)

pkg/transport/http.go

@@ -0,0 +1,163 @@
+package transport
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+)
+
+// HTTP implements the Transport interface by communicating with a MCP server over HTTP using JSON-RPC.
+type HTTP struct {
+	eventCh chan string
+	address string
+	debug   bool
+	nextID  int
+}
+
+// NewHTTP creates a new Http transport that will execute the given command.
+// It communicates with the command using JSON-RPC over HTTP.
+// Currently Http transport is implements MCP's Final draft version 2024-11-05,
+// https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse
+func NewHTTP(address string) (*HTTP, error) {
+	debug := os.Getenv("MCP_DEBUG") == "1"
+
+	_, uriErr := url.ParseRequestURI(address)
+	if uriErr != nil {
+		return nil, fmt.Errorf("invalid address: %w", uriErr)
+	}
+
+	resp, err := http.Get(address + "/sse")
+	if err != nil {
+		return nil, fmt.Errorf("error sending request: %w", err)
+	}
+
+	eventCh := make(chan string, 1)
+
+	go func() {
+		defer func() {
+			if closeErr := resp.Body.Close(); closeErr != nil {
+				fmt.Fprintf(os.Stderr, "Failed to close response body: %v\n", closeErr)
+			}
+		}()
+
+		reader := bufio.NewReader(resp.Body)
+		for {
+			line, lineErr := reader.ReadString('\n')
+			if lineErr != nil {
+				fmt.Fprintf(os.Stderr, "SSE read error: %v\n", lineErr)
+				return
+			}
+			line = strings.TrimSpace(line)
+			if debug {
+				fmt.Fprintf(os.Stderr, "DEBUG: Received SSE: %s\n", line)
+			}
+			if strings.HasPrefix(line, "data:") {
+				data := strings.TrimSpace(line[5:])
+				select {
+				case eventCh <- data:
+				default:
+				}
+			}
+		}
+	}()
+
+	// First event we receive from SSE is the message address. We will use this endpoint to keep
+	// a session alive.
+	var messageAddress string
+	select {
+	case msg := <-eventCh:
+		messageAddress = msg
+	case <-time.After(10 * time.Second):
+		return nil, fmt.Errorf("timeout waiting for SSE response")
+	}
+
+	return &HTTP{
+		// Use the SSE message address as the base address for the HTTP transport
+		address: address + messageAddress,
+		nextID:  1,
+		debug:   debug,
+		eventCh: eventCh,
+	}, nil
+}
+
+// Execute implements the Transport via JSON-RPC over HTTP.
+func (t *HTTP) Execute(method string, params any) (map[string]any, error) {
+	if t.debug {
+		fmt.Fprintf(os.Stderr, "DEBUG: Connecting to server: %s\n", t.address)
+	}
+
+	request := Request{
+		JSONRPC: "2.0",
+		Method:  method,
+		ID:      t.nextID,
+		Params:  params,
+	}
+	t.nextID++
+
+	requestJSON, err := json.Marshal(request)
+	if err != nil {
+		return nil, fmt.Errorf("error marshaling request: %w", err)
+	}
+
+	requestJSON = append(requestJSON, '\n')
+
+	if t.debug {
+		fmt.Fprintf(os.Stderr, "DEBUG: Sending request: %s\n", string(requestJSON))
+	}
+
+	resp, err := http.Post(t.address, "application/json", bytes.NewBuffer(requestJSON))
+	if err != nil {
+		return nil, fmt.Errorf("error sending request: %w", err)
+	}
+
+	if t.debug {
+		fmt.Fprintf(os.Stderr, "DEBUG: Sent request to server\n")
+	}
+
+	defer func() {
+		if closeErr := resp.Body.Close(); closeErr != nil {
+			fmt.Fprintf(os.Stderr, "Failed to close response body: %v\n", closeErr)
+		}
+	}()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("error reading response: %w", err)
+	}
+
+	if t.debug {
+		fmt.Fprintf(os.Stderr, "DEBUG: Read from server: %s\n", string(body))
+	}
+
+	if len(body) == 0 {
+		return nil, fmt.Errorf("no response from server")
+	}
+
+	// After sending the request, we listen the SSE channel for the response
+	var response Response
+	select {
+	case msg := <-t.eventCh:
+		if unmarshalErr := json.Unmarshal([]byte(msg), &response); unmarshalErr != nil {
+			return nil, fmt.Errorf("error unmarshaling response: %w, response: %s", unmarshalErr, msg)
+		}
+	case <-time.After(10 * time.Second):
+		return nil, fmt.Errorf("timeout waiting for SSE response")
+	}
+
+	if response.Error != nil {
+		return nil, fmt.Errorf("RPC error %d: %s", response.Error.Code, response.Error.Message)
+	}
+
+	if t.debug {
+		fmt.Fprintf(os.Stderr, "DEBUG: Successfully parsed response\n")
+	}
+
+	return response.Result, nil
+}

pkg/transport/transport.go

@@ -1,3 +1,4 @@
+// Package transport contains implementatations for different transport options for MCP.
 package transport
 
 import (