Scaled MCP Server

Scaled MCP Server

一个支持负载均衡部署、可水平扩展的 MCP（Message Context Protocol）服务实现。

概述

Scaled MCP Server 是一个 Go 库，实现了 MCP 2025-03 规范，并支持水平扩展。它旨在嵌入到你的应用程序中，并提供灵活的配置选项。

功能

• HTTP 传输：灵活的 HTTP 传输，包含主 /mcp 端点、可选的 SSE 端点以及能力协商
• 会话管理：支持使用 Redis 或内存选项进行分布式会话管理
• Actor 系统：采用基于 Actor 的架构来处理会话和消息路由
• 水平扩展：支持跨多个节点进行负载均衡部署

安装

go get github.com/traego/scaled-mcp@latest

**注意：**此库需要 Go 1.24 或更高版本。

使用方法

使用静态工具创建基本服务

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"syscall"

	"github.com/traego/scaled-mcp/pkg/config"
	"github.com/traego/scaled-mcp/pkg/resources"
	"github.com/traego/scaled-mcp/pkg/server"
)

func main() {
	// Configure logging
	logHandler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})
	slog.SetDefault(slog.New(logHandler))

	// Create a server with default configuration
	cfg := config.DefaultConfig()
	
	// Use in-memory session store for simplicity
	cfg.Session.UseInMemory = true
	
	// Create a static tool registry
	registry := resources.NewStaticToolRegistry()
	
	// Define and register a simple calculator tool
	calculatorTool := resources.NewTool("calculator").
		WithDescription("Performs basic arithmetic operations").
		WithInputs([]resources.ToolInput{
			{
				Name:        "operation",
				Type:        "string",
				Description: "Operation to perform (add, subtract, multiply, divide)",
				Required:    true,
			},
			{
				Name:        "a",
				Type:        "number",
				Description: "First operand",
				Required:    true,
			},
			{
				Name:        "b",
				Type:        "number",
				Description: "Second operand",
				Required:    true,
			},
		}).
		Build()
	
	// Register the tool with the registry
	registry.RegisterTool(calculatorTool)
	
	// Define a prompt for the server
	prompt := "You are a helpful AI assistant that can perform calculations using the calculator tool."
	
	// Create the server with the tool registry and prompt
	srv, err := server.NewMcpServer(cfg,
		server.WithToolRegistry(registry),
		server.WithServerInfo("Example MCP Server", "1.0.0"),
		server.WithPrompt(prompt),
	)
	if err != nil {
		slog.Error("Failed to create server", "error", err)
		os.Exit(1)
	}
	
	// Set up the tool handler
	registry.SetToolHandler("calculator", func(ctx context.Context, params map[string]interface{}) (interface{}, error) {
		// Extract parameters
		operation, ok := params["operation"].(string)
		if !ok {
			return nil, fmt.Errorf("%w: operation must be a string", resources.ErrInvalidParams)
		}
		
		a, ok := params["a"].(float64)
		if !ok {
			return nil, fmt.Errorf("%w: a must be a number", resources.ErrInvalidParams)
		}
		
		b, ok := params["b"].(float64)
		if !ok {
			return nil, fmt.Errorf("%w: b must be a number", resources.ErrInvalidParams)
		}
		
		// Perform the calculation
		var result float64
		switch operation {
		case "add":
			result = a + b
		case "subtract":
			result = a - b
		case "multiply":
			result = a * b
		case "divide":
			if b == 0 {
				return nil, fmt.Errorf("%w: division by zero", resources.ErrInvalidParams)
			}
			result = a / b
		default:
			return nil, fmt.Errorf("%w: unknown operation %s", resources.ErrInvalidParams, operation)
		}
		
		return map[string]interface{}{
			"result": result,
		}, nil
	})
	
	// Start the server in a goroutine
	go func() {
		if err := srv.Start(context.Background()); err != nil {
			slog.Error("Failed to start server", "error", err)
			os.Exit(1)
		}
	}()
	
	slog.Info("Server started", "host", cfg.HTTP.Host, "port", cfg.HTTP.Port)
	
	// Wait for termination signal
	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
	<-sig
	
	// Shutdown the server
	slog.Info("Shutting down server...")
	shutdownCtx, cancel := context.WithTimeout(context.Background(), cfg.HTTP.ShutdownTimeout)
	defer cancel()
	
	if err := srv.Stop(shutdownCtx); err != nil {
		slog.Error("Failed to stop server", "error", err)
	}
	
	slog.Info("Server stopped")
}

