MCP Server PoC

cmd/plugin-backend.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"os"
@@ -42,7 +43,7 @@ func main() {
 	}
 	logrus.SetLevel(lvl)
 	log.Infof("Starting %s at log level %s", appVersion, *logLevel)
-
+	log.Infof("Loading config from %s", *configPath)
 	cfg, err := config.ReadFile(buildVersion, buildDate, *configPath)
 	if err != nil {
 		log.WithError(err).Fatal("error reading config file")
@@ -62,11 +63,18 @@ func main() {
 		model.AddFlowLineMapping(decoders.NetworkEventsToString)
 	}
 
-	go server.StartMetrics(&server.MetricsConfig{
-		Port:     cfg.Server.MetricsPort,
-		CertPath: cfg.Server.CertPath,
-		KeyPath:  cfg.Server.KeyPath,
-	})
+	ctx := context.Background()
+	if cfg.Server.MCP.Enable {
+		if err := server.StartMCPServer(ctx, cfg); err != nil && !errors.Is(err, context.Canceled) {
+			panic(err)
+		}
+	} else {
+		go server.StartMetrics(&server.MetricsConfig{
+			Port:     cfg.Server.MetricsPort,
+			CertPath: cfg.Server.CertPath,
+			KeyPath:  cfg.Server.KeyPath,
+		})
 
-	server.Start(context.Background(), cfg, checker)
+		server.Start(ctx, cfg, checker)
+	}
 }

config/sample-config.yaml

@@ -2,6 +2,11 @@ server:
   port: 9001
   metricsPort: 9002
   authCheck: none
+  mcp:
+    enable: true
+    transport: stdio
+    sseAddress: localhost
+    ssePort: 3001
 loki:
   url: http://localhost:3100
   labels:
@@ -18,7 +23,7 @@ loki:
     #    - SrcK8S_Zone
     #    - DstK8S_Zone
   tenantID: netobserv
-  useMocks: false
+  useMocks: true
 prometheus:
   timeout: 30s
   metrics:

go.mod

