feat: watch for configuration changes

Watch kube config files for changes.
Automatically reload kubernetes client and list of tools.

Useful for logins or context changes after an MCP session has started.
This commit is contained in:
Marc Nuri
2025-03-21 18:05:41 +01:00
parent c9def7dd46
commit a98e69102c
7 changed files with 97 additions and 6 deletions

View File

@@ -13,8 +13,9 @@ https://github.com/user-attachments/assets/be2b67b3-fc1c-4d11-ae46-93deba8ed98e
A powerful and flexible Kubernetes [Model Context Protocol (MCP)](https://blog.marcnuri.com/model-context-protocol-mcp-introduction) server implementation with support for OpenShift.
- **✅ Configuration**: View and manage the [Kubernetes `.kube/config`](https://blog.marcnuri.com/where-is-my-default-kubeconfig-file).
- **View** the current configuration.
- **✅ Configuration**:
- Automatically detect changes in the Kubernetes configuration and update the MCP server.
- **View** and manage the current [Kubernetes `.kube/config`](https://blog.marcnuri.com/where-is-my-default-kubeconfig-file) or in-cluster configuration.
- **✅ Generic Kubernetes Resources**: Perform operations on any Kubernetes resource.
- Any CRUD operation (Create or Update, Get, List, Delete).
- **✅ Pods**: Perform Pod-specific operations.

2
go.mod
View File

@@ -3,6 +3,7 @@ module github.com/manusa/kubernetes-mcp-server
go 1.23.5
require (
github.com/fsnotify/fsnotify v1.8.0
github.com/mark3labs/mcp-go v0.15.0
github.com/spf13/afero v1.14.0
github.com/spf13/cobra v1.9.1
@@ -22,7 +23,6 @@ require (
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/emicklei/go-restful/v3 v3.11.0 // indirect
github.com/evanphx/json-patch/v5 v5.9.11 // indirect
github.com/fsnotify/fsnotify v1.8.0 // indirect
github.com/fxamacker/cbor/v2 v2.7.0 // indirect
github.com/go-logr/logr v1.4.2 // indirect
github.com/go-openapi/jsonpointer v0.21.0 // indirect

View File

@@ -42,6 +42,7 @@ Kubernetes Model Context Protocol (MCP) server
if err != nil {
panic(err)
}
defer mcpServer.Close()
var sseServer *server.SSEServer
if ssePort := viper.GetInt("sse-port"); ssePort > 0 {
@@ -49,13 +50,11 @@ Kubernetes Model Context Protocol (MCP) server
if err := sseServer.Start(fmt.Sprintf(":%d", ssePort)); err != nil {
panic(err)
}
defer sseServer.Shutdown(cmd.Context())
}
if err := mcpServer.ServeStdio(); err != nil && !errors.Is(err, context.Canceled) {
panic(err)
}
if sseServer != nil {
_ = sseServer.Shutdown(cmd.Context())
}
},
}

View File

@@ -1,6 +1,7 @@
package kubernetes
import (
"github.com/fsnotify/fsnotify"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/client-go/discovery"
"k8s.io/client-go/discovery/cached/memory"
@@ -17,8 +18,12 @@ import (
// Exposed for testing
var InClusterConfig = rest.InClusterConfig
type CloseWatchKubeConfig func() error
type Kubernetes struct {
cfg *rest.Config
kubeConfigFiles []string
CloseWatchKubeConfig CloseWatchKubeConfig
clientSet *kubernetes.Clientset
discoveryClient *discovery.DiscoveryClient
deferredDiscoveryRESTMapper *restmapper.DeferredDiscoveryRESTMapper
@@ -44,6 +49,7 @@ func NewKubernetes() (*Kubernetes, error) {
}
return &Kubernetes{
cfg: cfg,
kubeConfigFiles: resolveConfig().ConfigAccess().GetLoadingPrecedence(),
clientSet: clientSet,
discoveryClient: discoveryClient,
deferredDiscoveryRESTMapper: restmapper.NewDeferredDiscoveryRESTMapper(memory.NewMemCacheClient(discoveryClient)),
@@ -51,6 +57,44 @@ func NewKubernetes() (*Kubernetes, error) {
}, nil
}
func (k *Kubernetes) WatchKubeConfig(onKubeConfigChange func() error) {
if len(k.kubeConfigFiles) == 0 {
return
}
watcher, err := fsnotify.NewWatcher()
if err != nil {
return
}
for _, file := range k.kubeConfigFiles {
_ = watcher.Add(file)
}
go func() {
for {
select {
case _, ok := <-watcher.Events:
if !ok {
return
}
_ = onKubeConfigChange()
case _, ok := <-watcher.Errors:
if !ok {
return
}
}
}
}()
if k.CloseWatchKubeConfig != nil {
_ = k.CloseWatchKubeConfig()
}
k.CloseWatchKubeConfig = watcher.Close
}
func (k *Kubernetes) Close() {
if k.CloseWatchKubeConfig != nil {
_ = k.CloseWatchKubeConfig()
}
}
func marshal(v any) (string, error) {
switch t := v.(type) {
case []unstructured.Unstructured:

View File

@@ -129,6 +129,7 @@ func (c *mcpContext) beforeEach(t *testing.T) {
func (c *mcpContext) afterEach() {
c.cancel()
c.mcpServer.Close()
_ = c.mcpClient.Close()
c.mcpHttpServer.Close()
}

View File

@@ -27,6 +27,7 @@ func NewSever() (*Server, error) {
if err := s.reloadKubernetesClient(); err != nil {
return nil, err
}
s.k.WatchKubeConfig(s.reloadKubernetesClient)
return s, nil
}
@@ -57,6 +58,12 @@ func (s *Server) ServeSse(baseUrl string) *server.SSEServer {
return server.NewSSEServer(s.server, options...)
}
func (s *Server) Close() {
if s.k != nil {
s.k.Close()
}
}
func NewTextResult(content string, err error) *mcp.CallToolResult {
if err != nil {
return &mcp.CallToolResult{

View File

@@ -1,12 +1,51 @@
package mcp
import (
"context"
"github.com/mark3labs/mcp-go/mcp"
"os"
"path/filepath"
"slices"
"strings"
"testing"
"time"
)
func TestWatchKubeConfig(t *testing.T) {
testCase(t, func(c *mcpContext) {
// Given
withTimeout, cancel := context.WithTimeout(c.ctx, 5*time.Second)
defer cancel()
var notification *mcp.JSONRPCNotification
c.mcpClient.OnNotification(func(n mcp.JSONRPCNotification) {
notification = &n
})
// When
f, _ := os.OpenFile(filepath.Join(c.tempDir, "config"), os.O_APPEND|os.O_WRONLY, 0644)
_, _ = f.WriteString("\n")
for {
if notification != nil {
break
}
select {
case <-withTimeout.Done():
break
default:
time.Sleep(100 * time.Millisecond)
}
}
// Then
t.Run("WatchKubeConfig notifies tools change", func(t *testing.T) {
if notification == nil {
t.Fatalf("WatchKubeConfig did not notify")
}
if notification.Method != "notifications/tools/list_changed" {
t.Fatalf("WatchKubeConfig did not notify tools change, got %s", notification.Method)
}
})
})
}
func TestTools(t *testing.T) {
expectedNames := []string{
"configuration_view",