From 0ff02020e71a5170328f359d158ff770d44c01c3 Mon Sep 17 00:00:00 2001 From: cwz_eikoh Date: Mon, 12 Jan 2026 10:46:06 +0800 Subject: [PATCH] Automated cherry pick of #24025: feature(llm): add mcp-agent (#24026) * feature(llm): add mcp-agent * feature(llm): support openai in llm_client * feature(llm): chat-stream demo --- cmd/climc/shell/llm/mcp_agent.go | 71 +++ pkg/apis/llm/mcp_agent.go | 123 +++++ pkg/apis/llm/ollama_const.go | 1 + pkg/llm/drivers/llm_client/doc.go | 1 + pkg/llm/drivers/llm_client/ollama.go | 471 ++++++++++++++++++++ pkg/llm/drivers/llm_client/openai.go | 517 ++++++++++++++++++++++ pkg/llm/drivers/llm_container/ollama.go | 52 +++ pkg/llm/models/llm.go | 4 + pkg/llm/models/llm_client_driver.go | 81 ++++ pkg/llm/models/llm_container_driver.go | 11 +- pkg/llm/models/mcp_agent.go | 438 ++++++++++++++++++ pkg/llm/options/option.go | 4 + pkg/llm/service/handler.go | 1 + pkg/llm/service/service.go | 1 + pkg/llm/utils/mcp_client.go | 437 ++++++++++++++++++ pkg/mcclient/modules/llm/mod_mcp_agent.go | 24 + pkg/mcclient/options/llm/mcp_agent.go | 167 +++++++ 17 files changed, 2398 insertions(+), 6 deletions(-) create mode 100644 cmd/climc/shell/llm/mcp_agent.go create mode 100644 pkg/apis/llm/mcp_agent.go create mode 100644 pkg/llm/drivers/llm_client/doc.go create mode 100644 pkg/llm/drivers/llm_client/ollama.go create mode 100644 pkg/llm/drivers/llm_client/openai.go create mode 100644 pkg/llm/models/llm_client_driver.go create mode 100644 pkg/llm/models/mcp_agent.go create mode 100644 pkg/llm/utils/mcp_client.go create mode 100644 pkg/mcclient/modules/llm/mod_mcp_agent.go create mode 100644 pkg/mcclient/options/llm/mcp_agent.go diff --git a/cmd/climc/shell/llm/mcp_agent.go b/cmd/climc/shell/llm/mcp_agent.go new file mode 100644 index 0000000000..9fc57da40a --- /dev/null +++ b/cmd/climc/shell/llm/mcp_agent.go @@ -0,0 +1,71 @@ +package llm + +import ( + "fmt" + "io" + "net/url" + + "yunion.io/x/onecloud/cmd/climc/shell" + "yunion.io/x/onecloud/pkg/mcclient" + modules "yunion.io/x/onecloud/pkg/mcclient/modules/llm" + base_options "yunion.io/x/onecloud/pkg/mcclient/options" + options "yunion.io/x/onecloud/pkg/mcclient/options/llm" +) + +func init() { + cmd := shell.NewResourceCmd(&modules.MCPAgent) + cmd.List(new(options.MCPAgentListOptions)) + cmd.Show(new(options.MCPAgentShowOptions)) + cmd.Create(new(options.MCPAgentCreateOptions)) + cmd.Update(new(options.MCPAgentUpdateOptions)) + cmd.Delete(new(options.MCPAgentDeleteOptions)) + cmd.Perform("public", &base_options.BasePublicOptions{}) + cmd.Perform("private", &base_options.BaseIdOptions{}) + cmd.Get("mcp-tools", new(options.MCPAgentIdOptions)) + cmd.Get("tool-request", new(options.MCPAgentToolRequestOptions)) + // cmd.Get("chat-test", new(options.MCPAgentChatTestOptions)) + cmd.Get("request", new(options.MCPAgentMCPAgentRequestOptions)) + shell.R(&options.MCPAgentChatTestOptions{}, "mcp-agent-chat", "Chat with MCP Agent (Stream)", func(s *mcclient.ClientSession, args *options.MCPAgentChatTestOptions) error { + id, err := modules.MCPAgent.GetId(s, args.ID, nil) + if err != nil { + return err + } + + path := fmt.Sprintf("/mcp_agents/%s/chat-stream?message=%s", id, url.QueryEscape(args.Message)) + + resp, err := s.RawVersionRequest( + modules.MCPAgent.ServiceType(), + modules.MCPAgent.EndpointType(), + "GET", + path, + nil, + nil, + ) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + // Read error body + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("Error: %s %s", resp.Status, string(body)) + } + + buffer := make([]byte, 1024) + for { + n, err := resp.Body.Read(buffer) + if n > 0 { + fmt.Print(string(buffer[:n])) + } + if err != nil { + if err == io.EOF { + break + } + return err + } + } + fmt.Println() + return nil + }) +} diff --git a/pkg/apis/llm/mcp_agent.go b/pkg/apis/llm/mcp_agent.go new file mode 100644 index 0000000000..3714cc6949 --- /dev/null +++ b/pkg/apis/llm/mcp_agent.go @@ -0,0 +1,123 @@ +package llm + +import ( + "yunion.io/x/pkg/util/sets" + + "yunion.io/x/onecloud/pkg/apis" +) + +// LLMClientType 定义 LLM 驱动类型 +type LLMClientType string + +const ( + LLM_CLIENT_OLLAMA LLMClientType = "ollama" + LLM_CLIENT_OPENAI LLMClientType = "openai" + + MCP_AGENT_SYSTEM_PROMPT = `你是一个 Cloudpods 云平台管理助手。你可以使用提供的工具来帮助用户管理云资源。 + +## 你的能力 +- 查询云平台资源（虚拟机、镜像、网络、存储、区域等） +- 管理虚拟机（创建、启动、停止、重启、删除、重置密码） +- 获取虚拟机监控信息和实时统计数据 + +## 工作流程 +1. 理解用户的需求 +2. 选择合适的工具来完成任务 +3. 分析工具返回的结果 +4. 如果需要更多信息，继续调用其他工具 +5. 最后用自然语言总结结果给用户 + +## 注意事项 +- 认证信息已由系统自动处理，调用工具时无需提供认证参数 +- 如果工具调用失败，尝试分析错误原因并告知用户 +- 回复时使用中文，语言简洁明了 +` +) + +var ( + LLM_CLIENT_TYPES = sets.NewString( + string(LLM_CLIENT_OLLAMA), + string(LLM_CLIENT_OPENAI), + ) +) + +// IsLLMClientType 检查给定的字符串是否是有效的 LLM 驱动类型 +func IsLLMClientType(t string) bool { + return LLM_CLIENT_TYPES.Has(t) +} + +// MCP Agent 配置相关的 API 定义 +type MCPAgentListInput struct { + apis.SharableVirtualResourceListInput + + LLMDriver string `json:"llm_driver"` +} + +type MCPAgentCreateInput struct { + apis.SharableVirtualResourceCreateInput + + LLMId string `json:"llm_id" help:"LLM 实例 ID，如果提供则自动获取 llm_url"` + LLMUrl string `json:"llm_url" help:"后端大模型的 base 请求地址"` + LLMDriver string `json:"llm_driver" help:"使用的大模型驱动，可以是 ollama 或 openai"` + Model string `json:"model" help:"使用的模型名称"` + ApiKey string `json:"api_key" help:"在 llm_driver 中需要用到的认证"` + McpServer string `json:"mcp_server" help:"mcp 服务器的后端地址"` +} + +type MCPAgentUpdateInput struct { + apis.SharableVirtualResourceCreateInput + + LLMId *string `json:"llm_id,omitempty" help:"LLM 实例 ID，如果提供则自动获取 llm_url"` + LLMUrl *string `json:"llm_url,omitempty" help:"后端大模型的 base 请求地址"` + LLMDriver *string `json:"llm_driver,omitempty" help:"使用的大模型驱动，可以是 ollama 或 openai"` + Model *string `json:"model,omitempty" help:"使用的模型名称"` + ApiKey *string `json:"api_key,omitempty" help:"在 llm_driver 中需要用到的认证"` + McpServer *string `json:"mcp_server,omitempty" help:"mcp 服务器的后端地址"` +} + +type MCPAgentDetails struct { + apis.SharableVirtualResourceDetails + + LLMUrl string `json:"llm_url"` + LLMDriver string `json:"llm_driver"` + Model string `json:"model"` + ApiKey string `json:"api_key"` + McpServer string `json:"mcp_server"` +} + +type LLMToolRequestInput struct { + ToolName string `json:"tool_name"` + Arguments map[string]interface{} `json:"arguments"` +} + +type LLMChatTestInput struct { + Message string `json:"message" help:"test message to send to LLM"` +} + +type LLMMCPAgentRequestInput struct { + Query string `json:"query" help:"query to send to MCP agent"` +} + +// MCPAgentResponse 表示 Agent 响应 +type MCPAgentResponse struct { + // Success 是否成功 + Success bool `json:"success"` + // Answer 自然语言回答 + Answer string `json:"answer"` + // Error 错误信息 + Error string `json:"error,omitempty"` + // ToolCalls 工具调用记录 + ToolCalls []MCPAgentToolCallRecord `json:"tool_calls,omitempty"` +} + +// MCPAgentToolCallRecord 记录工具调用 +type MCPAgentToolCallRecord struct { + ToolName string `json:"tool_name"` + Arguments map[string]interface{} `json:"arguments"` + Result string `json:"result"` +} + +const ( + // MCPAgentMaxIterations 最大迭代次数，防止无限循环 + MCPAgentMaxIterations = 10 +) diff --git a/pkg/apis/llm/ollama_const.go b/pkg/apis/llm/ollama_const.go index ba9af4a1db..a69f610f1d 100644 --- a/pkg/apis/llm/ollama_const.go +++ b/pkg/apis/llm/ollama_const.go @@ -8,6 +8,7 @@ const ( LLM_OLLAMA_CREATE_ACTION = "create" LLM_OLLAMA_EXPORT_ENV_KEY = "OLLAMA_HOST" LLM_OLLAMA_EXPORT_ENV_VALUE = "0.0.0.0:11434" + LLM_OLLAMA_DEFAULT_PORT = 11434 ) const ( diff --git a/pkg/llm/drivers/llm_client/doc.go b/pkg/llm/drivers/llm_client/doc.go new file mode 100644 index 0000000000..97899e8eeb --- /dev/null +++ b/pkg/llm/drivers/llm_client/doc.go @@ -0,0 +1 @@ +package llm_client diff --git a/pkg/llm/drivers/llm_client/ollama.go b/pkg/llm/drivers/llm_client/ollama.go new file mode 100644 index 0000000000..2b05d9e6c5 --- /dev/null +++ b/pkg/llm/drivers/llm_client/ollama.go @@ -0,0 +1,471 @@ +package llm_client + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/mark3labs/mcp-go/mcp" + "yunion.io/x/pkg/errors" + + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/llm/models" +) + +func init() { + models.RegisterLLMClientDriver(newOllama()) +} + +type ollama struct{} + +func newOllama() models.ILLMClient { + return new(ollama) +} + +func (o *ollama) GetType() api.LLMClientType { + return api.LLM_CLIENT_OLLAMA +} + +func convertMessages(messages interface{}) ([]OllamaChatMessage, error) { + // 转换 messages + var ollamaMessages []OllamaChatMessage + if msgs, ok := messages.([]OllamaChatMessage); ok { + ollamaMessages = msgs + } else if msgs, ok := messages.([]models.ILLMChatMessage); ok { + ollamaMessages = make([]OllamaChatMessage, len(msgs)) + for i, msg := range msgs { + ollamaMessages[i] = OllamaChatMessage{ + Role: msg.GetRole(), + Content: msg.GetContent(), + } + // 转换工具调用 + if toolCalls := msg.GetToolCalls(); len(toolCalls) > 0 { + ollamaMessages[i].ToolCalls = make([]OllamaToolCall, len(toolCalls)) + for j, tc := range toolCalls { + fc := tc.GetFunction() + ollamaMessages[i].ToolCalls[j] = OllamaToolCall{ + Function: OllamaFunctionCall{ + Name: fc.GetName(), + Arguments: fc.GetArguments(), + }, + } + } + } + } + } else if msgs, ok := messages.([]interface{}); ok { + ollamaMessages = make([]OllamaChatMessage, 0, len(msgs)) + for _, msg := range msgs { + if m, ok := msg.(OllamaChatMessage); ok { + ollamaMessages = append(ollamaMessages, m) + } else if m, ok := msg.(models.ILLMChatMessage); ok { + ollamaMessages = append(ollamaMessages, OllamaChatMessage{ + Role: m.GetRole(), + Content: m.GetContent(), + }) + } + } + } else { + return nil, errors.Error("invalid messages type, expected []OllamaChatMessage or []ILLMChatMessage") + } + + return ollamaMessages, nil +} + +func convertTool(tools interface{}) ([]OllamaTool, error) { + // 转换 tools + var ollamaTools []OllamaTool + if ts, ok := tools.([]OllamaTool); ok { + ollamaTools = ts + } else if ts, ok := tools.([]models.ILLMTool); ok { + ollamaTools = make([]OllamaTool, len(ts)) + for i, t := range ts { + tf := t.GetFunction() + ollamaTools[i] = OllamaTool{ + Type: t.GetType(), + Function: OllamaToolFunction{ + Name: tf.GetName(), + Description: tf.GetDescription(), + Parameters: tf.GetParameters(), + }, + } + } + } else if ts, ok := tools.([]interface{}); ok && ts != nil { + ollamaTools = make([]OllamaTool, 0, len(ts)) + for _, tool := range ts { + if t, ok := tool.(OllamaTool); ok { + ollamaTools = append(ollamaTools, t) + } else if t, ok := tool.(models.ILLMTool); ok { + tf := t.GetFunction() + ollamaTools = append(ollamaTools, OllamaTool{ + Type: t.GetType(), + Function: OllamaToolFunction{ + Name: tf.GetName(), + Description: tf.GetDescription(), + Parameters: tf.GetParameters(), + }, + }) + } + } + } else if tools == nil { + ollamaTools = nil + } else { + return nil, errors.Error("invalid tools type, expected []OllamaTool or []ILLMTool or nil") + } + + return ollamaTools, nil +} + +func initRequestClient(ctx context.Context, endpoint, model string, stream bool, messages []OllamaChatMessage, tools []OllamaTool) (*http.Request, *http.Client, error) { + req := OllamaChatRequest{ + Model: model, + Messages: messages, + Tools: tools, + Stream: stream, + } + + reqBody, err := json.Marshal(req) + if err != nil { + return nil, nil, errors.Wrap(err, "marshal request") + } + + // 规范化 endpoint，确保以 / 结尾 + endpoint = strings.TrimSuffix(endpoint, "/") + + baseURL, err := url.Parse(endpoint) + if err != nil { + return nil, nil, errors.Wrapf(err, "invalid endpoint URL %s", endpoint) + } + + // 构建完整的 URL + apiURL := baseURL.JoinPath("/api/chat") + httpReq, err := http.NewRequestWithContext(ctx, "POST", apiURL.String(), bytes.NewReader(reqBody)) + if err != nil { + return nil, nil, errors.Wrap(err, "create request") + } + httpReq.Header.Set("Content-Type", "application/json") + httpReq.Header.Set("Accept", "application/json") + + client := &http.Client{ + Timeout: 300 * time.Second, + Transport: &http.Transport{ + MaxIdleConns: 100, + MaxIdleConnsPerHost: 10, + IdleConnTimeout: 90 * time.Second, + }, + } + + return httpReq, client, nil +} + +func (o *ollama) Chat(ctx context.Context, mcpAgent *models.SMCPAgent, messages interface{}, tools interface{}) (models.ILLMChatResponse, error) { + ollamaMessages, err := convertMessages(messages) + if err != nil { + return nil, err + } + + ollamaTools, err := convertTool(tools) + if err != nil { + return nil, err + } + + httpReq, client, err := initRequestClient(ctx, mcpAgent.LLMUrl, mcpAgent.Model, false, ollamaMessages, ollamaTools) + + // 调用底层方法 + return o.doChatRequest(ctx, httpReq, client) +} + +// doChatRequest 执行聊天请求 +func (o *ollama) doChatRequest(ctx context.Context, httpReq *http.Request, client *http.Client) (*OllamaChatResponse, error) { + resp, err := client.Do(httpReq) + if err != nil { + return nil, errors.Wrap(err, "do request") + } + defer resp.Body.Close() + + // 读取响应体以便错误处理 + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, errors.Wrap(err, "read response body") + } + + if resp.StatusCode != http.StatusOK { + return nil, errors.Errorf("unexpected status code %d: %s", resp.StatusCode, string(body)) + } + + var chatResp OllamaChatResponse + if err := json.Unmarshal(body, &chatResp); err != nil { + return nil, errors.Wrapf(err, "decode response: %s", string(body)) + } + + return &chatResp, nil +} + +func (o *ollama) NewUserMessage(content string) models.ILLMChatMessage { + return &OllamaChatMessage{ + Role: "user", + Content: content, + } +} + +func (o *ollama) NewAssistantMessageWithToolCalls(toolCalls []models.ILLMToolCall) models.ILLMChatMessage { + // to ollama tool calls + ollamaToolCalls := make([]OllamaToolCall, len(toolCalls)) + + for i, tc := range toolCalls { + if otc, ok := tc.(*OllamaToolCall); ok { + ollamaToolCalls[i] = *otc + } else { + fc := tc.GetFunction() + ollamaToolCalls[i] = OllamaToolCall{ + Function: OllamaFunctionCall{ + Name: fc.GetName(), + Arguments: fc.GetArguments(), + }, + } + } + } + + return &OllamaChatMessage{ + Role: "assistant", + Content: "", + ToolCalls: ollamaToolCalls, + } +} + +func (o *ollama) NewToolMessage(toolId string, toolName string, content string) models.ILLMChatMessage { + return &OllamaChatMessage{ + Role: "tool", + Content: fmt.Sprintf("[%s] %s", toolName, content), + } +} + +func (o *ollama) NewSystemMessage(content string) models.ILLMChatMessage { + return &OllamaChatMessage{ + Role: "system", + Content: content, + } +} + +func (o *ollama) ConvertMCPTools(mcpTools []mcp.Tool) []models.ILLMTool { + tools := make([]models.ILLMTool, len(mcpTools)) + for i, t := range mcpTools { + var params map[string]interface{} + if t.RawInputSchema != nil { + _ = json.Unmarshal(t.RawInputSchema, ¶ms) + } else { + schemaBytes, _ := json.Marshal(t.InputSchema) + _ = json.Unmarshal(schemaBytes, ¶ms) + } + tools[i] = &OllamaTool{ + Type: "function", + Function: OllamaToolFunction{ + Name: t.Name, + Description: t.Description, + Parameters: params, + }, + } + } + return tools +} + +// OllamaChatMessage 表示聊天消息 +// 实现 ILLMChatMessage 接口 +type OllamaChatMessage struct { + Role string `json:"role"` + Content string `json:"content"` + ToolCalls []OllamaToolCall `json:"tool_calls,omitempty"` +} + +// GetRole 实现 ILLMChatMessage 接口 +func (m OllamaChatMessage) GetRole() string { + return m.Role +} + +// GetContent 实现 ILLMChatMessage 接口 +func (m OllamaChatMessage) GetContent() string { + return m.Content +} + +// GetToolCalls 实现 ILLMChatMessage 接口 +func (m OllamaChatMessage) GetToolCalls() []models.ILLMToolCall { + if len(m.ToolCalls) == 0 { + return nil + } + toolCalls := make([]models.ILLMToolCall, len(m.ToolCalls)) + for i := range m.ToolCalls { + // 创建副本以避免引用问题 + tc := m.ToolCalls[i] + toolCalls[i] = &tc + } + return toolCalls +} + +// OllamaToolCall 表示工具调用 +// 实现 ILLMToolCall 接口 +type OllamaToolCall struct { + Function OllamaFunctionCall `json:"function"` +} + +// GetFunction 实现 ILLMToolCall 接口 +func (tc *OllamaToolCall) GetFunction() models.ILLMFunctionCall { + return &tc.Function +} + +// GetId 实现 ILLMToolCall 接口 +func (tc *OllamaToolCall) GetId() string { + return "" +} + +// OllamaFunctionCall 表示函数调用详情 +// 实现 ILLMFunctionCall 接口 +type OllamaFunctionCall struct { + Name string `json:"name"` + Arguments map[string]interface{} `json:"arguments"` +} + +// GetName 实现 ILLMFunctionCall 接口 +func (fc *OllamaFunctionCall) GetName() string { + return fc.Name +} + +// GetArguments 实现 ILLMFunctionCall 接口 +func (fc *OllamaFunctionCall) GetArguments() map[string]interface{} { + return fc.Arguments +} + +// OllamaTool 表示工具定义 +// 实现 ILLMTool 接口 +type OllamaTool struct { + Type string `json:"type"` + Function OllamaToolFunction `json:"function"` +} + +// GetType 实现 ILLMTool 接口 +func (t OllamaTool) GetType() string { + return t.Type +} + +// GetFunction 实现 ILLMTool 接口 +func (t OllamaTool) GetFunction() models.ILLMToolFunction { + return &t.Function +} + +// OllamaToolFunction 表示工具函数定义 +// 实现 ILLMToolFunction 接口 +type OllamaToolFunction struct { + Name string `json:"name"` + Description string `json:"description"` + Parameters map[string]interface{} `json:"parameters"` +} + +// GetName 实现 ILLMToolFunction 接口 +func (tf *OllamaToolFunction) GetName() string { + return tf.Name +} + +// GetDescription 实现 ILLMToolFunction 接口 +func (tf *OllamaToolFunction) GetDescription() string { + return tf.Description +} + +// GetParameters 实现 ILLMToolFunction 接口 +func (tf *OllamaToolFunction) GetParameters() map[string]interface{} { + return tf.Parameters +} + +// OllamaChatRequest 表示聊天请求 +type OllamaChatRequest struct { + Model string `json:"model"` + Messages []OllamaChatMessage `json:"messages"` + Tools []OllamaTool `json:"tools,omitempty"` + Stream bool `json:"stream"` +} + +// OllamaChatResponse 表示聊天响应 +type OllamaChatResponse struct { + Model string `json:"model"` + CreatedAt string `json:"created_at"` + Message OllamaChatMessage `json:"message"` + Done bool `json:"done"` + DoneReason string `json:"done_reason,omitempty"` +} + +// GetContent 获取响应内容 +func (r *OllamaChatResponse) GetContent() string { + return r.Message.Content +} + +// HasToolCalls 检查响应是否包含工具调用 +func (r *OllamaChatResponse) HasToolCalls() bool { + return len(r.Message.ToolCalls) > 0 +} + +// GetToolCalls 获取工具调用列表 +func (r *OllamaChatResponse) GetToolCalls() []models.ILLMToolCall { + if len(r.Message.ToolCalls) == 0 { + return nil + } + toolCalls := make([]models.ILLMToolCall, len(r.Message.ToolCalls)) + for i := range r.Message.ToolCalls { + toolCalls[i] = &r.Message.ToolCalls[i] + } + return toolCalls +} + +func (o *ollama) ChatStream(ctx context.Context, mcpAgent *models.SMCPAgent, messages interface{}, tools interface{}, onChunk func(models.ILLMChatResponse) error) error { + ollamaMessages, err := convertMessages(messages) + if err != nil { + return err + } + + ollamaTools, err := convertTool(tools) + if err != nil { + return err + } + + httpReq, client, err := initRequestClient(ctx, mcpAgent.LLMUrl, mcpAgent.Model, true, ollamaMessages, ollamaTools) + + return o.doChatStreamRequest(ctx, httpReq, client, onChunk) +} + +func (o *ollama) doChatStreamRequest(ctx context.Context, httpReq *http.Request, client *http.Client, onChunk func(models.ILLMChatResponse) error) error { + resp, err := client.Do(httpReq) + if err != nil { + return errors.Wrap(err, "do request") + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return errors.Errorf("unexpected status code %d: %s", resp.StatusCode, string(body)) + } + + decoder := json.NewDecoder(resp.Body) + for { + var chunk OllamaChatResponse + if err := decoder.Decode(&chunk); err != nil { + if err == io.EOF { + break + } + return errors.Wrap(err, "decode stream chunk") + } + + if onChunk != nil { + if err := onChunk(&chunk); err != nil { + return errors.Wrap(err, "process chunk") + } + } + + if chunk.Done { + break + } + } + + return nil +} diff --git a/pkg/llm/drivers/llm_client/openai.go b/pkg/llm/drivers/llm_client/openai.go new file mode 100644 index 0000000000..fd27eb21ba --- /dev/null +++ b/pkg/llm/drivers/llm_client/openai.go @@ -0,0 +1,517 @@ +package llm_client + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "strings" + "time" + + "github.com/mark3labs/mcp-go/mcp" + "yunion.io/x/pkg/errors" + + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/llm/models" +) + +func init() { + models.RegisterLLMClientDriver(newOpenAI()) +} + +type openai struct{} + +func newOpenAI() models.ILLMClient { + return new(openai) +} + +func (o *openai) GetType() api.LLMClientType { + return api.LLM_CLIENT_OPENAI +} + +func (o *openai) Chat(ctx context.Context, mcpAgent *models.SMCPAgent, messages interface{}, tools interface{}) (models.ILLMChatResponse, error) { + // 转换 messages + var openaiMessages []OpenAIChatMessage + if msgs, ok := messages.([]OpenAIChatMessage); ok { + openaiMessages = msgs + } else if msgs, ok := messages.([]models.ILLMChatMessage); ok { + openaiMessages = make([]OpenAIChatMessage, len(msgs)) + for i, msg := range msgs { + // Check if it's an OpenAIChatMessage to preserve ToolCallID + if om, ok := msg.(*OpenAIChatMessage); ok { + openaiMessages[i] = *om + } else { + // General conversion + openaiMessages[i] = OpenAIChatMessage{ + Role: msg.GetRole(), + Content: msg.GetContent(), + } + // 转换工具调用 + if toolCalls := msg.GetToolCalls(); len(toolCalls) > 0 { + openaiMessages[i].ToolCalls = make([]OpenAIToolCall, len(toolCalls)) + for j, tc := range toolCalls { + fc := tc.GetFunction() + argsBytes, _ := json.Marshal(fc.GetArguments()) + openaiMessages[i].ToolCalls[j] = OpenAIToolCall{ + ID: tc.GetId(), + Type: "function", + Function: OpenAIFunctionCall{ + Name: fc.GetName(), + Arguments: string(argsBytes), + }, + } + } + } + } + } + } else { + return nil, errors.Error("invalid messages type") + } + + // 转换 tools + var openaiTools []OpenAITool + if ts, ok := tools.([]OpenAITool); ok { + openaiTools = ts + } else if ts, ok := tools.([]models.ILLMTool); ok { + openaiTools = make([]OpenAITool, len(ts)) + for i, t := range ts { + tf := t.GetFunction() + openaiTools[i] = OpenAITool{ + Type: t.GetType(), + Function: OpenAIToolFunction{ + Name: tf.GetName(), + Description: tf.GetDescription(), + Parameters: tf.GetParameters(), + }, + } + } + } else if tools == nil { + openaiTools = nil + } + + return o.doChatRequest(ctx, mcpAgent, openaiMessages, openaiTools) +} + +func (o *openai) ChatStream(ctx context.Context, mcpAgent *models.SMCPAgent, messages interface{}, tools interface{}, onChunk func(models.ILLMChatResponse) error) error { + // 转换 messages + var openaiMessages []OpenAIChatMessage + + if msgs, ok := messages.([]OpenAIChatMessage); ok { + openaiMessages = msgs + } else { + var ilMsgs []models.ILLMChatMessage + if msgs, ok := messages.([]models.ILLMChatMessage); ok { + ilMsgs = msgs + } else if msg, ok := messages.(models.ILLMChatMessage); ok { + ilMsgs = []models.ILLMChatMessage{msg} + } else { + return errors.Error("invalid messages type") + } + + openaiMessages = make([]OpenAIChatMessage, len(ilMsgs)) + for i, msg := range ilMsgs { + // Check if it's an OpenAIChatMessage to preserve ToolCallID + if om, ok := msg.(*OpenAIChatMessage); ok { + openaiMessages[i] = *om + } else { + // General conversion + openaiMessages[i] = OpenAIChatMessage{ + Role: msg.GetRole(), + Content: msg.GetContent(), + } + // 转换工具调用 + if toolCalls := msg.GetToolCalls(); len(toolCalls) > 0 { + openaiMessages[i].ToolCalls = make([]OpenAIToolCall, len(toolCalls)) + for j, tc := range toolCalls { + fc := tc.GetFunction() + argsBytes, _ := json.Marshal(fc.GetArguments()) + openaiMessages[i].ToolCalls[j] = OpenAIToolCall{ + ID: tc.GetId(), + Type: "function", + Function: OpenAIFunctionCall{ + Name: fc.GetName(), + Arguments: string(argsBytes), + }, + } + } + } + } + } + } + + // 转换 tools + var openaiTools []OpenAITool + if ts, ok := tools.([]OpenAITool); ok { + openaiTools = ts + } else if ts, ok := tools.([]models.ILLMTool); ok { + openaiTools = make([]OpenAITool, len(ts)) + for i, t := range ts { + tf := t.GetFunction() + openaiTools[i] = OpenAITool{ + Type: t.GetType(), + Function: OpenAIToolFunction{ + Name: tf.GetName(), + Description: tf.GetDescription(), + Parameters: tf.GetParameters(), + }, + } + } + } else if tools == nil { + openaiTools = nil + } + + return o.doChatStreamRequest(ctx, mcpAgent, openaiMessages, openaiTools, onChunk) +} + +func (o *openai) doChatStreamRequest(ctx context.Context, mcpAgent *models.SMCPAgent, messages []OpenAIChatMessage, tools []OpenAITool, onChunk func(models.ILLMChatResponse) error) error { + req := OpenAIChatRequest{ + Model: mcpAgent.Model, + Messages: messages, + Tools: tools, + Stream: true, + } + + reqBody, err := json.Marshal(req) + if err != nil { + return errors.Wrap(err, "marshal request") + } + + endpoint := strings.TrimSuffix(mcpAgent.LLMUrl, "/") + // Default to /v1/chat/completions if not specified and not a custom path + if !strings.Contains(endpoint, "/chat/completions") { + if strings.HasSuffix(endpoint, "/v1") { + endpoint = endpoint + "/chat/completions" + } else { + endpoint = endpoint + "/v1/chat/completions" + } + } + + httpReq, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewReader(reqBody)) + if err != nil { + return errors.Wrap(err, "create request") + } + httpReq.Header.Set("Content-Type", "application/json") + if mcpAgent.ApiKey != "" { + httpReq.Header.Set("Authorization", "Bearer "+mcpAgent.ApiKey) + } + + client := &http.Client{ + // Stream request no timeout + Timeout: 0, + } + + resp, err := client.Do(httpReq) + if err != nil { + return errors.Wrap(err, "do request") + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return errors.Errorf("unexpected status code %d: %s", resp.StatusCode, string(body)) + } + + scanner := bufio.NewScanner(resp.Body) + for scanner.Scan() { + line := scanner.Text() + line = strings.TrimSpace(line) + if line == "" { + continue + } + + if !strings.HasPrefix(line, "data: ") { + continue + } + + data := strings.TrimPrefix(line, "data: ") + if data == "[DONE]" { + break + } + + var chunk OpenAIChatStreamResponse + if err := json.Unmarshal([]byte(data), &chunk); err != nil { + return errors.Wrapf(err, "decode stream chunk: %s", data) + } + + if onChunk != nil { + if err := onChunk(&chunk); err != nil { + return errors.Wrap(err, "process chunk") + } + } + } + + if err := scanner.Err(); err != nil { + return errors.Wrap(err, "read stream") + } + + return nil +} + +func (o *openai) doChatRequest(ctx context.Context, mcpAgent *models.SMCPAgent, messages []OpenAIChatMessage, tools []OpenAITool) (*OpenAIChatResponse, error) { + req := OpenAIChatRequest{ + Model: mcpAgent.Model, + Messages: messages, + Tools: tools, + } + + reqBody, err := json.Marshal(req) + if err != nil { + return nil, errors.Wrap(err, "marshal request") + } + + endpoint := strings.TrimSuffix(mcpAgent.LLMUrl, "/") + // Default to /v1/chat/completions if not specified and not a custom path + if !strings.Contains(endpoint, "/chat/completions") { + if strings.HasSuffix(endpoint, "/v1") { + endpoint = endpoint + "/chat/completions" + } else { + endpoint = endpoint + "/v1/chat/completions" + } + } + + httpReq, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewReader(reqBody)) + if err != nil { + return nil, errors.Wrap(err, "create request") + } + httpReq.Header.Set("Content-Type", "application/json") + if mcpAgent.ApiKey != "" { + httpReq.Header.Set("Authorization", "Bearer "+mcpAgent.ApiKey) + } + + client := &http.Client{ + Timeout: 300 * time.Second, + } + + resp, err := client.Do(httpReq) + if err != nil { + return nil, errors.Wrap(err, "do request") + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, errors.Wrap(err, "read response body") + } + + if resp.StatusCode != http.StatusOK { + return nil, errors.Errorf("unexpected status code %d: %s", resp.StatusCode, string(body)) + } + + var chatResp OpenAIChatResponse + if err := json.Unmarshal(body, &chatResp); err != nil { + return nil, errors.Wrapf(err, "decode response: %s", string(body)) + } + + if len(chatResp.Choices) == 0 { + return nil, errors.Error("no choices in response") + } + + return &chatResp, nil +} + +func (o *openai) NewUserMessage(content string) models.ILLMChatMessage { + return &OpenAIChatMessage{ + Role: "user", + Content: content, + } +} + +func (o *openai) NewAssistantMessageWithToolCalls(toolCalls []models.ILLMToolCall) models.ILLMChatMessage { + openaiToolCalls := make([]OpenAIToolCall, len(toolCalls)) + for i, tc := range toolCalls { + if otc, ok := tc.(*OpenAIToolCall); ok { + openaiToolCalls[i] = *otc + } else { + fc := tc.GetFunction() + argsBytes, _ := json.Marshal(fc.GetArguments()) + openaiToolCalls[i] = OpenAIToolCall{ + ID: tc.GetId(), + Type: "function", + Function: OpenAIFunctionCall{ + Name: fc.GetName(), + Arguments: string(argsBytes), + }, + } + } + } + + return &OpenAIChatMessage{ + Role: "assistant", + ToolCalls: openaiToolCalls, + } +} + +func (o *openai) NewToolMessage(toolId string, toolName string, content string) models.ILLMChatMessage { + return &OpenAIChatMessage{ + Role: "tool", + ToolCallID: toolId, + Content: content, + } +} + +func (o *openai) NewSystemMessage(content string) models.ILLMChatMessage { + return &OpenAIChatMessage{ + Role: "system", + Content: content, + } +} + +func (o *openai) ConvertMCPTools(mcpTools []mcp.Tool) []models.ILLMTool { + tools := make([]models.ILLMTool, len(mcpTools)) + for i, t := range mcpTools { + var params map[string]interface{} + if t.RawInputSchema != nil { + _ = json.Unmarshal(t.RawInputSchema, ¶ms) + } else { + schemaBytes, _ := json.Marshal(t.InputSchema) + _ = json.Unmarshal(schemaBytes, ¶ms) + } + tools[i] = &OpenAITool{ + Type: "function", + Function: OpenAIToolFunction{ + Name: t.Name, + Description: t.Description, + Parameters: params, + }, + } + } + return tools +} + +// Structures + +type OpenAIChatMessage struct { + Role string `json:"role"` + Content string `json:"content"` + ToolCalls []OpenAIToolCall `json:"tool_calls,omitempty"` + ToolCallID string `json:"tool_call_id,omitempty"` +} + +func (m *OpenAIChatMessage) GetRole() string { return m.Role } +func (m *OpenAIChatMessage) GetContent() string { return m.Content } +func (m *OpenAIChatMessage) GetToolCalls() []models.ILLMToolCall { + if len(m.ToolCalls) == 0 { + return nil + } + toolCalls := make([]models.ILLMToolCall, len(m.ToolCalls)) + for i := range m.ToolCalls { + tc := m.ToolCalls[i] + toolCalls[i] = &tc + } + return toolCalls +} + +type OpenAIToolCall struct { + ID string `json:"id"` + Type string `json:"type"` + Function OpenAIFunctionCall `json:"function"` +} + +func (tc *OpenAIToolCall) GetFunction() models.ILLMFunctionCall { return &tc.Function } +func (tc *OpenAIToolCall) GetId() string { return tc.ID } + +type OpenAIFunctionCall struct { + Name string `json:"name"` + Arguments string `json:"arguments"` +} + +func (fc *OpenAIFunctionCall) GetName() string { return fc.Name } +func (fc *OpenAIFunctionCall) GetArguments() map[string]interface{} { + var args map[string]interface{} + _ = json.Unmarshal([]byte(fc.Arguments), &args) + return args +} + +type OpenAITool struct { + Type string `json:"type"` + Function OpenAIToolFunction `json:"function"` +} + +func (t *OpenAITool) GetType() string { return t.Type } +func (t *OpenAITool) GetFunction() models.ILLMToolFunction { return &t.Function } + +type OpenAIToolFunction struct { + Name string `json:"name"` + Description string `json:"description"` + Parameters map[string]interface{} `json:"parameters"` +} + +func (tf *OpenAIToolFunction) GetName() string { return tf.Name } +func (tf *OpenAIToolFunction) GetDescription() string { return tf.Description } +func (tf *OpenAIToolFunction) GetParameters() map[string]interface{} { return tf.Parameters } + +type OpenAIChatRequest struct { + Model string `json:"model"` + Messages []OpenAIChatMessage `json:"messages"` + Tools []OpenAITool `json:"tools,omitempty"` + Stream bool `json:"stream,omitempty"` +} + +type OpenAIChatResponse struct { + ID string `json:"id"` + Choices []OpenAIChoice `json:"choices"` +} + +type OpenAIChoice struct { + Message OpenAIChatMessage `json:"message"` + FinishReason string `json:"finish_reason"` +} + +func (r *OpenAIChatResponse) GetContent() string { + if len(r.Choices) > 0 { + return r.Choices[0].Message.Content + } + return "" +} + +func (r *OpenAIChatResponse) HasToolCalls() bool { + return len(r.Choices) > 0 && len(r.Choices[0].Message.ToolCalls) > 0 +} + +func (r *OpenAIChatResponse) GetToolCalls() []models.ILLMToolCall { + if len(r.Choices) == 0 { + return nil + } + return r.Choices[0].Message.GetToolCalls() +} + +type OpenAIChatStreamResponse struct { + ID string `json:"id"` + Choices []OpenAIChatStreamChoice `json:"choices"` +} + +type OpenAIChatStreamChoice struct { + Delta OpenAIChatStreamDelta `json:"delta"` + FinishReason string `json:"finish_reason"` +} + +type OpenAIChatStreamDelta struct { + Role string `json:"role,omitempty"` + Content string `json:"content,omitempty"` + ToolCalls []OpenAIToolCall `json:"tool_calls,omitempty"` +} + +func (r *OpenAIChatStreamResponse) GetContent() string { + if len(r.Choices) > 0 { + return r.Choices[0].Delta.Content + } + return "" +} + +func (r *OpenAIChatStreamResponse) HasToolCalls() bool { + return len(r.Choices) > 0 && len(r.Choices[0].Delta.ToolCalls) > 0 +} + +func (r *OpenAIChatStreamResponse) GetToolCalls() []models.ILLMToolCall { + if len(r.Choices) == 0 { + return nil + } + toolCalls := make([]models.ILLMToolCall, len(r.Choices[0].Delta.ToolCalls)) + for i := range r.Choices[0].Delta.ToolCalls { + tc := r.Choices[0].Delta.ToolCalls[i] + toolCalls[i] = &tc + } + return toolCalls +} diff --git a/pkg/llm/drivers/llm_container/ollama.go b/pkg/llm/drivers/llm_container/ollama.go index 5a800e0295..13c52c4b44 100644 --- a/pkg/llm/drivers/llm_container/ollama.go +++ b/pkg/llm/drivers/llm_container/ollama.go @@ -3,6 +3,7 @@ package llm_container import ( "context" "crypto/sha256" + "database/sql" "encoding/hex" "encoding/json" "fmt" @@ -594,3 +595,54 @@ func parseModelName(path string) string { } return strings.TrimRight(model, `\`) } + +func (o *ollama) GetLLMUrl(ctx context.Context, userCred mcclient.TokenCredential, llm *models.SLLM) (string, error) { + sku, err := llm.GetLLMSku("") + if err != nil { + return "", errors.Wrap(err, "get llm sku") + } + // 查询 accessinfo + accessInfo := &models.SAccessInfo{} + q := models.GetAccessInfoManager().Query().Equals("llm_id", llm.Id) + err = q.First(accessInfo) + if err != nil { + if errors.Cause(err) == sql.ErrNoRows { + // 如果没有 accessinfo，使用默认 localhost + server, err := llm.GetServer(ctx) + if err != nil { + return "", errors.Wrap(err, "get server") + } + // 从 IPs 字符串中选择第一个 IP + ips := strings.Split(strings.TrimSpace(server.IPs), ",") + if len(ips) == 0 || len(strings.TrimSpace(ips[0])) == 0 { + return "", errors.Error("server IPs is empty") + } + firstIP := strings.TrimSpace(ips[0]) + return fmt.Sprintf("http://%s:%d", firstIP, api.LLM_OLLAMA_DEFAULT_PORT), nil + } + return "", errors.Wrap(err, "query accessinfo") + } + + // 判断网络类型 + networkType := sku.NetworkType + if networkType == string(computeapi.NETWORK_TYPE_GUEST) { + // guest 网络：使用 LLM IP + 默认端口 + if len(llm.LLMIp) == 0 { + return "", errors.Error("LLM IP is empty for guest network") + } + return fmt.Sprintf("http://%s:%d", llm.LLMIp, api.LLM_OLLAMA_DEFAULT_PORT), nil + } else { + // hostlocal 或其他网络类型：使用宿主机 IP + 映射端口 + server, err := llm.GetServer(ctx) + if err != nil { + return "", errors.Wrap(err, "get server") + } + if len(server.HostAccessIp) == 0 { + return "", errors.Error("host access IP is empty") + } + if accessInfo.AccessPort == 0 { + return "", errors.Error("access port is not set") + } + return fmt.Sprintf("http://%s:%d", server.HostAccessIp, accessInfo.AccessPort), nil + } +} diff --git a/pkg/llm/models/llm.go b/pkg/llm/models/llm.go index 51671920eb..65b6aed6b8 100644 --- a/pkg/llm/models/llm.go +++ b/pkg/llm/models/llm.go @@ -356,3 +356,7 @@ func (llm *SLLM) StartSyncStatusTask(ctx context.Context, userCred mcclient.Toke } return nil } + +func (llm *SLLM) GetLLMUrl(ctx context.Context, userCred mcclient.TokenCredential) (string, error) { + return llm.GetLLMContainerDriver().GetLLMUrl(ctx, userCred, llm) +} diff --git a/pkg/llm/models/llm_client_driver.go b/pkg/llm/models/llm_client_driver.go new file mode 100644 index 0000000000..63fef565f7 --- /dev/null +++ b/pkg/llm/models/llm_client_driver.go @@ -0,0 +1,81 @@ +package models + +import ( + "context" + + "github.com/mark3labs/mcp-go/mcp" + + llm "yunion.io/x/onecloud/pkg/apis/llm" +) + +type ILLMChatMessage interface { + GetRole() string + GetContent() string + GetToolCalls() []ILLMToolCall +} + +// ILLMToolCall 表示工具调用接口 +type ILLMToolCall interface { + GetId() string + GetFunction() ILLMFunctionCall +} + +// ILLMFunctionCall 表示函数调用详情接口 +type ILLMFunctionCall interface { + GetName() string + GetArguments() map[string]interface{} +} + +// ILLMTool 表示工具定义接口 +type ILLMTool interface { + GetType() string + GetFunction() ILLMToolFunction +} + +// ILLMToolFunction 表示工具函数定义接口 +type ILLMToolFunction interface { + GetName() string + GetDescription() string + GetParameters() map[string]interface{} +} + +// ILLMChatResponse 表示 LLM 聊天响应接口 +// 参考 mcp_agent.go 中的 LLMChatResponse 接口设计 +type ILLMChatResponse interface { + // HasToolCalls 检查响应是否包含工具调用 + HasToolCalls() bool + // GetToolCalls 获取工具调用列表 + GetToolCalls() []ILLMToolCall + // GetContent 获取响应内容 + GetContent() string +} + +type ILLMClient interface { + GetType() llm.LLMClientType + + Chat(ctx context.Context, mcpAgent *SMCPAgent, messages interface{}, tools interface{}) (ILLMChatResponse, error) + ChatStream(ctx context.Context, mcpAgent *SMCPAgent, messages interface{}, tools interface{}, onChunk func(ILLMChatResponse) error) error + + NewUserMessage(content string) ILLMChatMessage + NewAssistantMessageWithToolCalls(toolCalls []ILLMToolCall) ILLMChatMessage + NewToolMessage(toolId string, toolName string, content string) ILLMChatMessage + NewSystemMessage(content string) ILLMChatMessage + + ConvertMCPTools(mcpTools []mcp.Tool) []ILLMTool +} + +var ( + llmClientDrivers = newDrivers() +) + +func RegisterLLMClientDriver(drv ILLMClient) { + registerDriver(llmClientDrivers, drv.GetType(), drv) +} + +func GetLLMClientDriver(typ llm.LLMClientType) ILLMClient { + return getDriver[llm.LLMClientType, ILLMClient](llmClientDrivers, typ) +} + +func GetLLMClientDriverWithError(typ llm.LLMClientType) (ILLMClient, error) { + return getDriverWithError[llm.LLMClientType, ILLMClient](llmClientDrivers, typ) +} diff --git a/pkg/llm/models/llm_container_driver.go b/pkg/llm/models/llm_container_driver.go index 60b61db4e4..59b8c42edb 100644 --- a/pkg/llm/models/llm_container_driver.go +++ b/pkg/llm/models/llm_container_driver.go @@ -57,10 +57,6 @@ func getDriverWithError[K ~string, D any](drvs *drivers, typ K) (D, error) { return drv.(D), nil } -// type ILLMContainerPullModel interface { -// DownloadModel(ctx context.Context, userCred mcclient.TokenCredential, llm *SLLM, modelName string, modelTag string) error -// } - type ILLMContainerInstantApp interface { GetProbedInstantModelsExt(ctx context.Context, userCred mcclient.TokenCredential, llm *SLLM, mdlIds ...string) (map[string]llm.LLMInternalInstantMdlInfo, error) DetectModelPaths(ctx context.Context, userCred mcclient.TokenCredential, llm *SLLM, pkgInfo llm.LLMInternalInstantMdlInfo) ([]string, error) @@ -83,9 +79,12 @@ type ILLMContainerDriver interface { GetType() llm.LLMContainerType GetContainerSpec(ctx context.Context, llm *SLLM, image *SLLMImage, sku *SLLMSku, props []string, devices []computeapi.SIsolatedDevice, diskId string) *computeapi.PodContainerCreateInput - // ILLMContainerPullModel - ILLMContainerInstantApp + ILLMContainerMCPAgent +} + +type ILLMContainerMCPAgent interface { + GetLLMUrl(ctx context.Context, userCred mcclient.TokenCredential, llm *SLLM) (string, error) } var ( diff --git a/pkg/llm/models/mcp_agent.go b/pkg/llm/models/mcp_agent.go new file mode 100644 index 0000000000..cbc7c927af --- /dev/null +++ b/pkg/llm/models/mcp_agent.go @@ -0,0 +1,438 @@ +package models + +import ( + "context" + "fmt" + "net/http" + "strings" + "time" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + "yunion.io/x/sqlchemy" + + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/appsrv" + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/llm/options" + "yunion.io/x/onecloud/pkg/llm/utils" + "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/util/stringutils2" +) + +func init() { + GetMCPAgentManager() +} + +var mcpAgentManager *SMCPAgentManager + +var mcpAgentWorkerMan *appsrv.SWorkerManager + +func GetMCPAgentManager() *SMCPAgentManager { + if mcpAgentManager != nil { + return mcpAgentManager + } + mcpAgentManager = &SMCPAgentManager{ + SSharableVirtualResourceBaseManager: db.NewSharableVirtualResourceBaseManager( + SMCPAgent{}, + "mcp_agents_tbl", + "mcp_agent", + "mcp_agents", + ), + } + mcpAgentManager.SetVirtualObject(mcpAgentManager) + return mcpAgentManager +} + +type SMCPAgentManager struct { + db.SSharableVirtualResourceBaseManager +} + +type SMCPAgent struct { + db.SSharableVirtualResourceBase + + // LLMUrl 对应后端大模型的 base 请求地址 + LLMUrl string `width:"512" charset:"utf8" nullable:"false" list:"user" create:"required" update:"user"` + // LLMDriver 对应使用的大模型驱动（llm_client），现在可以被设置为 ollama 或 openai + LLMDriver string `width:"64" charset:"ascii" nullable:"false" list:"user" create:"required" update:"user"` + // Model 使用的模型名称 + Model string `width:"128" charset:"ascii" nullable:"false" list:"user" create:"required" update:"user"` + // ApiKey 即在 llm_driver 中需要用到的认证 + ApiKey string `width:"512" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` + // McpServer 即 mcp 服务器的后端地址 + McpServer string `width:"512" charset:"utf8" nullable:"false" list:"user" create:"optional" update:"user"` +} + +func (man *SMCPAgentManager) CustomizeHandlerInfo(info *appsrv.SHandlerInfo) { + man.SSharableVirtualResourceBaseManager.CustomizeHandlerInfo(info) + + // log.Infoln("query name of handler info", info.GetName(nil)) + + switch info.GetName(nil) { + case "get_specific": + info.SetProcessTimeout(time.Hour * 4).SetWorkerManager(mcpAgentWorkerMan) + } +} + +func (man *SMCPAgentManager) ValidateCreateData(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, input *api.MCPAgentCreateInput) (*api.MCPAgentCreateInput, error) { + var err error + input.SharableVirtualResourceCreateInput, err = man.SSharableVirtualResourceBaseManager.ValidateCreateData(ctx, userCred, ownerId, query, input.SharableVirtualResourceCreateInput) + if err != nil { + return input, errors.Wrap(err, "validate SharableVirtualResourceCreateInput") + } + + // 如果提供了 llm_id，则通过 LLM 获取 llm_url 和 model + if len(input.LLMId) > 0 { + llmObj, err := GetLLMManager().FetchByIdOrName(ctx, userCred, input.LLMId) + if err != nil { + return input, errors.Wrapf(err, "fetch LLM by id %s", input.LLMId) + } + llm := llmObj.(*SLLM) + llmUrl, err := llm.GetLLMUrl(ctx, userCred) + if err != nil { + return input, errors.Wrapf(err, "get LLM URL from LLM %s", input.LLMId) + } + input.LLMUrl = llmUrl + + sku, err := llm.GetLLMSku("") + if err != nil { + return input, errors.Wrapf(err, "get LLM Sku from LLM %s", input.LLMId) + } + input.Model = sku.LLMModelName + } + + // 验证 llm_url 不为空 + if len(input.LLMUrl) == 0 { + return input, errors.Wrap(httperrors.ErrInputParameter, "llm_url is required (or provide llm_id to auto-fetch)") + } + + // 验证 llm_driver 必须是 ollama 或 openai + input.LLMDriver = strings.ToLower(strings.TrimSpace(input.LLMDriver)) + if !api.IsLLMClientType(input.LLMDriver) { + return input, errors.Wrapf(httperrors.ErrInputParameter, "llm_driver must be one of: %s, got: %s", api.LLM_CLIENT_TYPES.List(), input.LLMDriver) + } + + // 验证 model 不为空 + if len(input.Model) == 0 { + return input, errors.Wrap(httperrors.ErrInputParameter, "model is required") + } + + // 验证 mcp_server 不为空 + if len(input.McpServer) == 0 { + input.McpServer = options.Options.MCPServerURL + } + + // 对于 openai 驱动，api_key 是必需的 + if input.LLMDriver == string(api.LLM_CLIENT_OPENAI) && len(input.ApiKey) == 0 { + return input, errors.Wrap(httperrors.ErrInputParameter, "api_key is required when llm_driver is openai") + } + + input.Status = api.STATUS_READY + return input, nil +} + +func (man *SMCPAgentManager) ValidateUpdateData(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, input *api.MCPAgentUpdateInput) (*api.MCPAgentUpdateInput, error) { + var err error + input.SharableVirtualResourceCreateInput, err = man.SSharableVirtualResourceBaseManager.ValidateCreateData(ctx, userCred, ownerId, query, input.SharableVirtualResourceCreateInput) + if err != nil { + return input, errors.Wrap(err, "validate SharableVirtualResourceCreateInput") + } + + // 如果提供了 llm_id，则通过 LLM 获取 llm_url 和 model + if input.LLMId != nil && len(*input.LLMId) > 0 { + llmObj, err := GetLLMManager().FetchByIdOrName(ctx, userCred, *input.LLMId) + if err != nil { + return input, errors.Wrapf(err, "fetch LLM by id %s", *input.LLMId) + } + llm := llmObj.(*SLLM) + llmUrl, err := llm.GetLLMUrl(ctx, userCred) + if err != nil { + return input, errors.Wrapf(err, "get LLM URL from LLM %s", *input.LLMId) + } + input.LLMUrl = &llmUrl + + sku, err := llm.GetLLMSku("") + if err != nil { + return input, errors.Wrapf(err, "get LLM Sku from LLM %s", *input.LLMId) + } + input.Model = &sku.LLMModelName + } + + // 如果更新 llm_driver，验证其值 + if input.LLMDriver != nil { + *input.LLMDriver = strings.ToLower(strings.TrimSpace(*input.LLMDriver)) + if !api.IsLLMClientType(*input.LLMDriver) { + return input, errors.Wrapf(httperrors.ErrInputParameter, "llm_driver must be one of: %s, got: %s", api.LLM_CLIENT_TYPES.List(), *input.LLMDriver) + } + } + + return input, nil +} + +func (man *SMCPAgentManager) ListItemFilter( + ctx context.Context, + q *sqlchemy.SQuery, + userCred mcclient.TokenCredential, + input api.MCPAgentListInput, +) (*sqlchemy.SQuery, error) { + q, err := man.SSharableVirtualResourceBaseManager.ListItemFilter(ctx, q, userCred, input.SharableVirtualResourceListInput) + if err != nil { + return nil, errors.Wrapf(err, "SSharableVirtualResourceBaseManager.ListItemFilter") + } + + if len(input.LLMDriver) > 0 { + q = q.Equals("llm_driver", strings.ToLower(strings.TrimSpace(input.LLMDriver))) + } + + return q, nil +} + +func (manager *SMCPAgentManager) FetchCustomizeColumns( + ctx context.Context, + userCred mcclient.TokenCredential, + query jsonutils.JSONObject, + objs []interface{}, + fields stringutils2.SSortedStrings, + isList bool, +) []api.MCPAgentDetails { + rows := make([]api.MCPAgentDetails, len(objs)) + vrows := manager.SSharableVirtualResourceBaseManager.FetchCustomizeColumns(ctx, userCred, query, objs, fields, isList) + agents := []SMCPAgent{} + jsonutils.Update(&agents, objs) + + for i := range rows { + rows[i].SharableVirtualResourceDetails = vrows[i] + if i < len(agents) { + rows[i].LLMUrl = agents[i].LLMUrl + rows[i].LLMDriver = agents[i].LLMDriver + rows[i].Model = agents[i].Model + rows[i].ApiKey = agents[i].ApiKey + rows[i].McpServer = agents[i].McpServer + } + } + + return rows +} + +func (mcp *SMCPAgent) GetLLMClientDriver() ILLMClient { + return GetLLMClientDriver(api.LLMClientType(mcp.LLMDriver)) +} + +func (mcp *SMCPAgent) GetDetailsMcpTools(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (jsonutils.JSONObject, error) { + // 创建 MCP 客户端 + timeout := time.Duration(options.Options.MCPAgentTimeout) * time.Second + mcpClient := utils.NewMCPClient(options.Options.MCPServerURL, timeout, userCred) + + // 获取工具列表 + tools, err := mcpClient.ListTools(ctx) + if err != nil { + return nil, errors.Wrap(err, "list MCP tools") + } + + return jsonutils.Marshal(tools), nil +} + +func (mcp *SMCPAgent) GetDetailsToolRequest( + ctx context.Context, + userCred mcclient.TokenCredential, + input api.LLMToolRequestInput, +) (jsonutils.JSONObject, error) { + // 创建 MCP 客户端 + timeout := time.Duration(options.Options.MCPAgentTimeout) * time.Second + mcpClient := utils.NewMCPClient(options.Options.MCPServerURL, timeout, userCred) + defer mcpClient.Close() + + // 调用工具 + result, err := mcpClient.CallTool(ctx, input.ToolName, input.Arguments) + if err != nil { + return nil, errors.Wrapf(err, "call tool %s", input.ToolName) + } + + return jsonutils.Marshal(result), nil +} + +// func (mcp *SMCPAgent) GetDetailsChatTest( +// ctx context.Context, +// userCred mcclient.TokenCredential, +// input api.LLMChatTestInput, +// ) (jsonutils.JSONObject, error) { +// llmClient := mcp.GetLLMClientDriver() +// if llmClient == nil { +// return nil, errors.Error("failed to get LLM client driver") +// } + +// message := llmClient.NewUserMessage(input.Message) + +// result, err := llmClient.Chat(ctx, mcp, []ILLMChatMessage{message}, nil) +// if err != nil { +// return nil, errors.Wrap(err, "chat with LLM") +// } + +// return jsonutils.Marshal(result), nil +// } + +func (mcp *SMCPAgent) GetDetailsChatStream( + ctx context.Context, + userCred mcclient.TokenCredential, + input api.LLMChatTestInput, +) (jsonutils.JSONObject, error) { + llmClient := mcp.GetLLMClientDriver() + if llmClient == nil { + return nil, errors.Error("failed to get LLM client driver") + } + + appParams := appsrv.AppContextGetParams(ctx) + if appParams == nil { + return nil, errors.Error("failed to get app params") + } + + w := appParams.Response + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + + message := llmClient.NewUserMessage(input.Message) + + err := llmClient.ChatStream(ctx, mcp, []ILLMChatMessage{message}, nil, func(chunk ILLMChatResponse) error { + content := chunk.GetContent() + if len(content) > 0 { + fmt.Fprintf(w, "%s", content) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + } + return nil + }) + if err != nil { + fmt.Fprintf(w, "\nError: %v\n", err) + } + + return nil, nil +} + +func (mcp *SMCPAgent) GetDetailsRequest( + ctx context.Context, + userCred mcclient.TokenCredential, + input api.LLMMCPAgentRequestInput, +) (jsonutils.JSONObject, error) { + // 调用 ProcessMCPAgentRequest + answer, err := mcp.process(ctx, userCred, &input) + if err != nil { + return nil, errors.Wrap(err, "process MCP agent request") + } + + // 返回结果 + result := map[string]interface{}{ + "answer": answer.Answer, + } + return jsonutils.Marshal(result), nil +} + +// process 处理用户请求 +func (mcp *SMCPAgent) process(ctx context.Context, userCred mcclient.TokenCredential, req *api.LLMMCPAgentRequestInput) (*api.MCPAgentResponse, error) { + // 获取 MCP Server 的工具列表 + mcpClient := utils.NewMCPClient(mcp.McpServer, 10*time.Minute, userCred) + defer mcpClient.Close() + mcpTools, err := mcpClient.ListTools(ctx) + if err != nil { + return nil, errors.Wrap(err, "list MCP tools") + } + log.Infof("Got %d tools from MCP Server", len(mcpTools)) + + // get llmClient + llmClient := mcp.GetLLMClientDriver() + if llmClient == nil { + return nil, errors.Error("failed to get LLM client driver") + } + + tools := llmClient.ConvertMCPTools(mcpTools) + + // 构建系统提示词 + systemPrompt := buildSystemPrompt() + + // 初始化消息历史，使用接口类型 + messages := []ILLMChatMessage{ + llmClient.NewSystemMessage(systemPrompt), + llmClient.NewUserMessage(req.Query), + } + + // 记录工具调用 + var toolCallRecords []api.MCPAgentToolCallRecord + + // Agent 循环 + for i := 0; i < api.MCPAgentMaxIterations; i++ { + log.Infof("Agent iteration %d", i+1) + + // 调用 LLM 客户端，传入接口类型 + resp, err := llmClient.Chat(ctx, mcp, messages, tools) + if err != nil { + return nil, errors.Wrap(err, "chat with LLM client") + } + + // 检查是否有工具调用 + if !resp.HasToolCalls() { + // 没有工具调用，返回最终答案 + return &api.MCPAgentResponse{ + Success: true, + Answer: resp.GetContent(), + ToolCalls: toolCallRecords, + }, nil + } + + // 处理工具调用 + toolCalls := resp.GetToolCalls() + log.Infof("Got %d tool calls from LLM", len(toolCalls)) + + // 添加助手消息（带工具调用），使用接口类型 + messages = append(messages, llmClient.NewAssistantMessageWithToolCalls(toolCalls)) + + // 执行每个工具调用 + for _, tc := range toolCalls { + fc := tc.GetFunction() + toolName := fc.GetName() + arguments := fc.GetArguments() + + // 确保 arguments 不为 nil + if arguments == nil { + arguments = make(map[string]interface{}) + } + + log.Infof("Calling tool: %s with arguments: %v", toolName, arguments) + + // 调用 MCP 工具 + result, err := mcpClient.CallTool(ctx, toolName, arguments) + resultText := utils.FormatToolResult(toolName, result, err) + log.Infoln("Get result from mcp query", resultText) + + // 记录工具调用 + toolCallRecords = append(toolCallRecords, api.MCPAgentToolCallRecord{ + ToolName: toolName, + Arguments: arguments, + Result: resultText, + }) + + // 添加工具结果消息，使用接口类型 + messages = append(messages, llmClient.NewToolMessage(tc.GetId(), toolName, resultText)) + } + } + + // 达到最大迭代次数 + return &api.MCPAgentResponse{ + Success: false, + Answer: "处理请求时达到最大迭代次数，请尝试简化您的问题。", + Error: "max iterations reached", + ToolCalls: toolCallRecords, + }, nil +} + +// buildSystemPrompt 构建系统提示词 +func buildSystemPrompt() string { + return api.MCP_AGENT_SYSTEM_PROMPT +} diff --git a/pkg/llm/options/option.go b/pkg/llm/options/option.go index 9d45f0af26..5a12236cae 100644 --- a/pkg/llm/options/option.go +++ b/pkg/llm/options/option.go @@ -26,6 +26,10 @@ type LLMOptions struct { ModelSyncTaskWaitSecs int `help:"model sync task wait seconds" default:"30"` StartTaskWorkerCount int `help:"start task worker count" default:"128"` + + // MCP Agent 配置 + MCPServerURL string `help:"MCP Server URL" default:"http://default-mcp-server:30876"` + MCPAgentTimeout int `help:"MCP Agent request timeout in seconds" default:"120"` } var ( diff --git a/pkg/llm/service/handler.go b/pkg/llm/service/handler.go index e16c3d1c63..782f8e682a 100644 --- a/pkg/llm/service/handler.go +++ b/pkg/llm/service/handler.go @@ -41,6 +41,7 @@ func InitHandlers(app *appsrv.Application, isSlave bool) { models.GetDifyManager(), models.GetInstantModelManager(), models.GetLLMInstantModelManager(), + models.GetMCPAgentManager(), } { db.RegisterModelManager(manager) handler := db.NewModelHandler(manager) diff --git a/pkg/llm/service/service.go b/pkg/llm/service/service.go index 69303c3b72..b893be0062 100644 --- a/pkg/llm/service/service.go +++ b/pkg/llm/service/service.go @@ -11,6 +11,7 @@ import ( app_common "yunion.io/x/onecloud/pkg/cloudcommon/app" "yunion.io/x/onecloud/pkg/cloudcommon/db" common_options "yunion.io/x/onecloud/pkg/cloudcommon/options" + _ "yunion.io/x/onecloud/pkg/llm/drivers/llm_client" _ "yunion.io/x/onecloud/pkg/llm/drivers/llm_container" "yunion.io/x/onecloud/pkg/llm/models" "yunion.io/x/onecloud/pkg/llm/options" diff --git a/pkg/llm/utils/mcp_client.go b/pkg/llm/utils/mcp_client.go new file mode 100644 index 0000000000..32da17f4b6 --- /dev/null +++ b/pkg/llm/utils/mcp_client.go @@ -0,0 +1,437 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/mark3labs/mcp-go/mcp" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + + "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/mcclient/auth" +) + +// mcpError represents the error object in a JSON-RPC response +type mcpError struct { + Code int `json:"code"` + Message string `json:"message"` + Data interface{} `json:"data,omitempty"` +} + +// rawMCPResponse 用于处理 MCP 响应，支持延迟解析 Result +type rawMCPResponse struct { + JSONRPC string `json:"jsonrpc"` + ID mcp.RequestId `json:"id"` + Result json.RawMessage `json:"result,omitempty"` + Error *mcpError `json:"error,omitempty"` +} + +// MCPClient 是 MCP Server 的客户端，通过 SSE 协议与 MCP Server 通信 +type MCPClient struct { + serverURL string + client *http.Client + sessionURL string + sseBody io.ReadCloser + messageID int64 + mu sync.Mutex + initialized bool + userCred mcclient.TokenCredential + + pendingReqs map[int64]chan *rawMCPResponse + reqMu sync.Mutex +} + +// NewMCPClient 创建一个新的 MCP 客户端 +func NewMCPClient(serverURL string, timeout time.Duration, userCred mcclient.TokenCredential) *MCPClient { + return &MCPClient{ + serverURL: strings.TrimSuffix(serverURL, "/"), + client: &http.Client{ + Timeout: timeout, + }, + userCred: userCred, + pendingReqs: make(map[int64]chan *rawMCPResponse), + } +} + +// connectSSE 连接 SSE 端点并开始事件循环 +func (c *MCPClient) connectSSE(ctx context.Context) error { + // 连接 SSE 端点获取 session URL + sseURL := c.serverURL + "/sse" + req, err := http.NewRequestWithContext(ctx, "GET", sseURL, nil) + if err != nil { + return errors.Wrap(err, "create SSE request") + } + req.Header.Set("Accept", "text/event-stream") + req.Header.Set("Cache-Control", "no-cache") + + resp, err := c.client.Do(req) + if err != nil { + return errors.Wrap(err, "connect to SSE") + } + + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + return errors.Errorf("SSE connection failed with status %d: %s", resp.StatusCode, string(body)) + } + + c.sseBody = resp.Body + + // Channel to signal session URL found + done := make(chan struct{}) + var initErr error + + // 读取 endpoint 事件获取 session URL + go func() { + reader := bufio.NewReader(c.sseBody) + foundSession := false + defer func() { + if !foundSession { + select { + case <-done: + default: + close(done) + } + } + }() + + for { + line, err := reader.ReadString('\n') + if err != nil { + if !foundSession { + initErr = err + } else { + log.Warningf("SSE connection closed: %v", err) + } + return + } + + line = strings.TrimSpace(line) + if strings.HasPrefix(line, "data:") { + data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if !foundSession { + if strings.Contains(data, "/message") { + // 解析 session URL + c.sessionURL = c.serverURL + data + log.Infof("MCP Client initialized with session URL: %s", c.sessionURL) + foundSession = true + close(done) + } + } else { + // 尝试解析为 JSON-RPC 响应 + var resp rawMCPResponse + if err := json.Unmarshal([]byte(data), &resp); err == nil && resp.JSONRPC == mcp.JSONRPC_VERSION { + // 提取 ID + var reqID int64 + if idVal, ok := resp.ID.Value().(int64); ok { + reqID = idVal + } else if idVal, ok := resp.ID.Value().(float64); ok { + reqID = int64(idVal) + } else { + // 可能是通知或 ID 类型不匹配，忽略 + continue + } + + c.reqMu.Lock() + ch, ok := c.pendingReqs[reqID] + if ok { + delete(c.pendingReqs, reqID) + } + c.reqMu.Unlock() + + if ok { + select { + case ch <- &resp: + default: + log.Warningf("response channel blocked for request %d", reqID) + } + } + } + } + } + } + }() + + // Wait for session URL + select { + case <-done: + if initErr != nil { + c.sseBody.Close() + return errors.Wrap(initErr, "read SSE event") + } + case <-time.After(10 * time.Second): + c.sseBody.Close() + return errors.Error("timeout waiting for session URL") + case <-ctx.Done(): + c.sseBody.Close() + return ctx.Err() + } + return nil +} + +// Initialize 初始化 MCP 客户端连接 +func (c *MCPClient) Initialize(ctx context.Context) error { + c.mu.Lock() + defer c.mu.Unlock() + + if c.initialized { + return nil + } + + if err := c.connectSSE(ctx); err != nil { + return err + } + + // 发送初始化请求 + initParams := mcp.InitializeParams{ + ProtocolVersion: "2024-11-05", + Capabilities: mcp.ClientCapabilities{}, + ClientInfo: mcp.Implementation{ + Name: "cloudpods-mcp-agent", + Version: "1.0.0", + }, + } + + initReq := mcp.JSONRPCRequest{ + JSONRPC: mcp.JSONRPC_VERSION, + ID: mcp.NewRequestId(c.nextMessageID()), + Params: initParams, + } + initReq.Method = string(mcp.MethodInitialize) + + _, err := c.sendRequest(ctx, initReq) + if err != nil { + c.sseBody.Close() + return errors.Wrap(err, "send initialize request") + } + + // 发送 initialized 通知 + notifyReq := mcp.JSONRPCRequest{ + JSONRPC: mcp.JSONRPC_VERSION, + } + notifyReq.Method = "notifications/initialized" + + _, err = c.sendRequest(ctx, notifyReq) + if err != nil { + log.Warningf("send initialized notification failed: %v", err) + } + + c.initialized = true + return nil +} + +// nextMessageID 生成下一个消息 ID +func (c *MCPClient) nextMessageID() int64 { + return atomic.AddInt64(&c.messageID, 1) +} + +// sendRequest 发送 JSON-RPC 请求 +func (c *MCPClient) sendRequest(ctx context.Context, req mcp.JSONRPCRequest) (*rawMCPResponse, error) { + var respChan chan *rawMCPResponse + var reqID int64 + var hasID bool + + if !req.ID.IsNil() { + if idVal, ok := req.ID.Value().(int64); ok { + reqID = idVal + hasID = true + } + } + + if hasID { + respChan = make(chan *rawMCPResponse, 1) + c.reqMu.Lock() + c.pendingReqs[reqID] = respChan + c.reqMu.Unlock() + + // 确保在出错返回时清理 pendingReqs + defer func() { + c.reqMu.Lock() + delete(c.pendingReqs, reqID) + c.reqMu.Unlock() + }() + } + + reqBody := jsonutils.Marshal(req) + log.Infof("MCP request: %s", reqBody.String()) + + cli := auth.Client() + if cli == nil { + cli = mcclient.NewClient("", 0, false, true, "", "") + } + + cred := c.userCred + if cred == nil { + log.Warningf("userCred is nil in sendRequest, creating empty token") + cred = &mcclient.SSimpleToken{} + } + + s := cli.NewSession(ctx, "", "", "", cred) + s.SetServiceUrl("mcp", c.sessionURL) + + _, respBody, err := s.JSONRequest("mcp", "", "POST", "", nil, reqBody) + if err != nil { + return nil, errors.Wrap(err, "send request") + } + + // 对于通知请求，可能没有响应体 + if !hasID { + return nil, nil + } + + // 如果有响应体，直接解析 + if respBody != nil { + log.Debugf("MCP response (HTTP): %s", respBody.String()) + var mcpResp rawMCPResponse + if err := respBody.Unmarshal(&mcpResp); err != nil { + return nil, errors.Wrap(err, "decode response") + } + if mcpResp.Error != nil { + return nil, errors.Errorf("MCP error %d: %s", mcpResp.Error.Code, mcpResp.Error.Message) + } + // 成功收到 HTTP 响应，从 pending 中移除（defer 会做，但我们可以提前返回） + return &mcpResp, nil + } + + // 如果响应为空，等待 SSE 推送 + select { + case mcpResp := <-respChan: + log.Debugf("MCP response (SSE): ID=%v", mcpResp.ID) + if mcpResp.Error != nil { + return nil, errors.Errorf("MCP error %d: %s", mcpResp.Error.Code, mcpResp.Error.Message) + } + return mcpResp, nil + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(30 * time.Second): + return nil, errors.Error("timeout waiting for SSE response") + } +} + +// ListTools 获取可用工具列表 +func (c *MCPClient) ListTools(ctx context.Context) ([]mcp.Tool, error) { + if !c.initialized { + if err := c.Initialize(ctx); err != nil { + return nil, errors.Wrap(err, "initialize client") + } + } + + req := mcp.JSONRPCRequest{ + JSONRPC: mcp.JSONRPC_VERSION, + ID: mcp.NewRequestId(c.nextMessageID()), + } + req.Method = string(mcp.MethodToolsList) + + resp, err := c.sendRequest(ctx, req) + if err != nil { + return nil, errors.Wrap(err, "send tools/list request") + } + + if resp == nil { + return nil, errors.Error("empty response for tools/list") + } + + var result mcp.ListToolsResult + if err := json.Unmarshal(resp.Result, &result); err != nil { + return nil, errors.Wrap(err, "decode tools list result") + } + + return result.Tools, nil +} + +// CallTool 调用工具 +func (c *MCPClient) CallTool(ctx context.Context, toolName string, arguments map[string]interface{}) (*mcp.CallToolResult, error) { + if !c.initialized { + if err := c.Initialize(ctx); err != nil { + return nil, errors.Wrap(err, "initialize client") + } + } + + params := mcp.CallToolParams{ + Name: toolName, + Arguments: arguments, + } + + req := mcp.JSONRPCRequest{ + JSONRPC: mcp.JSONRPC_VERSION, + ID: mcp.NewRequestId(c.nextMessageID()), + Params: params, + } + req.Method = string(mcp.MethodToolsCall) + + resp, err := c.sendRequest(ctx, req) + if err != nil { + return nil, errors.Wrap(err, "send tools/call request") + } + + if resp == nil { + return nil, errors.Error("empty response for tools/call") + } + + var result mcp.CallToolResult + if err := json.Unmarshal(resp.Result, &result); err != nil { + return nil, errors.Wrap(err, "decode tool call result") + } + + return &result, nil +} + +// GetToolResultText 从工具调用结果中提取文本 +func GetToolResultText(r *mcp.CallToolResult) string { + var texts []string + for _, content := range r.Content { + if textContent, ok := content.(mcp.TextContent); ok { + texts = append(texts, textContent.Text) + } + } + return strings.Join(texts, "\n") +} + +// FormatToolResult 格式化工具调用结果 +func FormatToolResult(toolName string, result *mcp.CallToolResult, err error) string { + if err != nil { + return fmt.Sprintf("工具 %s 调用失败: %v", toolName, err) + } + if result.IsError { + return fmt.Sprintf("工具 %s 返回错误: %s", toolName, GetToolResultText(result)) + } + return GetToolResultText(result) +} + +// Close 关闭客户端连接 +func (c *MCPClient) Close() error { + c.mu.Lock() + defer c.mu.Unlock() + c.initialized = false + c.sessionURL = "" + if c.sseBody != nil { + c.sseBody.Close() + c.sseBody = nil + } + return nil +} diff --git a/pkg/mcclient/modules/llm/mod_mcp_agent.go b/pkg/mcclient/modules/llm/mod_mcp_agent.go new file mode 100644 index 0000000000..d2153a1ae4 --- /dev/null +++ b/pkg/mcclient/modules/llm/mod_mcp_agent.go @@ -0,0 +1,24 @@ +package llm + +import ( + "yunion.io/x/onecloud/pkg/mcclient/modulebase" + "yunion.io/x/onecloud/pkg/mcclient/modules" +) + +type MCPAgentManager struct { + modulebase.ResourceManager +} + +var ( + MCPAgent MCPAgentManager +) + +func init() { + MCPAgent = MCPAgentManager{ + ResourceManager: modules.NewLLMManager("mcp_agent", "mcp_agents", + []string{}, + []string{}, + ), + } + modules.Register(&MCPAgent) +} diff --git a/pkg/mcclient/options/llm/mcp_agent.go b/pkg/mcclient/options/llm/mcp_agent.go new file mode 100644 index 0000000000..b9fb95bbc1 --- /dev/null +++ b/pkg/mcclient/options/llm/mcp_agent.go @@ -0,0 +1,167 @@ +package llm + +import ( + "strings" + + "yunion.io/x/jsonutils" + + "yunion.io/x/onecloud/pkg/apis" + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/mcclient/options" +) + +type MCPAgentListOptions struct { + options.BaseListOptions + + LLMDriver string `json:"llm_driver" help:"filter by llm driver (ollama or openai)"` +} + +func (o *MCPAgentListOptions) Params() (jsonutils.JSONObject, error) { + return options.ListStructToParams(o) +} + +type MCPAgentShowOptions struct { + options.BaseShowOptions +} + +func (o *MCPAgentShowOptions) Params() (jsonutils.JSONObject, error) { + return options.StructToParams(o) +} + +type MCPAgentCreateOptions struct { + apis.SharableVirtualResourceCreateInput + + LlmId string `help:"LLM 实例 ID，如果提供则自动获取 llm_url" json:"llm_id"` + LLM_URL string `help:"后端大模型的 base 请求地址" json:"llm_url"` + LLM_DRIVER string `help:"使用的大模型驱动，可以是 ollama 或 openai" json:"llm_driver" choices:"ollama|openai"` + MODEL string `help:"使用的模型名称" json:"model"` + API_KEY string `help:"访问大模型的密钥" json:"api_key"` + McpServer string `help:"mcp 服务器的后端地址" json:"mcp_server"` +} + +func (o *MCPAgentCreateOptions) Params() (jsonutils.JSONObject, error) { + return jsonutils.Marshal(o), nil +} + +type MCPAgentUpdateOptions struct { + apis.SharableVirtualResourceCreateInput + + ID string + LlmId *string `help:"LLM 实例 ID，如果提供则自动获取 llm_url" json:"llm_id,omitempty"` + LlmUrl *string `help:"后端大模型的 base 请求地址" json:"llm_url,omitempty"` + LlmDriver *string `help:"使用的大模型驱动，可以是 ollama 或 openai" json:"llm_driver,omitempty" choices:"ollama|openai"` + Model *string `help:"使用的模型名称" json:"model,omitempty"` + ApiKey *string `help:"访问大模型的密钥" json:"api_key,omitempty"` + McpServer *string `help:"mcp 服务器的后端地址" json:"mcp_server,omitempty"` +} + +func (o *MCPAgentUpdateOptions) GetId() string { + return o.ID +} + +func (o *MCPAgentUpdateOptions) Params() (jsonutils.JSONObject, error) { + // 只包含非空字段 + params := jsonutils.NewDict() + if o.LlmId != nil && len(*o.LlmId) > 0 { + params.Set("llm_id", jsonutils.NewString(*o.LlmId)) + } + if o.LlmUrl != nil && len(*o.LlmUrl) > 0 { + params.Set("llm_url", jsonutils.NewString(*o.LlmUrl)) + } + if o.LlmDriver != nil && len(*o.LlmDriver) > 0 { + params.Set("llm_driver", jsonutils.NewString(*o.LlmDriver)) + } + if o.Model != nil && len(*o.Model) > 0 { + params.Set("model", jsonutils.NewString(*o.Model)) + } + if o.ApiKey != nil && len(*o.ApiKey) > 0 { + params.Set("api_key", jsonutils.NewString(*o.ApiKey)) + } + if o.McpServer != nil && len(*o.McpServer) > 0 { + params.Set("mcp_server", jsonutils.NewString(*o.McpServer)) + } + + // 添加基础字段 + baseParams, err := options.StructToParams(&o.SharableVirtualResourceCreateInput) + if err != nil { + return nil, err + } + if baseParams != nil { + params.Update(baseParams) + } + + return params, nil +} + +type MCPAgentDeleteOptions struct { + options.BaseIdOptions +} + +func (o *MCPAgentDeleteOptions) GetId() string { + return o.ID +} + +func (o *MCPAgentDeleteOptions) Params() (jsonutils.JSONObject, error) { + return options.StructToParams(o) +} + +type MCPAgentIdOptions struct { + ID string `help:"mcp agent id" json:"-"` +} + +func (opts *MCPAgentIdOptions) GetId() string { + return opts.ID +} + +func (opts *MCPAgentIdOptions) Params() (jsonutils.JSONObject, error) { + return jsonutils.Marshal(opts), nil +} + +type MCPAgentToolRequestOptions struct { + MCPAgentIdOptions + + TOOL_NAME string `help:"tool name" json:"tool_name"` + Argument []string `help:"tool arguments, e.g. key=value" json:"argument"` +} + +func (opts *MCPAgentToolRequestOptions) Params() (jsonutils.JSONObject, error) { + input := api.LLMToolRequestInput{ + ToolName: opts.TOOL_NAME, + Arguments: make(map[string]interface{}), + } + for _, arg := range opts.Argument { + idx := strings.Index(arg, "=") + if idx > 0 { + key := arg[:idx] + val := arg[idx+1:] + input.Arguments[key] = val + } + } + return jsonutils.Marshal(input), nil +} + +type MCPAgentChatTestOptions struct { + MCPAgentIdOptions + + Message string `help:"test message to send to LLM" json:"message"` +} + +func (opts *MCPAgentChatTestOptions) Params() (jsonutils.JSONObject, error) { + input := api.LLMChatTestInput{ + Message: opts.Message, + } + return jsonutils.Marshal(input), nil +} + +type MCPAgentMCPAgentRequestOptions struct { + MCPAgentIdOptions + + Query string `help:"query to send to MCP agent" json:"query"` +} + +func (opts *MCPAgentMCPAgentRequestOptions) Params() (jsonutils.JSONObject, error) { + input := api.LLMMCPAgentRequestInput{ + Query: opts.Query, + } + return jsonutils.Marshal(input), nil +}