feat: add mcp client, convert eino tools

internal/mcp/config.go

@@ -0,0 +1,89 @@
+package mcp
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"time"
+)
+
+// MCPSettings represents the main configuration structure
+type MCPSettings struct {
+	MCPServers map[string]ServerConfig `json:"mcpServers"`
+}
+
+// ServerConfig represents configuration for a single MCP server
+type ServerConfig struct {
+	TransportType string        `json:"transportType,omitempty"` // "sse" or "stdio"
+	AutoApprove   []string      `json:"autoApprove,omitempty"`
+	Disabled      bool          `json:"disabled,omitempty"`
+	Timeout       time.Duration `json:"timeout,omitempty"`
+
+	// SSE specific config
+	URL string `json:"url,omitempty"`
+
+	// Stdio specific config
+	Command string            `json:"command"`
+	Args    []string          `json:"args"`
+	Env     map[string]string `json:"env,omitempty"`
+}
+
+const (
+	DefaultMCPTimeoutSeconds = 30
+	MinMCPTimeoutSeconds     = 5
+)
+
+// GetTimeoutDuration converts timeout seconds to time.Duration
+func (c *ServerConfig) GetTimeoutDuration() time.Duration {
+	if c.Timeout == 0 {
+		return time.Duration(DefaultMCPTimeoutSeconds) * time.Second
+	}
+	return c.Timeout
+}
+
+// LoadSettings loads MCP settings from the config file
+func LoadSettings(path string) (*MCPSettings, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read settings file: %w", err)
+	}
+
+	var settings MCPSettings
+	if err := json.Unmarshal(data, &settings); err != nil {
+		return nil, fmt.Errorf("failed to parse settings: %w", err)
+	}
+
+	if err := validateSettings(&settings); err != nil {
+		return nil, fmt.Errorf("invalid settings: %w", err)
+	}
+
+	return &settings, nil
+}
+
+// validateSettings validates the MCP settings
+func validateSettings(settings *MCPSettings) error {
+	if settings == nil {
+		return fmt.Errorf("settings cannot be nil")
+	}
+
+	for name, server := range settings.MCPServers {
+		if server.Timeout > 0 && server.Timeout < time.Duration(MinMCPTimeoutSeconds)*time.Second {
+			return fmt.Errorf("server %s: timeout must be at least %d seconds", name, MinMCPTimeoutSeconds)
+		}
+
+		switch server.TransportType {
+		case "sse":
+			if server.URL == "" {
+				return fmt.Errorf("server %s: URL is required for SSE transport", name)
+			}
+		case "stdio", "":
+			if server.Command == "" {
+				return fmt.Errorf("server %s: command is required for stdio transport", name)
+			}
+		default:
+			return fmt.Errorf("server %s: unsupported transport type: %s", name, server.TransportType)
+		}
+	}
+
+	return nil
+}

internal/mcp/config_test.go

@@ -0,0 +1,31 @@
+package mcp
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLoadSettings(t *testing.T) {
+	// Load settings from test.mcp.json
+	settings, err := LoadSettings("test.mcp.json")
+	if err != nil {
+		t.Fatalf("Failed to load settings: %v", err)
+	}
+
+	// Verify settings are loaded correctly
+	assert.NotNil(t, settings)
+	assert.Contains(t, settings.MCPServers, "filesystem")
+	assert.Contains(t, settings.MCPServers, "weather")
+
+	// Verify specific server configurations
+	filesystemConfig := settings.MCPServers["filesystem"]
+	assert.Equal(t, "npx", filesystemConfig.Command)
+	assert.Equal(t, []string{"-y", "@modelcontextprotocol/server-filesystem", "/Users/debugtalk/Downloads/tmp"}, filesystemConfig.Args)
+
+	weatherConfig := settings.MCPServers["weather"]
+	assert.Equal(t, "uv", weatherConfig.Command)
+	assert.Equal(t, []string{"--directory", "/Users/debugtalk/MyProjects/ByteDance/EvalTools/quickstart-resources/weather-server-python/", "run", "weather.py"}, weatherConfig.Args)
+	assert.Equal(t, []string{"get_forecast"}, weatherConfig.AutoApprove)
+	assert.Equal(t, map[string]string{"ABC": "123"}, weatherConfig.Env)
+}

internal/mcp/hub.go