### Using an External HTTP Server

You can use your own HTTP server with the MCP transport:

```go
package main

import (
	"context"
	"log"
	"net/http"

	"github.com/go-chi/chi/v5"
	"github.com/go-chi/chi/v5/middleware"
	"github.com/go-chi/cors"
	"github.com/traego/scaled-mcp/pkg/config"
	"github.com/traego/scaled-mcp/pkg/server"
	"github.com/traego/scaled-mcp/pkg/transport"
)

func main() {
	// Create a server with default configuration
	cfg := config.DefaultConfig()
	cfg.Session.UseInMemory = true
	
	// Create the MCP server but don't start the HTTP server
	srv, err := server.NewServer(cfg)
	if err != nil {
		log.Fatalf("Failed to create server: %v", err)
	}
	
	// Create a custom router
	r := chi.NewRouter()
	
	// Add middleware
	r.Use(middleware.RequestID)
	r.Use(middleware.RealIP)
	r.Use(middleware.Logger)
	r.Use(middleware.Recoverer)
	
	// Add CORS middleware - important when using an external server
	r.Use(cors.Handler(cors.Options{
		AllowedOrigins:   []string{"*"},
		AllowedMethods:   []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},
		AllowedHeaders:   []string{"Accept", "Authorization", "Content-Type", "X-CSRF-Token"},
		ExposedHeaders:   []string{"Link"},
		AllowCredentials: true,
		MaxAge:           300,
	}))
	
	// Add your custom routes
	r.Get("/", func(w http.ResponseWriter, r *http.Request) {
		w.Write([]byte("Welcome to the MCP server!"))
	})
	
	// Create the HTTP transport with the custom router
	httpTransport := transport.NewHTTPTransport(
		cfg,
		srv.GetActorSystem(),
		srv.GetSessionManager(),
		transport.WithExternalRouter(r),
	)
	
	// Start the MCP server without HTTP
	if err := srv.Start(context.Background()); err != nil {
		log.Fatalf("Failed to start server: %v", err)
	}
	
	// Start the HTTP transport
	if err := httpTransport.Start(); err != nil {
		log.Fatalf("Failed to start HTTP transport: %v", err)
	}
	
	// Start your HTTP server
	log.Printf("Starting HTTP server on %s:%d", cfg.HTTP.Host, cfg.HTTP.Port)
	if err := http.ListenAndServe(":8080", r); err != nil {
		log.Fatalf("HTTP server error: %v", err)
	}
}

## Dynamic Tool Registry Example

This library supports both static and dynamic tool registries. Here's an client_example of using a dynamic tool registry:

```go
// Create a custom tool provider
toolProvider := NewExampleToolProvider()

// Create a dynamic tool registry with the provider
registry := resources.NewDynamicToolRegistry(toolProvider)

// Create server with the dynamic tool registry
cfg := config.DefaultConfig()
mcpServer, err := server.NewMcpServer(cfg,
    server.WithToolRegistry(registry),
)

工具定义

此库提供三种定义工具输入的方式：

1. 使用基于结构体的反射（推荐）

// Define input struct with mcp tags
type CalculatorInput struct {
    Operation string  `mcp:"operation,The operation to perform (add subtract multiply divide),required"`
    A         float64 `mcp:"a,First operand,required"`
    B         float64 `mcp:"b,Second operand,required"`
}

// Type-safe handler function
func calculatorHandler(ctx context.Context, input *CalculatorInput) (interface{}, error) {
    // Direct access to typed fields: input.Operation, input.A, input.B
    switch input.Operation {
    case "add":
        return map[string]interface{}{"result": input.A + input.B}, nil
    // ...
    }
}

// Register with automatic schema generation (method 1)
calculatorType := reflect.TypeOf(CalculatorInput{})
registry.RegisterStructToolWithHandler("calculator", "Performs arithmetic operations", calculatorType, calculatorHandler)

// Or use the convenience function (method 2)
resources.RegisterStructTool(registry, "calculator", "Performs arithmetic operations", calculatorHandler)

### Generics-Based Tool Registration (New in June 2025)

For even better type safety, you can use the new generics-based approach with two type parameters for input and output:

