What is Mcp Host

MCP_Host is a Go package designed to manage and simplify communication with multiple MCP servers. It facilitates connection management, tool invocation, and integration with large language models (LLMs).

Use cases

Use cases for MCP_Host include managing multiple connections to different MCP servers, executing tools to retrieve data or perform actions, and integrating with LLMs for enhanced functionality in applications.

How to use

To use MCP_Host, install it via ‘go get github.com/TIANLI0/MCP_Host’. You can connect to MCP servers using various methods such as SSE, Stdio, or in-process connections, and execute tools with simple function calls.

Key features

Key features include multi-server connection management, support for various connection methods (SSE, Stdio, in-process), built-in integration with LLMs like OpenAI, simplified tool invocation management, and a flexible notification system for handling server notifications.

Where to use

MCP_Host can be used in fields that require interaction with multiple MCP servers, such as AI applications, data processing, and real-time communication systems.

MCP_Host

MCP_Host 是一个用于管理和简化与多个 MCP (Model Context Protocol) 服务器通信的 Go 语言库。它提供了连接管理、工具调用、大语言模型集成和协议适配等功能。

TODO

• [ ] 完善文档
• [x] 支持调用Tools
• [ ] 支持资源查找
• [x] stdio到SSE协议适配器

特性

• 多服务器连接管理：统一管理多个 MCP 服务器连接
• 多种连接方式：支持 SSE、Stdio 和进程内连接方式
• 大语言模型集成：内置与 OpenAI 等 LLM 的集成，支持文本模式和函数调用模式
• 工具调用管理：简化工具调用流程，支持自动工具执行和多轮工具调用
• 灵活的通知系统：支持服务器通知的处理和转发
• 协议适配：提供 stdio 到 SSE 的协议适配器
• 流式响应：支持流式输出和实时状态通知
• 资源管理：支持 MCP 资源的列表和读取

安装

go get github.com/TIANLI0/MCP_Host

基本用法

连接 MCP 服务器

import (
    "context"
    "fmt"
    "time"
    
    "github.com/TIANLI0/MCP_Host"
)

func main() {
    host := MCP_Host.NewMCPHost()
    defer host.DisconnectAll()
    
    ctx := context.Background()
    
    // 连接到 SSE 服务器
    conn, err := host.ConnectSSE(ctx, "server1", "http://your-mcp-server-url/sse")
    if err != nil {
        panic(err)
    }
    
    fmt.Printf("已连接到: %s (版本 %s)\n", 
        conn.ServerInfo.ServerInfo.Name, 
        conn.ServerInfo.ServerInfo.Version)
}

执行工具调用

func executeTools(host *MCP_Host.MCPHost, ctx context.Context) {
    // 列出可用工具
    tools, err := host.ListTools(ctx, "server1")
    if err != nil {
        fmt.Printf("无法列出工具: %v\n", err)
        return
    }
    
    fmt.Printf("发现 %d 个可用工具\n", len(tools.Tools))
    
    // 执行工具
    result, err := host.ExecuteTool(ctx, "server1", "get_current_time", nil)
    if err != nil {
        fmt.Printf("工具执行失败: %v\n", err)
        return
    }
    
    fmt.Printf("工具执行结果: %v\n", result.Content)
}

资源管理

func manageResources(host *MCP_Host.MCPHost, ctx context.Context) {
    // 列出可用资源
    resources, err := host.ListResources(ctx, "server1")
    if err != nil {
        fmt.Printf("无法列出资源: %v\n", err)
        return
    }
    
    fmt.Printf("发现 %d 个可用资源\n", len(resources.Resources))
    
    // 读取特定资源
    if len(resources.Resources) > 0 {
        resource, err := host.ReadResource(ctx, "server1", resources.Resources[0].URI)
        if err != nil {
            fmt.Printf("读取资源失败: %v\n", err)
            return
        }
        
        fmt.Printf("资源内容: %v\n", resource.Contents)
    }
}

与大语言模型集成

MCP_Host 内置了与 LLM 的集成，支持文本模式和函数调用模式的工具使用：

基本使用

import (
    "github.com/TIANLI0/MCP_Host"
    "github.com/TIANLI0/MCP_Host/llm"
)