@@ -0,0 +1,264 @@
+package mcp
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	mcpp "github.com/cloudwego/eino-ext/components/tool/mcp"
+	"github.com/cloudwego/eino/components/tool"
+	"github.com/httprunner/httprunner/v5/internal/version"
+	"github.com/mark3labs/mcp-go/client"
+	"github.com/mark3labs/mcp-go/mcp"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+)
+
+type MCPTools struct {
+	Name  string
+	Tools []mcp.Tool
+	Err   error
+}
+
+type MCPHub struct {
+	mu          sync.RWMutex
+	connections map[string]*Connection
+	config      *MCPSettings
+}
+
+type Connection struct {
+	Client client.MCPClient
+	Config ServerConfig
+}
+
+func NewMCPHub(configPath string) (*MCPHub, error) {
+	settings, err := LoadSettings(configPath)
+	if err != nil {
+		return nil, err
+	}
+	return &MCPHub{
+		connections: make(map[string]*Connection),
+		config:      settings,
+	}, nil
+}
+
+// InitServers initializes all enabled MCP servers
+func (h *MCPHub) InitServers(ctx context.Context) error {
+	for name, config := range h.config.MCPServers {
+		if config.Disabled {
+			continue
+		}
+
+		if err := h.connectToServer(ctx, name, config); err != nil {
+			return fmt.Errorf("failed to connect to server %s: %w", name, err)
+		}
+	}
+
+	return nil
+}
+
+// GetClient returns the client for the specified server
+func (h *MCPHub) GetClient(serverName string) (client.MCPClient, error) {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	conn, exists := h.connections[serverName]
+	if !exists {
+		return nil, fmt.Errorf("no connection found for server %s", serverName)
+	}
+
+	if conn.Config.Disabled {
+		return nil, fmt.Errorf("server %s is disabled", serverName)
+	}
+
+	return conn.Client, nil
+}
+
+// connectToServer establishes connection to a single MCP server
+func (h *MCPHub) connectToServer(ctx context.Context, serverName string, config ServerConfig) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	// Close existing connection if any
+	if existing, exists := h.connections[serverName]; exists {
+		if err := existing.Client.Close(); err != nil {
+			return fmt.Errorf("failed to close existing connection: %w", err)
+		}
+		delete(h.connections, serverName)
+	}
+
+	var mcpClient client.MCPClient
+	var err error
+
+	// create client
+	switch config.TransportType {
+	case "sse":
+		mcpClient, err = client.NewSSEMCPClient(config.URL)
+	case "stdio", "": // default to stdio
+		var env []string
+		for k, v := range config.Env {
+			env = append(env, fmt.Sprintf("%s=%s", k, v))
+		}
+		mcpClient, err = client.NewStdioMCPClient(config.Command,
+			env, config.Args...)
+	default:
+		return fmt.Errorf("unsupported transport type: %s", config.TransportType)
+	}
+	if err != nil {
+		return fmt.Errorf("failed to create client: %w", err)
+	}
+
+	// prepare client init request
+	initRequest := mcp.InitializeRequest{}
+	initRequest.Params.ProtocolVersion = mcp.LATEST_PROTOCOL_VERSION
+	initRequest.Params.Capabilities = mcp.ClientCapabilities{}
+	initRequest.Params.ClientInfo = mcp.Implementation{
+		Name:    "HttpRunner",
+		Version: version.VERSION,
+	}
+
+	// initialize client
+	_, err = mcpClient.Initialize(ctx, initRequest)
+	if err != nil {
+		mcpClient.Close()
+		return errors.Wrapf(err, "initialize MCP client for %s failed", serverName)
+	}
+
+	log.Info().Str("server", serverName).Msg("connected to MCP server")
+	h.connections[serverName] = &Connection{
+		Client: mcpClient,
+		Config: config,
+	}
+	return nil
+}
+
+// GetTools fetches available tools from all connected MCP servers
+func (h *MCPHub) GetTools(ctx context.Context) map[string]MCPTools {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	results := make(map[string]MCPTools)
+
+	for serverName, conn := range h.connections {
+		if conn.Config.Disabled {
+			continue
+		}
+
+		// get tools from MCP server tools
+		listResults, err := conn.Client.ListTools(ctx, mcp.ListToolsRequest{})
+		if err != nil {
+			results[serverName] = MCPTools{
+				Name:  serverName,
+				Tools: nil,
+				Err:   fmt.Errorf("failed to get tools: %w", err),
+			}
+			continue
+		}
+
+		results[serverName] = MCPTools{
+			Name:  serverName,
+			Tools: listResults.Tools,
+			Err:   nil,
+		}
+	}
+
+	return results
+}
+
+func (h *MCPHub) GetTool(ctx context.Context, serverName, toolName string) (*mcp.Tool, error) {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	// filter MCP server by serverName
+	mcpTools, exists := h.GetTools(ctx)[serverName]
+	if !exists {
+		return nil, fmt.Errorf("no connection found for server %s", serverName)
+	} else if mcpTools.Err != nil {
+		return nil, mcpTools.Err
+	}
+
+	// filter tool by toolName
+	for _, tool := range mcpTools.Tools {
+		if tool.Name == toolName {
+			return &tool, nil
+		}
+	}
+
+	return nil, fmt.Errorf("tool %s not found", toolName)
+}
+
+// InvokeTool calls a tool with the given arguments
+func (h *MCPHub) InvokeTool(ctx context.Context,
+	serverName, toolName string, arguments map[string]interface{},
+) (*mcp.CallToolResult, error) {
+	conn, err := h.GetClient(serverName)
+	if err != nil {
+		return nil, errors.Wrapf(err,
+			"get mcp client for server %s failed", serverName)
+	}
+
+	mcpTool, err := h.GetTool(ctx, serverName, toolName)
+	if err != nil {
+		return nil, errors.Wrapf(err,
+			"get mcp tool %s/%s failed", serverName, toolName)
+	}
+
+	req := mcp.CallToolRequest{}
+	req.Params.Name = mcpTool.Name
+	req.Params.Arguments = arguments
+	callToolResult, err := conn.CallTool(ctx, req)
+	if err != nil {
+		return nil, errors.Wrapf(err,
+			"call tool %s/%s failed", serverName, toolName)
+	}
+
+	return callToolResult, nil
+}
+
+// GetEinoTool returns an eino tool from the MCP server
+func (h *MCPHub) GetEinoTool(ctx context.Context, serverName, toolName string) (tool.BaseTool, error) {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	// filter MCP server by serverName
+	conn, exists := h.connections[serverName]
+	if !exists {
+		return nil, fmt.Errorf("no connection found for server %s", serverName)
+	}
+
+	if conn.Config.Disabled {
+		return nil, fmt.Errorf("server %s is disabled", serverName)
+	}
+
+	// get tools from MCP server and convert to eino tools
+	tools, err := mcpp.GetTools(ctx, &mcpp.Config{
+		Cli:          conn.Client,
+		ToolNameList: []string{toolName},
+	})
+	if err != nil || len(tools) == 0 {
+		log.Error().Err(err).
+			Str("server", serverName).Str("tool", toolName).
+			Msg("get MCP tool failed")
+		return nil, err
+	}
+
+	return tools[0], nil
+}
+
+// CloseServers closes all connected MCP servers
+func (h *MCPHub) CloseServers() error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	log.Info().Msg("Shutting down MCP servers...")
+	for name, client := range h.connections {
+		if err := client.Client.Close(); err != nil {
+			log.Error().Str("name", name).Err(err).Msg("Failed to close server")
+		} else {
+			delete(h.connections, name)
+			log.Info().Str("name", name).Msg("Server closed")
+		}
+	}
+
+	return nil
+}