```go
type CalculatorInput struct {
    Operation string  `mcp:"operation,The operation to perform,required"`
    A         float64 `mcp:"a,First operand,required"`
    B         float64 `mcp:"b,Second operand,required"`
}

type CalculatorOutput struct {
    Result    float64 `mcp:"result,The calculation result,required"`
    Operation string  `mcp:"operation,The operation performed,required"`
}

func calculatorHandler(ctx context.Context, input *CalculatorInput) (*CalculatorOutput, error) {
    result := input.A + input.B
    return &CalculatorOutput{
        Result:    result,
        Operation: input.Operation,
    }, nil
}

// Register with compile-time type safety for both input and output
err := resources.RegisterStructToolWithTypes(registry, "calculator", "Performs arithmetic", calculatorHandler)

此方法提供：

• 编译时类型安全，适用于输入和输出
• 自动生成架构，适用于输入和输出类型
• 注册时零反射（改用泛型）
• 更好的 IDE 支持，具备完整的类型检查

#### 2. Using WithInputs

```go
weatherTool := resources.NewTool("weather").
    WithDescription("Get weather information for a location").
    WithInputs([]resources.ToolInput{
        {
            Name:        "location",
            Type:        "string",
            Description: "The location to get weather for",
            Required:    true,
        },
        {
            Name:        "units",
            Type:        "string",
            Description: "Temperature units (celsius or fahrenheit)",
            Default:     "celsius",
        },
    }).
    Build()

3. 使用单独的参数方法

calculatorTool := resources.NewTool("calculator").
    WithDescription("Performs basic arithmetic operations").
    WithString("operation").
    Required().
    Description("Operation to perform (add, subtract, multiply, divide)").
    Add().
    WithNumber("a").
    Required().
    Description("First operand").
    Add().
    WithNumber("b").
    Required().
    Description("Second operand").
    Add().
    Build()

结构体标签格式

mcp 结构体标签遵循以下格式：mcp:"name,description,required,default=value"

• name：架构中的字段名称（默认为小写字段名称）
• description：字段的可读描述
• required：将字段标记为必填
• default=value：为可选字段设置默认值
• -：跳过字段（不包含在架构中）

支持的 Go 类型：

• string → "string"
• int、int64 等 → "integer"
• float64、float32 → "number"
• bool → "boolean"
• []T、[N]T → "array"
• struct、map → "object"
• *T → 与 T 相同（指针类型）

重要说明

CORS 配置

通过 MCP 传输使用外部 HTTP 服务时，需要在路由器上配置 CORS 设置。如上例所示，使用外部路由器时，MCP 传输不会应用 CORS 设置。

会话管理

对于生产部署，建议使用 Redis 进行会话管理，以支持水平扩展。内存会话存储仅应用于开发或测试。

待办事项

• 授权示例 + Auth 上下文贯穿流程
• Metrics 端点（prometheus），涵盖 Actor 启动 / 停止、平均会话时长等
• Session Actor 钩子
• MCP 规范
  • 列表变更通知
  • 采样
  • 根目录
  • 补全
  • 日志记录
• A2A 规范
  • 此处添加了详细信息 https://github.com/Traego/scaled-mcp/wiki/A2A-Support
• 添加流水线支持插件
• 已接入并测试 K8S 集群
• 搜索支持（https://github.com/modelcontextprotocol/modelcontextprotocol/pull/322）
• 全面测试所有功能（资源、提示词等）
• 测试覆盖率达到 80%
• 稍微清理一下新的服务 API，让启动更简单
• 改进默认设置
• 升级到 goakt 的 vNext，并替换自定义消息调度（原先是用于规避一个 bug）

参与贡献

欢迎贡献！请随时提交 Pull Request。

1. Fork 此仓库
2. 创建功能分支（git checkout -b feature/amazing-feature）
3. 提交更改（git commit -m 'Add some amazing feature'）
4. 推送到分支（git push origin feature/amazing-feature）
5. 打开 Pull Request

开发

测试

运行测试：

go test ./...

运行测试并生成覆盖率报告：

go test -race -coverprofile=coverage.txt -covermode=atomic ./...

在浏览器中查看覆盖率报告：

go tool cover -html=coverage.txt

代码覆盖率

本项目使用 Codecov 生成代码覆盖率报告。CI 运行期间会自动生成并上传覆盖率报告。

如需查看覆盖率仪表板，请访问 codecov.io/gh/traego/scaled-mcp。

许可证

本项目采用 MIT 许可证。

文档

请参阅 GoDoc，获取详细的 API 文档。