func useLLM(host *MCP_Host.MCPHost, ctx context.Context) {
    // 创建 OpenAI 客户端
    openaiClient, err := llm.NewOpenAIClient(
        llm.WithToken("your-api-key"),
        llm.WithOpenAIModel("gpt-4"),
        llm.WithBaseURL("https://api.openai.com/v1"),
    )
    if err != nil {
        panic(err)
    }
    
    // 创建 MCP 客户端包装
    mcpClient := llm.NewMCPClient(openaiClient, host)
    
    // 自动执行工具调用
    gen, err := mcpClient.Generate(ctx, "现在是几点？",
        llm.WithMCPWorkMode(llm.TextMode),
        llm.WithMCPAutoExecute(true),
    )
    if err != nil {
        panic(err)
    }
    
    fmt.Println(gen.Content)
}

工作模式

文本模式 (TextMode)

在文本模式下，LLM 会在响应中生成特定格式的工具调用标签：

gen, err := mcpClient.Generate(ctx, "查询天气信息",
    llm.WithMCPWorkMode(llm.TextMode),
    llm.WithMCPAutoExecute(true),
    llm.WithMCPTaskTag("MCP_HOST_TASK"),
    llm.WithMCPResultTag("MCP_HOST_RESULT"),
)

函数调用模式 (FunctionCallMode)

在函数调用模式下，使用标准的 OpenAI 函数调用格式：

gen, err := mcpClient.Generate(ctx, "查询天气信息",
    llm.WithMCPWorkMode(llm.FunctionCallMode),
    llm.WithMCPAutoExecute(true),
)

多轮工具执行

gen, err := mcpClient.Generate(ctx, "帮我规划从北京到上海的出行方案",
    llm.WithMCPWorkMode(llm.TextMode),
    llm.WithMCPAutoExecute(true),
    llm.WithMCPMaxToolExecutionRounds(5), // 最多执行5轮工具调用
)

高级功能

流式输出

// 设置流式输出回调
gen, err := mcpClient.Generate(ctx, "需要执行的任务",
    llm.WithMCPAutoExecute(true),
    llm.WithStreamingFunc(func(ctx context.Context, chunk []byte) error {
        fmt.Print(string(chunk))
        return nil
    }),
)

状态通知

gen, err := mcpClient.Generate(ctx, "复杂任务",
    llm.WithMCPAutoExecute(true),
    llm.WithStateNotifyFunc(func(ctx context.Context, state llm.MCPExecutionState) error {
        switch state.Type {
        case "tool_call":
            fmt.Printf("\n[调用工具: %s.%s]\n", state.ServerID, state.ToolName)
        case "tool_result":
            fmt.Printf("\n[工具结果]\n")
        case "execution_round":
            if round, ok := state.Data["round"].(int); ok {
                fmt.Printf("\n[开始第 %d 轮执行]\n", round)
            }
        case "process_complete":
            fmt.Printf("\n[处理完成]\n")
        }
        return nil
    }),
)

禁用特定工具

// 禁用特定工具
disabledTools := []string{
    "server1.dangerous_tool",
    "server2.slow_tool",
}

gen, err := mcpClient.Generate(ctx, "执行安全的任务",
    llm.WithMCPDisabledTools(disabledTools),
)

手动工具执行

// 不自动执行，手动控制工具调用
gen, err := mcpClient.Generate(ctx, "现在是几点",
    llm.WithMCPWorkMode(llm.TextMode),
    llm.WithMCPAutoExecute(false),
)

// 提取工具调用任务
tasks, err := mcpClient.ExtractMCPTasks(gen.Content)
if err == nil && len(tasks) > 0 {
    // 手动执行工具调用
    results, err := mcpClient.ExecuteMCPTasksWithResults(ctx, gen.Content)
    if err != nil {
        log.Printf("执行失败: %v", err)
    }
    
    for _, result := range results {
        fmt.Printf("工具 %s.%s 结果: %v\n", 
            result.Task.Server, result.Task.Tool, result.Result)
    }
}

自定义 MCP 服务器连接

除了 SSE 连接外，MCP_Host 还支持其他连接方式：

标准输入输出连接

conn, err := host.ConnectStdio(ctx, "local-server", "./mcp-server", 
    []string{"ENV=production"}, "--debug")

进程内连接

import "github.com/mark3labs/mcp-go/server"