internal/mcp/hub_test.go

@@ -0,0 +1,56 @@
+package mcp
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cloudwego/eino/components/tool"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetTools(t *testing.T) {
+	hub, err := NewMCPHub("./test.mcp.json")
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	err = hub.InitServers(ctx)
+	require.NoError(t, err)
+
+	tools := hub.GetTools(ctx)
+	assert.NoError(t, err)
+	assert.Equal(t, 2, len(tools))
+}
+
+func TestCallTool(t *testing.T) {
+	hub, err := NewMCPHub("./test.mcp.json")
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	err = hub.InitServers(ctx)
+	require.NoError(t, err)
+
+	result, err := hub.InvokeTool(ctx, "weather", "get_alerts", map[string]interface{}{
+		"state": "CA",
+	})
+	require.NoError(t, err)
+	t.Logf("Result: %v", result)
+}
+
+func TestCallEinoTool(t *testing.T) {
+	hub, err := NewMCPHub("./test.mcp.json")
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	err = hub.InitServers(ctx)
+	require.NoError(t, err)
+
+	einoTool, err := hub.GetEinoTool(ctx, "weather", "get_alerts")
+	require.NoError(t, err)
+	t.Logf("Tool: %v", einoTool)
+
+	tool := einoTool.(tool.InvokableTool)
+	result, err := tool.InvokableRun(ctx, `{"state": "CA"}`)
+	require.NoError(t, err)
+	t.Logf("Result: %v", result)
+}

internal/mcp/test.mcp.json

@@ -0,0 +1,27 @@
+{
+    "mcpServers": {
+      "filesystem": {
+        "command": "npx",
+        "args": [
+          "-y",
+          "@modelcontextprotocol/server-filesystem",
+          "/Users/debugtalk/Downloads/tmp"
+        ]
+      },
+      "weather": {
+        "command": "uv",
+        "args": [
+          "--directory",
+          "/Users/debugtalk/MyProjects/ByteDance/EvalTools/quickstart-resources/weather-server-python/",
+          "run",
+          "weather.py"
+        ],
+        "autoApprove": [
+          "get_forecast"
+        ],
+        "env": {
+          "ABC": "123"
+        }
+      }
+    }
+  }

internal/version/VERSION

@@ -1 +1 @@
-v5.0.0-beta-2503311454
+v5.0.0-beta-2504012111