Copilot commented on code in PR #702: URL: https://github.com/apache/dubbo-go-pixiu/pull/702#discussion_r2218986929
########## pkg/filter/mcp/mcpserver/filter.go: ########## @@ -0,0 +1,244 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 mcpserver + +import ( + "encoding/json" + "fmt" + "io" + "net/http" +) + +import ( + "github.com/creasty/defaults" + "github.com/mark3labs/mcp-go/mcp" +) + +import ( + "github.com/apache/dubbo-go-pixiu/pkg/common/constant" + "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter" + h "github.com/apache/dubbo-go-pixiu/pkg/context/http" + "github.com/apache/dubbo-go-pixiu/pkg/logger" +) + +// FilterFactory and MCPServerFilter types +type ( + // FilterFactory is a factory to create MCP server filters. + FilterFactory struct { + cfg *Config + registry *ToolRegistry + } + + // MCPServerFilter is a filter that handles MCP protocol. + MCPServerFilter struct { + cfg *Config + registry *ToolRegistry + errorHandler *ErrorHandler + responseBuilder *ResponseBuilder + } +) + +// Apply prepares the MCP server and tool registry. +func (f *FilterFactory) Apply() error { + // Set configuration default values + if err := defaults.Set(f.cfg); err != nil { + return fmt.Errorf("failed to set config defaults: %v", err) + } + + // Initialize tool registry + f.registry = NewToolRegistry() + + // Register statically configured tools + for _, tool := range f.cfg.Tools { + if err := f.registry.RegisterTool(tool); err != nil { + return fmt.Errorf("failed to register tool %s: %v", tool.Name, err) + } + logger.Debugf("[dubbo-go-pixiu] mcp server registered tool '%s' -> cluster:%s", tool.Name, tool.Cluster) + } + + // Register statically configured resources + for _, resource := range f.cfg.Resources { + if err := f.registry.RegisterResource(resource); err != nil { + return fmt.Errorf("failed to register resource %s: %v", resource.Name, err) + } + logger.Debugf("[dubbo-go-pixiu] mcp server registered resource '%s' -> uri:%s", resource.Name, resource.URI) + } + + // Register statically configured resource templates + for _, template := range f.cfg.ResourceTemplates { + if err := f.registry.RegisterResourceTemplate(template); err != nil { + return fmt.Errorf("failed to register resource template %s: %v", template.Name, err) + } + logger.Debugf("[dubbo-go-pixiu] mcp server registered template '%s' -> pattern:%s", template.Name, template.URITemplate) + } + + // Register statically configured prompts + for _, prompt := range f.cfg.Prompts { + if err := f.registry.RegisterPrompt(prompt); err != nil { + return fmt.Errorf("failed to register prompt %s: %v", prompt.Name, err) + } + logger.Debugf("[dubbo-go-pixiu] mcp server registered prompt '%s'", prompt.Name) + } + + return nil +} + +// Config returns the configuration struct +func (f *FilterFactory) Config() any { + return f.cfg +} + +// PrepareFilterChain prepares the filter chain +func (f *FilterFactory) PrepareFilterChain(ctx *h.HttpContext, chain filter.FilterChain) error { + mcpFilter := &MCPServerFilter{ + cfg: f.cfg, + registry: f.registry, + errorHandler: GetErrorHandler(), + responseBuilder: GetResponseBuilder(), + } + chain.AppendDecodeFilters(mcpFilter) + chain.AppendEncodeFilters(mcpFilter) // Add to Encode chain + return nil +} + +// Decode processes incoming HTTP requests for MCP protocol. +func (f *MCPServerFilter) Decode(ctx *h.HttpContext) filter.FilterStatus { + // Check if it's an MCP request + if !f.isMCPRequest(ctx) { + return filter.Continue + } + + // Create MCP context wrapper + mcpCtx := NewMCPContext(ctx) + + // Read request body + body, err := io.ReadAll(ctx.Request.Body) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to read request body: %v", err) + return f.errorHandler.SendInternalError(mcpCtx, nil, "failed to read request body") + } + + // Parse JSON-RPC request + var jsonrpcReq mcp.JSONRPCRequest + if err := json.Unmarshal(body, &jsonrpcReq); err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to parse JSON-RPC request: %v", err) + return f.errorHandler.SendInternalError(mcpCtx, nil, "invalid JSON-RPC request") + } + + logger.Infof("[dubbo-go-pixiu] mcp server received request: %s (id: %v)", jsonrpcReq.Method, jsonrpcReq.ID) + + // Store information in MCP context + mcpCtx.SetMCPMethod(jsonrpcReq.Method) + mcpCtx.SetMCPRequestID(jsonrpcReq.ID) + + // Handle terminal methods (methods that don't need forwarding to backend) + if f.isTerminalMethod(jsonrpcReq.Method) { + return f.handleTerminalMethod(mcpCtx, jsonrpcReq) + } else if jsonrpcReq.Method == string(mcp.MethodToolsCall) { + // Tool call will be processed in Encode stage (IsMCPToolCall() checks method) + return f.handleToolCall(mcpCtx, jsonrpcReq) + } else { + // Unknown method + logger.Warnf("[dubbo-go-pixiu] mcp server unsupported method: %s", jsonrpcReq.Method) + return f.errorHandler.SendMethodNotFound(mcpCtx, jsonrpcReq.ID) + } +} + +// isTerminalMethod checks if it's a terminal method (methods that don't need forwarding to backend) +func (f *MCPServerFilter) isTerminalMethod(method string) bool { + switch method { + case string(mcp.MethodInitialize), string(mcp.MethodToolsList), string(mcp.MethodResourcesList), string(mcp.MethodResourcesRead), + "resources/templates/list", string(mcp.MethodPromptsList), string(mcp.MethodPromptsGet), + "notifications/initialized", string(mcp.MethodPing): + return true + default: + return false + } +} + +// handleTerminalMethod handles terminal methods +func (f *MCPServerFilter) handleTerminalMethod(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + switch req.Method { + case string(mcp.MethodInitialize): + return f.handleInitialize(ctx, req) + case string(mcp.MethodToolsList): + return f.handleToolsList(ctx, req) + case string(mcp.MethodResourcesList): + return f.handleResourcesList(ctx, req) + case string(mcp.MethodResourcesRead): + return f.handleResourceRead(ctx, req) + case "resources/templates/list": + return f.handleResourceTemplatesList(ctx, req) + case string(mcp.MethodPromptsList): + return f.handlePromptsList(ctx, req) + case string(mcp.MethodPromptsGet): + return f.handlePromptsGet(ctx, req) + case "notifications/initialized": + return f.handleNotificationsInitialized(ctx, req) + case string(mcp.MethodPing): + return f.handlePing(ctx, req) + default: + logger.Warnf("[dubbo-go-pixiu] mcp server unsupported method: %s", req.Method) + return f.errorHandler.SendMethodNotFound(ctx, req.ID) + } +} + +// Encode processes outgoing HTTP responses. +func (f *MCPServerFilter) Encode(ctx *h.HttpContext) filter.FilterStatus { + // Create MCP context wrapper and load stored MCP data + mcpCtx := NewMCPContextFromHttpContext(ctx) + + // Check if it's a tool call response + if mcpCtx.IsMCPToolCall() { + logger.Debugf("[dubbo-go-pixiu] mcp server processing tool call response: %s", ctx.Request.URL.Path) + return f.handleToolCallResponse(mcpCtx) + } + + // For regular MCP requests, no special processing needed + if mcpCtx.IsMCPRequest() { + logger.Debugf("[dubbo-go-pixiu] mcp server regular MCP request, no special processing needed") + } + + return filter.Continue +} + +// isMCPRequest checks if it's an MCP request +func (f *MCPServerFilter) isMCPRequest(ctx *h.HttpContext) bool { + return ctx.Request.URL.Path == f.cfg.Endpoint +} + +// sendJSONResponse sends a JSON response +func (f *MCPServerFilter) sendJSONResponse(ctx *MCPContext, response any) filter.FilterStatus { + responseBody, err := json.Marshal(response) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal response: %v", err) + ctx.SendLocalReply(http.StatusInternalServerError, []byte("internal server error")) + return filter.Stop + } + + // Get method and request ID for logging + method := ctx.GetMCPMethod() + requestID := ctx.GetMCPRequestID() + + logger.Infof("[dubbo-go-pixiu] mcp server response sent: %s (id: %v)", method, requestID) + + // Critical: Clear Content-Length header to prevent mismatch errors + ctx.Writer.Header().Del(constant.HeaderKeyContentLength) Review Comment: Another instance of Content-Length header deletion. Consider consolidating this repeated pattern into a shared utility method. ```suggestion clearContentLengthHeader(ctx.Writer.Header()) ``` ########## pkg/filter/mcp/mcpserver/handlers.go: ########## @@ -0,0 +1,561 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 mcpserver + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "strings" +) + +import ( + "github.com/mark3labs/mcp-go/mcp" +) + +import ( + "github.com/apache/dubbo-go-pixiu/pkg/client" + "github.com/apache/dubbo-go-pixiu/pkg/common/constant" + "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter" + "github.com/apache/dubbo-go-pixiu/pkg/logger" + "github.com/apache/dubbo-go-pixiu/pkg/model" +) + +// handleInitialize handles the initialize method +func (f *MCPServerFilter) handleInitialize(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Build server capabilities using mcp-go structures + capabilities := mcp.ServerCapabilities{ + Tools: &struct { + ListChanged bool `json:"listChanged,omitempty"` + }{ + // TODO: Dynamic update capabilities - enable after Nacos integration + // Currently set to false, future Nacos integration will support: + // 1. Dynamic discovery and registration of new backend services + // 2. Automatic generation of corresponding MCP tools + // 3. Send notifications/tools/list_changed notifications + ListChanged: false, + }, + Resources: &struct { + Subscribe bool `json:"subscribe,omitempty"` + ListChanged bool `json:"listChanged,omitempty"` + }{ + Subscribe: false, + ListChanged: false, + }, + Prompts: &struct { + ListChanged bool `json:"listChanged,omitempty"` + }{ + ListChanged: false, + }, + } + + // Build server info using mcp-go structures + serverInfo := mcp.Implementation{ + Name: f.cfg.ServerInfo.Name, + Version: f.cfg.ServerInfo.Version, + } + + // Create initialization result using mcp-go API + instructions := f.cfg.ServerInfo.Instructions + if instructions == "" { + instructions = "This MCP server provides API access through tools, documentation through resources, and AI assistance through prompts." + } + result := mcp.NewInitializeResult(mcp.LATEST_PROTOCOL_VERSION, capabilities, serverInfo, instructions) + + // Create JSON-RPC response + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleToolsList handles the tools/list method using mcp-go APIs +func (f *MCPServerFilter) handleToolsList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + tools := make([]mcp.Tool, 0, len(f.cfg.Tools)) + + // Build tools using mcp-go API for standard compliance + for _, toolCfg := range f.cfg.Tools { + // Start with basic tool options + toolOptions := []mcp.ToolOption{ + mcp.WithDescription(toolCfg.Description), + } + + // Add parameter definitions using mcp-go APIs + for _, arg := range toolCfg.Args { + switch arg.Type { + case "string": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + if defaultStr, ok := arg.Default.(string); ok { + opts = append(opts, mcp.DefaultString(defaultStr)) + } + } + if len(arg.Enum) > 0 { + opts = append(opts, mcp.Enum(arg.Enum...)) + } + toolOptions = append(toolOptions, mcp.WithString(arg.Name, opts...)) + + case "integer", "number": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + // Handle both int and float64 types from YAML parsing + switch defaultVal := arg.Default.(type) { + case float64: + opts = append(opts, mcp.DefaultNumber(defaultVal)) + case int: + opts = append(opts, mcp.DefaultNumber(float64(defaultVal))) + case int64: + opts = append(opts, mcp.DefaultNumber(float64(defaultVal))) + } + } + toolOptions = append(toolOptions, mcp.WithNumber(arg.Name, opts...)) + + case "boolean": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + if defaultBool, ok := arg.Default.(bool); ok { + opts = append(opts, mcp.DefaultBool(defaultBool)) + } + } + toolOptions = append(toolOptions, mcp.WithBoolean(arg.Name, opts...)) + } + } + + // Create tool using mcp-go API + tool := mcp.NewTool(toolCfg.Name, toolOptions...) + tools = append(tools, tool) + } + + // Build standard MCP tools list response using mcp-go structures + result := mcp.NewListToolsResult(tools, "") // empty cursor for no pagination + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleResourcesList handles the resources/list method +func (f *MCPServerFilter) handleResourcesList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Get all resources + mcpResources, err := f.registry.ToMCPResources() + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to get MCP resources: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to get resources") + } + + // Build resources list response using mcp-go structures + result := mcp.NewListResourcesResult(mcpResources, "") // empty cursor for no pagination + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handlePing handles the ping method +func (f *MCPServerFilter) handlePing(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling ping request") + + // Simple ping response + result := map[string]any{} + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleNotificationsInitialized handles notifications/initialized notification +func (f *MCPServerFilter) handleNotificationsInitialized(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server received initialized notification from client") + + // Store client initialization state + // This notification indicates that the client has completed initialization + // and is ready to receive requests + + // For notifications, we don't send a response, just return Stop + return filter.Stop +} + +// handleResourceRead handles the resources/read method +func (f *MCPServerFilter) handleResourceRead(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling resources/read") + + // Parse request parameters + paramsBytes, err := json.Marshal(req.Params) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + var params struct { + URI string `json:"uri"` + } + if err := json.Unmarshal(paramsBytes, ¶ms); err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to parse resource read params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + // Find resource (by URI) + resource, exists := f.registry.GetResourceByURI(params.URI) + if !exists { + logger.Warnf("[dubbo-go-pixiu] mcp server resource not found: %s", params.URI) + return f.errorHandler.SendInternalError(ctx, req.ID, fmt.Sprintf("resource not found: %s", params.URI)) + } + + // Build resource content response + // TODO: Implement actual resource content loading from source + content := fmt.Sprintf("Resource content for %s (source: %s)", resource.URI, resource.Source.Type) + + result := map[string]any{ + "contents": []map[string]any{ + { + "uri": resource.URI, + "mimeType": resource.MIMEType, + "text": content, + }, + }, + } + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleResourceTemplatesList handles the resources/templates/list method +func (f *MCPServerFilter) handleResourceTemplatesList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Get all resource templates (parameterized resource patterns) + mcpResourceTemplates, err := f.registry.ToMCPResourceTemplates() + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to get MCP resource templates: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to get resource templates") + } + + // Build resource templates list response + result := map[string]any{ + "resourceTemplates": mcpResourceTemplates, + } + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handlePromptsList handles the prompts/list method +func (f *MCPServerFilter) handlePromptsList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling prompts/list request") + + // Get all prompts + mcpPrompts, err := f.registry.ToMCPPrompts() + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to get prompts: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to get prompts") + } + + // Build prompts list response + result := map[string]any{ + "prompts": mcpPrompts, + } + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handlePromptsGet handles the prompts/get method +func (f *MCPServerFilter) handlePromptsGet(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling prompts/get request") + + // Parse request parameters + paramsBytes, err := json.Marshal(req.Params) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + var params struct { + Name string `json:"name"` + Arguments map[string]any `json:"arguments,omitempty"` + } + if err := json.Unmarshal(paramsBytes, ¶ms); err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to parse prompts/get params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + // Find prompt configuration + promptConfig, exists := f.registry.GetPrompt(params.Name) + if !exists { + logger.Warnf("[dubbo-go-pixiu] mcp server prompt not found: %s", params.Name) + return f.errorHandler.SendInternalError(ctx, req.ID, fmt.Sprintf("prompt not found: %s", params.Name)) + } + + // Build prompt messages with parameter replacement + messages, err := f.buildPromptMessages(promptConfig, params.Arguments) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to build prompt messages: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to build prompt messages") + } + + // Build prompts/get response + result := map[string]any{ + "description": promptConfig.Description, + "messages": messages, + } + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// buildPromptMessages builds prompt messages with parameter replacement support +func (f *MCPServerFilter) buildPromptMessages(promptConfig PromptConfig, arguments map[string]any) ([]map[string]any, error) { + messages := make([]map[string]any, 0, len(promptConfig.Messages)) + + for _, msg := range promptConfig.Messages { + // Replace parameter placeholders in content + content := f.replacePromptArguments(msg.Content, arguments) + + message := map[string]any{ + "role": msg.Role, + "content": content, + } + messages = append(messages, message) + } + + return messages, nil +} + +// replacePromptArguments replaces parameter placeholders in prompt content +func (f *MCPServerFilter) replacePromptArguments(content string, arguments map[string]any) string { + if arguments == nil { + return content + } + + result := content + for key, value := range arguments { + placeholder := fmt.Sprintf("{{%s}}", key) + replacement := fmt.Sprintf("%v", value) + result = strings.ReplaceAll(result, placeholder, replacement) + } + + return result +} + +// handleToolCall handles tool call requests by forwarding to backend +func (f *MCPServerFilter) handleToolCall(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Parse tool call parameters + paramsBytes, err := json.Marshal(req.Params) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal tool call params: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "invalid tool call parameters") + } + + var params struct { + Name string `json:"name"` + Arguments map[string]any `json:"arguments,omitempty"` + } + if err := json.Unmarshal(paramsBytes, ¶ms); err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to parse tool call params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid tool call parameters") + } + + // Find tool configuration + toolConfig, exists := f.registry.GetTool(params.Name) + if !exists { + logger.Warnf("[dubbo-go-pixiu] mcp server tool not found: %s", params.Name) + return f.errorHandler.SendToolCallError(ctx, req.ID, fmt.Sprintf("tool not found: %s", params.Name)) + } + + // Build backend request + err = f.buildBackendRequest(ctx, toolConfig, params.Arguments) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to build backend request: %v", err) + return f.errorHandler.SendToolCallError(ctx, req.ID, "failed to build backend request") + } + + // Set cluster information for routing + if ctx.Params == nil { + ctx.Params = make(map[string]any) + } + + logger.Infof("[dubbo-go-pixiu] mcp server forwarding tool call: %s -> %s %s (cluster: %s)", + params.Name, toolConfig.Request.Method, ctx.Request.URL.Path, toolConfig.Cluster) + + // Store MCP data for Encode stage processing + ctx.StoreMCPDataInParams() + + ctx.Route = &model.RouteAction{ + Cluster: toolConfig.Cluster, + } + + // Continue to next filter for backend forwarding + return filter.Continue +} + +// buildBackendRequest builds the complete backend request including path, body, and headers +func (f *MCPServerFilter) buildBackendRequest(ctx *MCPContext, toolConfig ToolConfig, arguments map[string]any) error { + // Set HTTP method + ctx.Request.Method = toolConfig.Request.Method + + // Build request path and body based on argument locations + path := toolConfig.Request.Path + bodyParams := make(map[string]any) + queryParams := make(map[string]string) + + // Process arguments based on their location (path, query, body) + if arguments != nil { + for argName, argValue := range arguments { + // Find argument configuration + var argConfig *ArgConfig + for _, arg := range toolConfig.Args { + if arg.Name == argName { + argConfig = &arg + break + } + } + + if argConfig == nil { + continue // Skip unknown arguments + } + + switch argConfig.In { + case "path": + // Replace path parameters + placeholder := fmt.Sprintf("{%s}", argName) + replacement := fmt.Sprintf("%v", argValue) + path = strings.ReplaceAll(path, placeholder, replacement) + + case "query": + // Add to query parameters + queryParams[argName] = fmt.Sprintf("%v", argValue) + + case "body": + // Add to request body + bodyParams[argName] = argValue + } + } + } + + // Set the request path + ctx.Request.URL.Path = path + + // Add query parameters + if len(queryParams) > 0 { + query := ctx.Request.URL.Query() + for key, value := range queryParams { + query.Set(key, value) + } + ctx.Request.URL.RawQuery = query.Encode() + } + + // Build request body for POST/PUT requests + if len(bodyParams) > 0 && (toolConfig.Request.Method == constant.Post || toolConfig.Request.Method == constant.Put) { + bodyJSON, err := json.Marshal(bodyParams) + if err != nil { + return fmt.Errorf("failed to marshal request body: %v", err) + } + + // Set request body + ctx.Request.Body = io.NopCloser(strings.NewReader(string(bodyJSON))) + ctx.Request.ContentLength = int64(len(bodyJSON)) Review Comment: Setting ContentLength manually can cause issues with HTTP/1.1 chunked transfer encoding. Consider letting the HTTP library handle content length automatically or use io.LimitReader for safer handling. ```suggestion ``` ########## pkg/filter/mcp/mcpserver/config.go: ########## @@ -0,0 +1,217 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 mcpserver + +import ( + "regexp" +) + +type ( + // Config MCP Server Filter configuration + Config struct { + ServerInfo ServerInfo `yaml:"server_info" json:"server_info"` + Endpoint string `yaml:"endpoint" json:"endpoint" default:"/mcp"` + Tools []ToolConfig `yaml:"tools,omitempty" json:"tools,omitempty"` + Resources []ResourceConfig `yaml:"resources,omitempty" json:"resources,omitempty"` + ResourceTemplates []ResourceTemplateConfig `yaml:"resource_templates,omitempty" json:"resource_templates,omitempty"` + Prompts []PromptConfig `yaml:"prompts,omitempty" json:"prompts,omitempty"` + } + + // ServerInfo server information + ServerInfo struct { + Name string `yaml:"name" json:"name" default:"Pixiu MCP Server"` + Version string `yaml:"version" json:"version" default:"1.0.0"` + Description string `yaml:"description,omitempty" json:"description,omitempty" default:"MCP Server powered by Apache Dubbo-go-pixiu"` + Instructions string `yaml:"instructions,omitempty" json:"instructions,omitempty" default:"Use the provided tools to interact with backend services."` + } + + // ToolConfig tool configuration + ToolConfig struct { + Name string `yaml:"name" json:"name"` + Description string `yaml:"description" json:"description"` + Cluster string `yaml:"cluster" json:"cluster"` + Request RequestConfig `yaml:"request" json:"request"` + Args []ArgConfig `yaml:"args,omitempty" json:"args,omitempty"` + } + + // RequestConfig request configuration + RequestConfig struct { + Method string `yaml:"method" json:"method" default:"GET"` + Path string `yaml:"path" json:"path"` + Headers map[string]string `yaml:"headers,omitempty" json:"headers,omitempty"` + Timeout string `yaml:"timeout,omitempty" json:"timeout,omitempty" default:"30s"` + } + + // ArgConfig parameter configuration (simplified) + ArgConfig struct { + Name string `yaml:"name" json:"name"` + Type string `yaml:"type" json:"type" default:"string"` + In string `yaml:"in" json:"in"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` + Required bool `yaml:"required,omitempty" json:"required,omitempty"` + Default any `yaml:"default,omitempty" json:"default,omitempty"` + Enum []string `yaml:"enum,omitempty" json:"enum,omitempty"` + } + + // ResourceConfig resource configuration + ResourceConfig struct { + Name string `yaml:"name" json:"name"` + URI string `yaml:"uri" json:"uri"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` + MIMEType string `yaml:"mime_type,omitempty" json:"mime_type,omitempty"` + Source ResourceSource `yaml:"source" json:"source"` + } + + // ResourceSource resource source configuration (simplified) + ResourceSource struct { + Type string `yaml:"type" json:"type"` + Path string `yaml:"path,omitempty" json:"path,omitempty"` // for file type + URL string `yaml:"url,omitempty" json:"url,omitempty"` // for url type + Content string `yaml:"content,omitempty" json:"content,omitempty"` // for inline type + Template string `yaml:"template,omitempty" json:"template,omitempty"` // for template type + } + + // ResourceTemplateConfig resource template configuration + ResourceTemplateConfig struct { + Name string `yaml:"name" json:"name"` + URITemplate string `yaml:"uri_template" json:"uri_template"` + Title string `yaml:"title,omitempty" json:"title,omitempty"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` + MIMEType string `yaml:"mime_type,omitempty" json:"mime_type,omitempty"` + Parameters []ResourceTemplateParameter `yaml:"parameters,omitempty" json:"parameters,omitempty"` + Annotations *ResourceTemplateAnnotations `yaml:"annotations,omitempty" json:"annotations,omitempty"` + } + + // ResourceTemplateParameter resource template parameter + ResourceTemplateParameter struct { + Name string `yaml:"name" json:"name"` + Type string `yaml:"type" json:"type" default:"string"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` + Required bool `yaml:"required,omitempty" json:"required,omitempty" default:"false"` + Enum []string `yaml:"enum,omitempty" json:"enum,omitempty"` + Default any `yaml:"default,omitempty" json:"default,omitempty"` + } + + // ResourceTemplateAnnotations resource template annotations + ResourceTemplateAnnotations struct { + Audience []string `yaml:"audience,omitempty" json:"audience,omitempty"` + Priority *float64 `yaml:"priority,omitempty" json:"priority,omitempty"` + LastModified string `yaml:"last_modified,omitempty" json:"last_modified,omitempty"` + } + + // ComputedParameter computed parameter (for internal processing, simplified) + ComputedParameter struct { + Name string + Type string + In string + Description string + Required bool + Enum []string + Default any + } +) + +// GetAllParameters gets all parameters of the tool +func (tc *ToolConfig) GetAllParameters() ([]ComputedParameter, error) { + var allParams []ComputedParameter + + // 1. Automatically extract path parameters + pathParams := GetPathParameterNames(tc.Request.Path) + for _, paramName := range pathParams { + // Find corresponding arg configuration + var argConfig *ArgConfig + for _, arg := range tc.Args { + if arg.Name == paramName && arg.In == "path" { + argConfig = &arg + break + } + } + + // Create computed parameter + computed := ComputedParameter{ + Name: paramName, + Type: "string", // Default type + In: "path", + Required: true, // Path parameters are always required + } + + // Apply arg configuration + if argConfig != nil { + computed.Type = argConfig.Type + computed.Description = argConfig.Description + // Simplified: removed Pattern and Format fields + } + + allParams = append(allParams, computed) + } + + // 2. Add non-path parameters + for _, arg := range tc.Args { + if arg.In != "path" { + computed := ComputedParameter{ + Name: arg.Name, + Type: arg.Type, + In: arg.In, + Description: arg.Description, + Required: arg.Required, + Enum: arg.Enum, + Default: arg.Default, + // Simplified: removed complex validation fields + } + allParams = append(allParams, computed) + } + } + + return allParams, nil +} + +// GetPathParameterNames gets all parameter names in the path template +func GetPathParameterNames(pathTemplate string) []string { + re := regexp.MustCompile(`\{([^}]+)}`) + matches := re.FindAllStringSubmatch(pathTemplate, -1) + + // Initialize as empty slice instead of nil + names := []string{} Review Comment: [nitpick] Initialize slice with make([]string, 0) or var names []string instead of []string{} for better clarity when creating an empty slice that will be appended to. ```suggestion names := make([]string, 0) ``` ########## pkg/filter/mcp/mcpserver/response.go: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 mcpserver + +import ( + "encoding/json" + "fmt" + "net/http" + "sync" +) + +import ( + "github.com/mark3labs/mcp-go/mcp" +) + +import ( + "github.com/apache/dubbo-go-pixiu/pkg/common/constant" + "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter" + "github.com/apache/dubbo-go-pixiu/pkg/logger" +) + +// ResponseBuilder provides methods to create standardized MCP responses +type ResponseBuilder struct{} + +var ( + responseBuilderInstance *ResponseBuilder + responseBuilderOnce sync.Once +) + +// GetResponseBuilder returns the singleton ResponseBuilder instance +func GetResponseBuilder() *ResponseBuilder { + responseBuilderOnce.Do(func() { + responseBuilderInstance = &ResponseBuilder{} + }) + return responseBuilderInstance +} + +// Success creates a successful JSON-RPC response +func (rb *ResponseBuilder) Success(id any, result any) mcp.JSONRPCResponse { + return mcp.JSONRPCResponse{ + JSONRPC: mcp.JSONRPC_VERSION, + ID: mcp.NewRequestId(id), + Result: result, + } +} + +// Error creates an error JSON-RPC response +func (rb *ResponseBuilder) Error(id any, code int, message string) mcp.JSONRPCError { + return mcp.NewJSONRPCError(mcp.NewRequestId(id), code, message, nil) +} + +// ToolCallSuccess creates a successful tool call response +func (rb *ResponseBuilder) ToolCallSuccess(id any, content string) mcp.JSONRPCResponse { + // Use mcp-go API to create text content + textContent := mcp.NewTextContent(content) + + // Build MCP tool call result using mcp-go structures + result := mcp.CallToolResult{ + Content: []mcp.Content{textContent}, + IsError: false, + } + + return rb.Success(id, result) +} + +// ToolCallError creates an error tool call response +func (rb *ResponseBuilder) ToolCallError(id any, message string) mcp.JSONRPCResponse { + errorText := fmt.Sprintf("Error: %s", message) + textContent := mcp.NewTextContent(errorText) + + // Build MCP tool call error result using mcp-go structures + result := mcp.CallToolResult{ + Content: []mcp.Content{textContent}, + IsError: true, + } + + return rb.Success(id, result) +} + +// ErrorHandler provides centralized error handling for MCP responses +type ErrorHandler struct { + responseBuilder *ResponseBuilder +} + +var ( + errorHandlerInstance *ErrorHandler + errorHandlerOnce sync.Once +) + +// GetErrorHandler returns the singleton ErrorHandler instance +func GetErrorHandler() *ErrorHandler { + errorHandlerOnce.Do(func() { + errorHandlerInstance = &ErrorHandler{ + responseBuilder: GetResponseBuilder(), + } + }) + return errorHandlerInstance +} + +// SendInternalError sends an internal server error response +func (eh *ErrorHandler) SendInternalError(ctx *MCPContext, id any, message string) filter.FilterStatus { + response := eh.responseBuilder.Error(id, mcp.INTERNAL_ERROR, message) + return eh.sendResponse(ctx, response) +} + +// SendMethodNotFound sends a method not found error response +func (eh *ErrorHandler) SendMethodNotFound(ctx *MCPContext, id any) filter.FilterStatus { + response := eh.responseBuilder.Error(id, mcp.METHOD_NOT_FOUND, "Method not found") + return eh.sendResponse(ctx, response) +} + +// SendInvalidParams sends an invalid parameters error response +func (eh *ErrorHandler) SendInvalidParams(ctx *MCPContext, id any, message string) filter.FilterStatus { + response := eh.responseBuilder.Error(id, mcp.INVALID_PARAMS, fmt.Sprintf("Invalid params: %s", message)) + return eh.sendResponse(ctx, response) +} + +// SendToolCallError sends a tool call error response +func (eh *ErrorHandler) SendToolCallError(ctx *MCPContext, id any, message string) filter.FilterStatus { + response := eh.responseBuilder.ToolCallError(id, message) + return eh.sendResponse(ctx, response) +} + +// sendResponse sends any response and handles Content-Length cleanup +func (eh *ErrorHandler) sendResponse(ctx *MCPContext, response any) filter.FilterStatus { + responseBody, err := json.Marshal(response) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal response: %v", err) + ctx.SendLocalReply(http.StatusInternalServerError, []byte("internal server error")) + return filter.Stop + } + + // Critical: Clear Content-Length header to prevent mismatch errors + ctx.Writer.Header().Del(constant.HeaderKeyContentLength) Review Comment: This pattern of deleting Content-Length header appears multiple times across the codebase. Consider creating a helper method to encapsulate this logic and reduce code duplication. ```suggestion eh.ClearContentLengthHeader(ctx) ``` ########## pkg/filter/mcp/mcpserver/registry.go: ########## @@ -0,0 +1,373 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 mcpserver + +import ( + "fmt" + "sync" +) + +import ( + "github.com/mark3labs/mcp-go/mcp" +) + +import ( + "github.com/apache/dubbo-go-pixiu/pkg/logger" +) + +// ToolRegistry tool registry, thread-safe (optimized with single indexing) +type ToolRegistry struct { + mu sync.RWMutex + tools map[string]ToolConfig + resources map[string]ResourceConfig // indexed by name only + resourceTemplates map[string]ResourceTemplateConfig // indexed by name + prompts map[string]PromptConfig + + // TODO: Dynamic update support - add when integrating with Nacos + // changeListeners []ChangeListener // change listeners + // nacosClient nacos.ConfigClient // Nacos config client + // serviceDiscovery nacos.NamingClient // Nacos service discovery client +} + +// NewToolRegistry creates a new tool registry +func NewToolRegistry() *ToolRegistry { + return &ToolRegistry{ + tools: make(map[string]ToolConfig), + resources: make(map[string]ResourceConfig), + resourceTemplates: make(map[string]ResourceTemplateConfig), + prompts: make(map[string]PromptConfig), + } +} + +// RegisterTool registers a tool +func (r *ToolRegistry) RegisterTool(tool ToolConfig) error { + r.mu.Lock() + defer r.mu.Unlock() + + if _, exists := r.tools[tool.Name]; exists { + return fmt.Errorf("tool %s already exists", tool.Name) + } + + r.tools[tool.Name] = tool + + // TODO: Dynamic update notification - enable when integrating with Nacos + // r.notifyToolsListChanged() + + return nil +} + +// RegisterResource registers a resource (simplified, no URI duplication check) +func (r *ToolRegistry) RegisterResource(resource ResourceConfig) error { + r.mu.Lock() + defer r.mu.Unlock() + + if _, exists := r.resources[resource.Name]; exists { + return fmt.Errorf("resource %s already exists", resource.Name) + } + + // Note: URI uniqueness is not enforced to simplify implementation + // Multiple resources can have the same URI if needed + r.resources[resource.Name] = resource + return nil +} + +// GetTool gets tool configuration +func (r *ToolRegistry) GetTool(name string) (ToolConfig, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + + tool, exists := r.tools[name] + return tool, exists +} + +// GetResource gets resource configuration (by name) +func (r *ToolRegistry) GetResource(name string) (ResourceConfig, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + + resource, exists := r.resources[name] + return resource, exists +} + +// GetResourceByURI gets resource configuration (by URI, using linear search) +func (r *ToolRegistry) GetResourceByURI(uri string) (ResourceConfig, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + + // Linear search through resources to find matching URI + for _, resource := range r.resources { + if resource.URI == uri { + return resource, true + } Review Comment: Linear search through all resources to find matching URI could be inefficient with large numbers of resources. Consider adding a URI-based index or using a more efficient lookup data structure. ```suggestion resource, exists := r.resourcesByURI[uri] if exists { return resource, true ``` ########## pkg/filter/mcp/mcpserver/handlers.go: ########## @@ -0,0 +1,561 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 mcpserver + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "strings" +) + +import ( + "github.com/mark3labs/mcp-go/mcp" +) + +import ( + "github.com/apache/dubbo-go-pixiu/pkg/client" + "github.com/apache/dubbo-go-pixiu/pkg/common/constant" + "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter" + "github.com/apache/dubbo-go-pixiu/pkg/logger" + "github.com/apache/dubbo-go-pixiu/pkg/model" +) + +// handleInitialize handles the initialize method +func (f *MCPServerFilter) handleInitialize(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Build server capabilities using mcp-go structures + capabilities := mcp.ServerCapabilities{ + Tools: &struct { + ListChanged bool `json:"listChanged,omitempty"` + }{ + // TODO: Dynamic update capabilities - enable after Nacos integration + // Currently set to false, future Nacos integration will support: + // 1. Dynamic discovery and registration of new backend services + // 2. Automatic generation of corresponding MCP tools + // 3. Send notifications/tools/list_changed notifications + ListChanged: false, + }, + Resources: &struct { + Subscribe bool `json:"subscribe,omitempty"` + ListChanged bool `json:"listChanged,omitempty"` + }{ + Subscribe: false, + ListChanged: false, + }, + Prompts: &struct { + ListChanged bool `json:"listChanged,omitempty"` + }{ + ListChanged: false, + }, + } + + // Build server info using mcp-go structures + serverInfo := mcp.Implementation{ + Name: f.cfg.ServerInfo.Name, + Version: f.cfg.ServerInfo.Version, + } + + // Create initialization result using mcp-go API + instructions := f.cfg.ServerInfo.Instructions + if instructions == "" { + instructions = "This MCP server provides API access through tools, documentation through resources, and AI assistance through prompts." + } + result := mcp.NewInitializeResult(mcp.LATEST_PROTOCOL_VERSION, capabilities, serverInfo, instructions) + + // Create JSON-RPC response + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleToolsList handles the tools/list method using mcp-go APIs +func (f *MCPServerFilter) handleToolsList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + tools := make([]mcp.Tool, 0, len(f.cfg.Tools)) + + // Build tools using mcp-go API for standard compliance + for _, toolCfg := range f.cfg.Tools { + // Start with basic tool options + toolOptions := []mcp.ToolOption{ + mcp.WithDescription(toolCfg.Description), + } + + // Add parameter definitions using mcp-go APIs + for _, arg := range toolCfg.Args { + switch arg.Type { + case "string": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + if defaultStr, ok := arg.Default.(string); ok { + opts = append(opts, mcp.DefaultString(defaultStr)) + } + } + if len(arg.Enum) > 0 { + opts = append(opts, mcp.Enum(arg.Enum...)) + } + toolOptions = append(toolOptions, mcp.WithString(arg.Name, opts...)) + + case "integer", "number": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + // Handle both int and float64 types from YAML parsing + switch defaultVal := arg.Default.(type) { + case float64: + opts = append(opts, mcp.DefaultNumber(defaultVal)) + case int: + opts = append(opts, mcp.DefaultNumber(float64(defaultVal))) + case int64: + opts = append(opts, mcp.DefaultNumber(float64(defaultVal))) + } + } + toolOptions = append(toolOptions, mcp.WithNumber(arg.Name, opts...)) + + case "boolean": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + if defaultBool, ok := arg.Default.(bool); ok { + opts = append(opts, mcp.DefaultBool(defaultBool)) + } + } + toolOptions = append(toolOptions, mcp.WithBoolean(arg.Name, opts...)) + } + } + + // Create tool using mcp-go API + tool := mcp.NewTool(toolCfg.Name, toolOptions...) + tools = append(tools, tool) + } + + // Build standard MCP tools list response using mcp-go structures + result := mcp.NewListToolsResult(tools, "") // empty cursor for no pagination + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleResourcesList handles the resources/list method +func (f *MCPServerFilter) handleResourcesList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Get all resources + mcpResources, err := f.registry.ToMCPResources() + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to get MCP resources: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to get resources") + } + + // Build resources list response using mcp-go structures + result := mcp.NewListResourcesResult(mcpResources, "") // empty cursor for no pagination + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handlePing handles the ping method +func (f *MCPServerFilter) handlePing(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling ping request") + + // Simple ping response + result := map[string]any{} + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleNotificationsInitialized handles notifications/initialized notification +func (f *MCPServerFilter) handleNotificationsInitialized(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server received initialized notification from client") + + // Store client initialization state + // This notification indicates that the client has completed initialization + // and is ready to receive requests + + // For notifications, we don't send a response, just return Stop + return filter.Stop +} + +// handleResourceRead handles the resources/read method +func (f *MCPServerFilter) handleResourceRead(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling resources/read") + + // Parse request parameters + paramsBytes, err := json.Marshal(req.Params) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + var params struct { + URI string `json:"uri"` + } + if err := json.Unmarshal(paramsBytes, ¶ms); err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to parse resource read params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + // Find resource (by URI) + resource, exists := f.registry.GetResourceByURI(params.URI) + if !exists { + logger.Warnf("[dubbo-go-pixiu] mcp server resource not found: %s", params.URI) + return f.errorHandler.SendInternalError(ctx, req.ID, fmt.Sprintf("resource not found: %s", params.URI)) + } + + // Build resource content response + // TODO: Implement actual resource content loading from source + content := fmt.Sprintf("Resource content for %s (source: %s)", resource.URI, resource.Source.Type) Review Comment: This appears to be placeholder content with a TODO comment. Consider implementing actual resource content loading or adding a clear indication that this is a stub implementation. ########## pkg/filter/mcp/mcpserver/handlers.go: ########## @@ -0,0 +1,561 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 mcpserver + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "strings" +) + +import ( + "github.com/mark3labs/mcp-go/mcp" +) + +import ( + "github.com/apache/dubbo-go-pixiu/pkg/client" + "github.com/apache/dubbo-go-pixiu/pkg/common/constant" + "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter" + "github.com/apache/dubbo-go-pixiu/pkg/logger" + "github.com/apache/dubbo-go-pixiu/pkg/model" +) + +// handleInitialize handles the initialize method +func (f *MCPServerFilter) handleInitialize(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Build server capabilities using mcp-go structures + capabilities := mcp.ServerCapabilities{ + Tools: &struct { + ListChanged bool `json:"listChanged,omitempty"` + }{ + // TODO: Dynamic update capabilities - enable after Nacos integration + // Currently set to false, future Nacos integration will support: + // 1. Dynamic discovery and registration of new backend services + // 2. Automatic generation of corresponding MCP tools + // 3. Send notifications/tools/list_changed notifications + ListChanged: false, + }, + Resources: &struct { + Subscribe bool `json:"subscribe,omitempty"` + ListChanged bool `json:"listChanged,omitempty"` + }{ + Subscribe: false, + ListChanged: false, + }, + Prompts: &struct { + ListChanged bool `json:"listChanged,omitempty"` + }{ + ListChanged: false, + }, + } + + // Build server info using mcp-go structures + serverInfo := mcp.Implementation{ + Name: f.cfg.ServerInfo.Name, + Version: f.cfg.ServerInfo.Version, + } + + // Create initialization result using mcp-go API + instructions := f.cfg.ServerInfo.Instructions + if instructions == "" { + instructions = "This MCP server provides API access through tools, documentation through resources, and AI assistance through prompts." + } + result := mcp.NewInitializeResult(mcp.LATEST_PROTOCOL_VERSION, capabilities, serverInfo, instructions) + + // Create JSON-RPC response + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleToolsList handles the tools/list method using mcp-go APIs +func (f *MCPServerFilter) handleToolsList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + tools := make([]mcp.Tool, 0, len(f.cfg.Tools)) + + // Build tools using mcp-go API for standard compliance + for _, toolCfg := range f.cfg.Tools { + // Start with basic tool options + toolOptions := []mcp.ToolOption{ + mcp.WithDescription(toolCfg.Description), + } + + // Add parameter definitions using mcp-go APIs + for _, arg := range toolCfg.Args { + switch arg.Type { + case "string": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + if defaultStr, ok := arg.Default.(string); ok { + opts = append(opts, mcp.DefaultString(defaultStr)) + } + } + if len(arg.Enum) > 0 { + opts = append(opts, mcp.Enum(arg.Enum...)) + } + toolOptions = append(toolOptions, mcp.WithString(arg.Name, opts...)) + + case "integer", "number": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + // Handle both int and float64 types from YAML parsing + switch defaultVal := arg.Default.(type) { + case float64: + opts = append(opts, mcp.DefaultNumber(defaultVal)) + case int: + opts = append(opts, mcp.DefaultNumber(float64(defaultVal))) + case int64: + opts = append(opts, mcp.DefaultNumber(float64(defaultVal))) + } + } + toolOptions = append(toolOptions, mcp.WithNumber(arg.Name, opts...)) + + case "boolean": + opts := []mcp.PropertyOption{mcp.Description(arg.Description)} + if arg.Required { + opts = append(opts, mcp.Required()) + } + if arg.Default != nil { + if defaultBool, ok := arg.Default.(bool); ok { + opts = append(opts, mcp.DefaultBool(defaultBool)) + } + } + toolOptions = append(toolOptions, mcp.WithBoolean(arg.Name, opts...)) + } + } + + // Create tool using mcp-go API + tool := mcp.NewTool(toolCfg.Name, toolOptions...) + tools = append(tools, tool) + } + + // Build standard MCP tools list response using mcp-go structures + result := mcp.NewListToolsResult(tools, "") // empty cursor for no pagination + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleResourcesList handles the resources/list method +func (f *MCPServerFilter) handleResourcesList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Get all resources + mcpResources, err := f.registry.ToMCPResources() + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to get MCP resources: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to get resources") + } + + // Build resources list response using mcp-go structures + result := mcp.NewListResourcesResult(mcpResources, "") // empty cursor for no pagination + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handlePing handles the ping method +func (f *MCPServerFilter) handlePing(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling ping request") + + // Simple ping response + result := map[string]any{} + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleNotificationsInitialized handles notifications/initialized notification +func (f *MCPServerFilter) handleNotificationsInitialized(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server received initialized notification from client") + + // Store client initialization state + // This notification indicates that the client has completed initialization + // and is ready to receive requests + + // For notifications, we don't send a response, just return Stop + return filter.Stop +} + +// handleResourceRead handles the resources/read method +func (f *MCPServerFilter) handleResourceRead(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling resources/read") + + // Parse request parameters + paramsBytes, err := json.Marshal(req.Params) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + var params struct { + URI string `json:"uri"` + } + if err := json.Unmarshal(paramsBytes, ¶ms); err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to parse resource read params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + // Find resource (by URI) + resource, exists := f.registry.GetResourceByURI(params.URI) + if !exists { + logger.Warnf("[dubbo-go-pixiu] mcp server resource not found: %s", params.URI) + return f.errorHandler.SendInternalError(ctx, req.ID, fmt.Sprintf("resource not found: %s", params.URI)) + } + + // Build resource content response + // TODO: Implement actual resource content loading from source + content := fmt.Sprintf("Resource content for %s (source: %s)", resource.URI, resource.Source.Type) + + result := map[string]any{ + "contents": []map[string]any{ + { + "uri": resource.URI, + "mimeType": resource.MIMEType, + "text": content, + }, + }, + } + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handleResourceTemplatesList handles the resources/templates/list method +func (f *MCPServerFilter) handleResourceTemplatesList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Get all resource templates (parameterized resource patterns) + mcpResourceTemplates, err := f.registry.ToMCPResourceTemplates() + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to get MCP resource templates: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to get resource templates") + } + + // Build resource templates list response + result := map[string]any{ + "resourceTemplates": mcpResourceTemplates, + } + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handlePromptsList handles the prompts/list method +func (f *MCPServerFilter) handlePromptsList(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling prompts/list request") + + // Get all prompts + mcpPrompts, err := f.registry.ToMCPPrompts() + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to get prompts: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to get prompts") + } + + // Build prompts list response + result := map[string]any{ + "prompts": mcpPrompts, + } + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// handlePromptsGet handles the prompts/get method +func (f *MCPServerFilter) handlePromptsGet(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling prompts/get request") + + // Parse request parameters + paramsBytes, err := json.Marshal(req.Params) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + var params struct { + Name string `json:"name"` + Arguments map[string]any `json:"arguments,omitempty"` + } + if err := json.Unmarshal(paramsBytes, ¶ms); err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to parse prompts/get params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid parameters") + } + + // Find prompt configuration + promptConfig, exists := f.registry.GetPrompt(params.Name) + if !exists { + logger.Warnf("[dubbo-go-pixiu] mcp server prompt not found: %s", params.Name) + return f.errorHandler.SendInternalError(ctx, req.ID, fmt.Sprintf("prompt not found: %s", params.Name)) + } + + // Build prompt messages with parameter replacement + messages, err := f.buildPromptMessages(promptConfig, params.Arguments) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to build prompt messages: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "failed to build prompt messages") + } + + // Build prompts/get response + result := map[string]any{ + "description": promptConfig.Description, + "messages": messages, + } + + response := f.responseBuilder.Success(req.ID, result) + return f.sendJSONResponse(ctx, response) +} + +// buildPromptMessages builds prompt messages with parameter replacement support +func (f *MCPServerFilter) buildPromptMessages(promptConfig PromptConfig, arguments map[string]any) ([]map[string]any, error) { + messages := make([]map[string]any, 0, len(promptConfig.Messages)) + + for _, msg := range promptConfig.Messages { + // Replace parameter placeholders in content + content := f.replacePromptArguments(msg.Content, arguments) + + message := map[string]any{ + "role": msg.Role, + "content": content, + } + messages = append(messages, message) + } + + return messages, nil +} + +// replacePromptArguments replaces parameter placeholders in prompt content +func (f *MCPServerFilter) replacePromptArguments(content string, arguments map[string]any) string { + if arguments == nil { + return content + } + + result := content + for key, value := range arguments { + placeholder := fmt.Sprintf("{{%s}}", key) + replacement := fmt.Sprintf("%v", value) + result = strings.ReplaceAll(result, placeholder, replacement) + } + + return result +} + +// handleToolCall handles tool call requests by forwarding to backend +func (f *MCPServerFilter) handleToolCall(ctx *MCPContext, req mcp.JSONRPCRequest) filter.FilterStatus { + // Parse tool call parameters + paramsBytes, err := json.Marshal(req.Params) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal tool call params: %v", err) + return f.errorHandler.SendInternalError(ctx, req.ID, "invalid tool call parameters") + } + + var params struct { + Name string `json:"name"` + Arguments map[string]any `json:"arguments,omitempty"` + } + if err := json.Unmarshal(paramsBytes, ¶ms); err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to parse tool call params: %v", err) + return f.errorHandler.SendInvalidParams(ctx, req.ID, "invalid tool call parameters") + } + + // Find tool configuration + toolConfig, exists := f.registry.GetTool(params.Name) + if !exists { + logger.Warnf("[dubbo-go-pixiu] mcp server tool not found: %s", params.Name) + return f.errorHandler.SendToolCallError(ctx, req.ID, fmt.Sprintf("tool not found: %s", params.Name)) + } + + // Build backend request + err = f.buildBackendRequest(ctx, toolConfig, params.Arguments) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to build backend request: %v", err) + return f.errorHandler.SendToolCallError(ctx, req.ID, "failed to build backend request") + } + + // Set cluster information for routing + if ctx.Params == nil { + ctx.Params = make(map[string]any) + } + + logger.Infof("[dubbo-go-pixiu] mcp server forwarding tool call: %s -> %s %s (cluster: %s)", + params.Name, toolConfig.Request.Method, ctx.Request.URL.Path, toolConfig.Cluster) + + // Store MCP data for Encode stage processing + ctx.StoreMCPDataInParams() + + ctx.Route = &model.RouteAction{ + Cluster: toolConfig.Cluster, + } + + // Continue to next filter for backend forwarding + return filter.Continue +} + +// buildBackendRequest builds the complete backend request including path, body, and headers +func (f *MCPServerFilter) buildBackendRequest(ctx *MCPContext, toolConfig ToolConfig, arguments map[string]any) error { + // Set HTTP method + ctx.Request.Method = toolConfig.Request.Method + + // Build request path and body based on argument locations + path := toolConfig.Request.Path + bodyParams := make(map[string]any) + queryParams := make(map[string]string) + + // Process arguments based on their location (path, query, body) + if arguments != nil { + for argName, argValue := range arguments { + // Find argument configuration + var argConfig *ArgConfig + for _, arg := range toolConfig.Args { + if arg.Name == argName { + argConfig = &arg + break + } + } + + if argConfig == nil { + continue // Skip unknown arguments + } + + switch argConfig.In { + case "path": + // Replace path parameters + placeholder := fmt.Sprintf("{%s}", argName) + replacement := fmt.Sprintf("%v", argValue) + path = strings.ReplaceAll(path, placeholder, replacement) + + case "query": + // Add to query parameters + queryParams[argName] = fmt.Sprintf("%v", argValue) + + case "body": + // Add to request body + bodyParams[argName] = argValue + } + } + } + + // Set the request path + ctx.Request.URL.Path = path + + // Add query parameters + if len(queryParams) > 0 { + query := ctx.Request.URL.Query() + for key, value := range queryParams { + query.Set(key, value) + } + ctx.Request.URL.RawQuery = query.Encode() + } + + // Build request body for POST/PUT requests + if len(bodyParams) > 0 && (toolConfig.Request.Method == constant.Post || toolConfig.Request.Method == constant.Put) { + bodyJSON, err := json.Marshal(bodyParams) + if err != nil { + return fmt.Errorf("failed to marshal request body: %v", err) + } + + // Set request body + ctx.Request.Body = io.NopCloser(strings.NewReader(string(bodyJSON))) + ctx.Request.ContentLength = int64(len(bodyJSON)) + + // Set Content-Type header + ctx.Request.Header.Set(constant.HeaderKeyContextType, constant.HeaderValueApplicationJson) + + logger.Debugf("[dubbo-go-pixiu] mcp server built request body: %s", string(bodyJSON)) + } + + return nil +} + +// handleToolCallResponse handles tool call responses, wrapping backend responses in MCP format +func (f *MCPServerFilter) handleToolCallResponse(ctx *MCPContext) filter.FilterStatus { + logger.Debugf("[dubbo-go-pixiu] mcp server handling tool call response") + + // Extract request information + requestID := ctx.GetMCPRequestID() + if requestID == nil { + logger.Errorf("[dubbo-go-pixiu] mcp server missing request ID for tool call response") + return filter.Continue + } + + // Extract backend response + responseBody, statusCode, err := f.extractBackendResponse(ctx) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to extract backend response: %v", err) + return f.errorHandler.SendToolCallError(ctx, requestID, "failed to process backend response") + } + + // Process the response + return f.processToolCallResponse(ctx, requestID, responseBody, statusCode) +} + +// extractBackendResponse extracts response data from the context +func (f *MCPServerFilter) extractBackendResponse(ctx *MCPContext) ([]byte, int, error) { + if ctx.TargetResp == nil { + return nil, 0, fmt.Errorf("no target response available") + } + + unaryResp, ok := ctx.TargetResp.(*client.UnaryResponse) + if !ok { + return nil, 0, fmt.Errorf("unexpected response type") + } + + responseBody := unaryResp.Data + statusCode := ctx.GetStatusCode() + + if len(responseBody) == 0 { + return nil, statusCode, fmt.Errorf("empty response body") + } + + logger.Debugf("[dubbo-go-pixiu] mcp server backend response: status=%d, size=%d bytes", statusCode, len(responseBody)) + return responseBody, statusCode, nil +} + +// processToolCallResponse processes the tool call response and sends the result +func (f *MCPServerFilter) processToolCallResponse(ctx *MCPContext, requestID any, responseBody []byte, statusCode int) filter.FilterStatus { + // Check for backend errors + if statusCode >= 400 { + logger.Errorf("[dubbo-go-pixiu] mcp server backend returned error status: %d", statusCode) + return f.errorHandler.SendToolCallError(ctx, requestID, fmt.Sprintf("backend error: %d", statusCode)) + } + + // Build successful response using ToolCallSuccess method + content := strings.TrimSpace(string(responseBody)) + mcpResponse := f.responseBuilder.ToolCallSuccess(requestID, content) + return f.sendMCPResponse(ctx, mcpResponse) +} + +// sendMCPResponse sends an MCP response and updates the target response +func (f *MCPServerFilter) sendMCPResponse(ctx *MCPContext, response mcp.JSONRPCResponse) filter.FilterStatus { + mcpResponseBody, err := json.Marshal(response) + if err != nil { + logger.Errorf("[dubbo-go-pixiu] mcp server failed to marshal MCP response: %v", err) + return filter.Continue + } + + // Override TargetResp to ensure MCP format response is sent + ctx.TargetResp = &client.UnaryResponse{Data: mcpResponseBody} + ctx.StatusCode(http.StatusOK) + ctx.AddHeader(constant.HeaderKeyContextType, constant.HeaderValueApplicationJson) + + // Critical: Clear Content-Length header to prevent mismatch errors + ctx.Writer.Header().Del(constant.HeaderKeyContentLength) Review Comment: Duplicate Content-Length header deletion logic. This should be extracted to a common helper method to maintain consistency and reduce duplication. ```suggestion clearContentLengthHeader(ctx) ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