// 创建服务器实例
server := server.NewMCPServer()
// 添加工具
server.RegisterTool("get_time", func(ctx context.Context, args map[string]any) (any, error) {
    return time.Now().String(), nil
})

// 连接
conn, err := host.ConnectInProcess(ctx, "embedded", server)

Stdio 到 SSE 适配器

MCP_Host 提供了一个适配器，可以将使用 stdio 协议的 MCP 服务器转换为 SSE 服务器：

使用适配器

import "github.com/TIANLI0/MCP_Host/adapters/stdio2sse"

// 创建适配器
adapter := stdio2sse.NewStdioToSSEAdapter(
    "python", // 命令
    []string{"mcp_server.py"}, // 参数
    stdio2sse.WithEnvironment([]string{"PATH=" + os.Getenv("PATH")}),
)

// 初始化适配器
if err := adapter.Initialize(); err != nil {
    log.Fatalf("初始化失败: %v", err)
}

// 创建 HTTP 服务器
mux := http.NewServeMux()
mux.Handle("/sse", adapter.GetSSEServer().SSEHandler())
mux.Handle("/message", adapter.GetSSEServer().MessageHandler())

// 健康检查
mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
    if err := adapter.HealthCheck(r.Context()); err != nil {
        http.Error(w, err.Error(), http.StatusServiceUnavailable)
        return
    }
    w.WriteHeader(http.StatusOK)
})

server := &http.Server{Addr: ":8080", Handler: mux}
server.ListenAndServe()

命令行工具

你也可以直接使用命令行工具来启动适配器：

go run examples/stdio2sse/main.go python mcp_server.py

# 服务器将在 :8057 端口启动
# SSE 端点: http://localhost:8057/sse
# 健康检查: http://localhost:8057/health

API 参考

MCPHost 方法

// 连接管理
func (h *MCPHost) ConnectSSE(ctx context.Context, serverID, url string) (*Connection, error)
func (h *MCPHost) ConnectStdio(ctx context.Context, serverID, command string, env []string, args ...string) (*Connection, error)
func (h *MCPHost) DisconnectServer(serverID string) error
func (h *MCPHost) DisconnectAll()

// 工具操作
func (h *MCPHost) ListTools(ctx context.Context, serverID string) (*mcp.ListToolsResult, error)
func (h *MCPHost) ExecuteTool(ctx context.Context, serverID, toolName string, args map[string]any) (*mcp.CallToolResult, error)

// 资源操作
func (h *MCPHost) ListResources(ctx context.Context, serverID string) (*mcp.ListResourcesResult, error)
func (h *MCPHost) ReadResource(ctx context.Context, serverID, uri string) (*mcp.ReadResourceResult, error)

// 通知处理
func (h *MCPHost) SetNotificationHandler(serverID string, handler func(mcp.JSONRPCNotification)) error
func (h *MCPHost) SetGlobalNotificationHandler(handler func(serverID string, notification mcp.JSONRPCNotification))

MCPClient 选项

// 工作模式
llm.WithMCPWorkMode(llm.TextMode)           // 文本模式
llm.WithMCPWorkMode(llm.FunctionCallMode)   // 函数调用模式

// 执行控制
llm.WithMCPAutoExecute(true)                // 自动执行工具调用
llm.WithMCPMaxToolExecutionRounds(5)        // 最大执行轮次
llm.WithMCPDisabledTools([]string{"server.tool"}) // 禁用工具

// 流式和通知
llm.WithStreamingFunc(func(ctx context.Context, chunk []byte) error { ... })
llm.WithStateNotifyFunc(func(ctx context.Context, state llm.MCPExecutionState) error { ... })

// 标签自定义
llm.WithMCPTaskTag("CUSTOM_TASK")          // 自定义任务标签
llm.WithMCPResultTag("CUSTOM_RESULT")      // 自定义结果标签
llm.WithMCPPrompt("custom prompt...")       // 自定义提示词

示例

完整示例可以在 examples 目录中找到：

• examples/simple/main.go - 基本连接和工具调用示例
• examples/chat_simple/main.go - 与 LLM 集成的基本示例
• examples/auto_exec/main.go - 自动执行工具的高级示例，包含状态通知
• examples/stdio2sse/main.go - Stdio 到 SSE 适配器示例

许可证

MIT License