@@ -7,6 +7,7 @@ toolchain go1.23.5
 require (
 	github.com/gorilla/mux v1.8.1
 	github.com/json-iterator/go v1.1.12
+	github.com/mark3labs/mcp-go v0.16.0
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0
 	github.com/prometheus/client_golang v1.22.0
@@ -51,6 +52,7 @@ require (
 	github.com/spf13/pflag v1.0.6 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
+	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	golang.org/x/net v0.39.0 // indirect
 	golang.org/x/oauth2 v0.29.0 // indirect
 	golang.org/x/sys v0.32.0 // indirect

go.sum

@@ -76,6 +76,8 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/mailru/easyjson v0.9.0 h1:PrnmzHw7262yW8sTBwxi1PdJA3Iw/EKBa8psRf7d9a4=
 github.com/mailru/easyjson v0.9.0/go.mod h1:1+xMtQp2MRNVL/V1bOzuP3aP8VNwRW55fQUto+XFtTU=
+github.com/mark3labs/mcp-go v0.16.0 h1:hNOr0EqhSUra5jm1Wv6+BOynzIa+bMtfP3zgde70MvY=
+github.com/mark3labs/mcp-go v0.16.0/go.mod h1:KmJndYv7GIgcPVwEKJjNcbhVQ+hJGJhrCCB/9xITzpE=
 github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
 github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -153,6 +155,8 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 h1:gEOO8jv9F4OT7lGCjxCBTO/36wtF6j2nSip77qHd4x4=
 github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM=
+github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
+github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=

pkg/config/config.go

@@ -19,6 +19,13 @@ var (
 	log = logrus.WithField("module", "config")
 )
 
+type MCP struct {
+	Enable     bool   `yaml:"enable,omitempty" json:"enable,omitempty"`
+	Transport  string `yaml:"transport,omitempty" json:"transport,omitempty"`
+	SSEAddress string `yaml:"sseAddress,omitempty" json:"sseAddress,omitempty"`
+	SSEPort    int    `yaml:"ssePort,omitempty" json:"ssePort,omitempty"`
+}
+
 type Server struct {
 	Port        int    `yaml:"port,omitempty" json:"port,omitempty"`
 	MetricsPort int    `yaml:"metricsPort,omitempty" json:"metricsPort,omitempty"`
@@ -29,6 +36,7 @@ type Server struct {
 	CORSHeaders string `yaml:"corsHeaders,omitempty" json:"corsHeaders,omitempty"`
 	CORSMaxAge  string `yaml:"corsMaxAge,omitempty" json:"corsMaxAge,omitempty"`
 	AuthCheck   string `yaml:"authCheck,omitempty" json:"authCheck,omitempty"`
+	MCP         MCP    `yaml:"mcp,omitempty" json:"mcp,omitempty"`
 }
 
 type Prometheus struct {

pkg/handler/clients.go

@@ -22,7 +22,7 @@
 	promDev   api.Client
 }
 
-func newClients(cfg *config.Config, requestHeader http.Header, useLokiStatus bool, namespace string) (clients, error) {
+func NewClients(cfg *config.Config, requestHeader http.Header, useLokiStatus bool, namespace string) (clients, error) {
 	lokiClients := newLokiClients(cfg, requestHeader, useLokiStatus)
 	promClients, err := newPromClients(cfg, requestHeader, namespace)
 	return clients{loki: lokiClients.loki, promAdmin: promClients.promAdmin, promDev: promClients.promDev}, err
@@ -49,7 +49,7 @@
 func newLokiClients(cfg *config.Config, requestHeader http.Header, useLokiStatus bool) clients {
 	var lokiClient httpclient.Caller
 	if cfg.IsLokiEnabled() {
-		lokiClient = newLokiClient(&cfg.Loki, requestHeader, useLokiStatus)
+		lokiClient = NewLokiClient(&cfg.Loki, requestHeader, useLokiStatus)
 	}
 	return clients{loki: lokiClient}
 }

pkg/handler/export.go

@@ -22,7 +22,7 @@ func (h *Handlers) ExportFlows(ctx context.Context) func(w http.ResponseWriter,
 			writeError(w, http.StatusBadRequest, "Cannot perform flows query with disabled Loki")
 			return
 		}
-		cl := newLokiClient(&h.Cfg.Loki, r.Header, false)
+		cl := NewLokiClient(&h.Cfg.Loki, r.Header, false)
 		var code int
 		startTime := time.Now()
 		defer func() {
@@ -32,7 +32,7 @@ func (h *Handlers) ExportFlows(ctx context.Context) func(w http.ResponseWriter,
 		params := r.URL.Query()
 		hlog.Debugf("ExportFlows query params: %s", params)
 
-		flows, code, err := h.getFlows(ctx, cl, params)
+		flows, code, err := h.QueryFlows(ctx, cl, params)
 		if err != nil {
 			writeError(w, code, err.Error())
 			return

pkg/handler/flows.go

@@ -34,7 +34,7 @@ func (h *Handlers) GetFlows(ctx context.Context) func(w http.ResponseWriter, r *
 			return
 		}
 
-		cl := newLokiClient(&h.Cfg.Loki, r.Header, false)
+		cl := NewLokiClient(&h.Cfg.Loki, r.Header, false)
 		var code int
 		startTime := time.Now()
 		defer func() {
@@ -44,7 +44,7 @@ func (h *Handlers) GetFlows(ctx context.Context) func(w http.ResponseWriter, r *
 		params := r.URL.Query()
 		hlog.Debugf("GetFlows query params: %s", params)
 
-		flows, code, err := h.getFlows(ctx, cl, params)
+		flows, code, err := h.QueryFlows(ctx, cl, params)
 		if err != nil {
 			writeError(w, code, err.Error())
 			return
@@ -55,7 +55,7 @@ func (h *Handlers) GetFlows(ctx context.Context) func(w http.ResponseWriter, r *
 	}
 }
 
-func (h *Handlers) getFlows(ctx context.Context, lokiClient httpclient.Caller, params url.Values) (*model.AggregatedQueryResponse, int, error) {
+func (h *Handlers) QueryFlows(ctx context.Context, lokiClient httpclient.Caller, params url.Values) (*model.AggregatedQueryResponse, int, error) {
 	start, _, err := getStartTime(params)
 	if err != nil {
 		return nil, http.StatusBadRequest, err

pkg/handler/frontend-config.go

@@ -20,7 +20,7 @@ func (h *Handlers) GetFrontendConfig() func(w http.ResponseWriter, r *http.Reque
 		}
 		if h.Cfg.IsLokiEnabled() {
 			// (Re)load Loki max chunk age
-			lokiClient := newLokiClient(&h.Cfg.Loki, r.Header, true)
+			lokiClient := NewLokiClient(&h.Cfg.Loki, r.Header, true)
 			if maxChunkAge, err := h.fetchIngesterMaxChunkAge(lokiClient); err != nil {
 				// Log the error, but keep returning known config
 				hlog.Errorf("Could not get max chunk age: %v", err)

pkg/handler/loki.go

@@ -39,7 +39,7 @@ const (
 	lokiOrgIDHeader = "X-Scope-OrgID"
 )
 
-func newLokiClient(cfg *config.Loki, requestHeader http.Header, useStatusConfig bool) httpclient.Caller {
+func NewLokiClient(cfg *config.Loki, requestHeader http.Header, useStatusConfig bool) httpclient.Caller {
 	headers := map[string][]string{}
 	if cfg.TenantID != "" {
 		headers[lokiOrgIDHeader] = []string{cfg.TenantID}
@@ -198,7 +198,7 @@ func getLokiNamesForPrefix(cfg *config.Loki, lokiClient httpclient.Caller, filts
 }
 
 func (h *Handlers) getLokiStatus(r *http.Request) ([]byte, int, error) {
-	lokiClient := newLokiClient(&h.Cfg.Loki, r.Header, true)
+	lokiClient := NewLokiClient(&h.Cfg.Loki, r.Header, true)
 	baseURL := strings.TrimRight(h.Cfg.Loki.GetStatusURL(), "/")
 	return executeLokiQuery(fmt.Sprintf("%s/%s", baseURL, "ready"), lokiClient)
 }
@@ -231,7 +231,7 @@ func (h *Handlers) LokiMetrics() func(w http.ResponseWriter, r *http.Request) {
 			writeError(w, http.StatusBadRequest, "Loki is disabled")
 			return
 		}
-		lokiClient := newLokiClient(&h.Cfg.Loki, r.Header, true)
+		lokiClient := NewLokiClient(&h.Cfg.Loki, r.Header, true)
 		baseURL := strings.TrimRight(h.Cfg.Loki.GetStatusURL(), "/")
 
 		resp, code, err := executeLokiQuery(fmt.Sprintf("%s/%s", baseURL, "metrics"), lokiClient)
@@ -250,7 +250,7 @@ func (h *Handlers) LokiBuildInfos() func(w http.ResponseWriter, r *http.Request)
 			writeError(w, http.StatusBadRequest, "Loki is disabled")
 			return
 		}
-		lokiClient := newLokiClient(&h.Cfg.Loki, r.Header, true)
+		lokiClient := NewLokiClient(&h.Cfg.Loki, r.Header, true)
 		baseURL := strings.TrimRight(h.Cfg.Loki.GetStatusURL(), "/")
 
 		resp, code, err := executeLokiQuery(fmt.Sprintf("%s/%s", baseURL, "loki/api/v1/status/buildinfo"), lokiClient)
@@ -293,7 +293,7 @@ func (h *Handlers) LokiLimits() func(w http.ResponseWriter, r *http.Request) {
 			writeJSON(w, http.StatusNoContent, "Loki is disabled")
 			return
 		}
-		lokiClient := newLokiClient(&h.Cfg.Loki, r.Header, true)
+		lokiClient := NewLokiClient(&h.Cfg.Loki, r.Header, true)
 		limits, err := h.fetchLokiLimits(lokiClient)
 		if err != nil {
 			hlog.WithError(err).Error("cannot fetch Loki limits")

pkg/handler/lokiclientmock/loki_client_mock.go

@@ -12,6 +12,8 @@ import (
 
 var mlog = logrus.WithField("module", "lokiclientmock")
 
+var folderPath = "/home/julien/dev/me/network-observability-console-plugin/"
+
 type LokiClientMock struct {
 }
 
@@ -23,10 +25,10 @@ func (o *LokiClientMock) Get(url string) ([]byte, int, error) {
 
 	isLabel := strings.Contains(url, "/label/")
 	if isLabel {
-		path = "mocks/loki/namespaces.json"
+		path = folderPath + "mocks/loki/namespaces.json"
 	} else {
 		if strings.Contains(url, "query=topk") {
-			path = "mocks/loki/flow_metrics"
+			path = folderPath + "mocks/loki/flow_metrics"
 
 			if strings.Contains(url, "|unwrap%20PktDrop") {
 				path += "_dropped"
@@ -55,7 +57,7 @@ func (o *LokiClientMock) Get(url string) ([]byte, int, error) {
 				path += "_resource.json"
 			}
 		} else {
-			path = "mocks/loki/flow_records"
+			path = folderPath + "mocks/loki/flow_records"
 			//nolint:gocritic // if-else is ok
 			if strings.Contains(url, "|~`\"Packets\":0[,}]|~`\"PktDropPackets\":[1-9][0-9]*[,}]") {
 				path += "_dropped.json"

pkg/handler/resources.go

@@ -21,7 +21,7 @@ func (h *Handlers) GetClusters(ctx context.Context) func(w http.ResponseWriter,
 		namespace := params.Get(namespaceKey)
 		isDev := namespace != ""
 
-		clients, err := newClients(h.Cfg, r.Header, false, namespace)
+		clients, err := NewClients(h.Cfg, r.Header, false, namespace)
 		if err != nil {
 			writeError(w, http.StatusInternalServerError, err.Error())
 			return
@@ -50,7 +50,7 @@ func (h *Handlers) GetUDNs(ctx context.Context) func(w http.ResponseWriter, r *h
 		namespace := params.Get(namespaceKey)
 		isDev := namespace != ""
 
-		clients, err := newClients(h.Cfg, r.Header, false, namespace)
+		clients, err := NewClients(h.Cfg, r.Header, false, namespace)
 		if err != nil {
 			writeError(w, http.StatusInternalServerError, err.Error())
 			return
@@ -79,7 +79,7 @@ func (h *Handlers) GetZones(ctx context.Context) func(w http.ResponseWriter, r *
 		namespace := params.Get(namespaceKey)
 		isDev := namespace != ""
 
-		clients, err := newClients(h.Cfg, r.Header, false, namespace)
+		clients, err := NewClients(h.Cfg, r.Header, false, namespace)
 		if err != nil {
 			writeError(w, http.StatusInternalServerError, err.Error())
 			return
@@ -139,7 +139,7 @@ func (h *Handlers) GetNamespaces(ctx context.Context) func(w http.ResponseWriter
 		namespace := params.Get(namespaceKey)
 		isDev := namespace != ""
 
-		clients, err := newClients(h.Cfg, r.Header, false, namespace)
+		clients, err := NewClients(h.Cfg, r.Header, false, namespace)
 		if err != nil {
 			writeError(w, http.StatusInternalServerError, err.Error())
 			return
@@ -196,7 +196,7 @@ func (h *Handlers) GetNames(ctx context.Context) func(w http.ResponseWriter, r *
 		namespace := params.Get(namespaceKey)
 		kind := params.Get("kind")
 
-		clients, err := newClients(h.Cfg, r.Header, false, namespace)
+		clients, err := NewClients(h.Cfg, r.Header, false, namespace)
 		if err != nil {
 			writeError(w, http.StatusInternalServerError, err.Error())
 			return

pkg/handler/topology.go

@@ -38,27 +38,26 @@ const (
 func (h *Handlers) GetTopology(ctx context.Context) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		params := r.URL.Query()
-		namespace := params.Get(namespaceKey)
-
-		clients, err := newClients(h.Cfg, r.Header, false, namespace)
-		if err != nil {
-			writeError(w, http.StatusInternalServerError, err.Error())
-			return
-		}
-
 		var code int
 		startTime := time.Now()
 		defer func() {
 			metrics.ObserveHTTPCall("GetTopology", code, startTime)
 		}()
 
+		namespace := params.Get(namespaceKey)
+		clients, err := NewClients(h.Cfg, r.Header, false, namespace)
+		if err != nil {
+			writeError(w, http.StatusInternalServerError, err.Error())
+			return
+		}
+
 		ds, err := getDatasource(params)
 		if err != nil {
 			writeError(w, http.StatusBadRequest, err.Error())
 			return
 		}
 
-		flows, code, err := h.getTopologyFlows(ctx, clients, params, ds)
+		flows, code, err := h.QueryMetrics(ctx, clients, params, ds)
 		var dsErr *datasourceError
 		if err != nil &&
 			ds == constants.DataSourceAuto &&
@@ -70,7 +69,7 @@ func (h *Handlers) GetTopology(ctx context.Context) func(w http.ResponseWriter,
 			// This is because multi-tenancy is currently not managed for prom datasource, hence such queries have to go with Loki
 			// Unfortunately we don't know a safe and generic way to pre-flight check if the user will be authorized
 			hlog.Info("Retrying with Loki...")
-			flows, code, err = h.getTopologyFlows(ctx, clients, params, constants.DataSourceLoki)
+			flows, code, err = h.QueryMetrics(ctx, clients, params, constants.DataSourceLoki)
 		}
 		if err != nil {
 			writeError(w, code, err.Error())
@@ -148,7 +147,7 @@ func (h *Handlers) extractTopologyQueryParams(params url.Values, ds constants.Da
 	return &in, filterGroups, qr, reqLimit, err
 }
 
-func (h *Handlers) getTopologyFlows(ctx context.Context, cl clients, params url.Values, ds constants.DataSource) (*model.AggregatedQueryResponse, int, error) {
+func (h *Handlers) QueryMetrics(ctx context.Context, cl clients, params url.Values, ds constants.DataSource) (*model.AggregatedQueryResponse, int, error) {
 	hlog.Debugf("GetTopology query params: %s", params)
 
 	dataSources := make(map[constants.DataSource]bool)