Skip to content

feat(transport): add Streamable HTTP transport at /mcp #117

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
basebandit wants to merge 2 commits into from
Closed

feat(transport): add Streamable HTTP transport at /mcp #117

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5f52641
feat(transport): add Streamable HTTP transport at /mcp
basebandit May 7, 2026
63d6a01
feat(cluster): honor --request-timeout flag in K8s client config (#121)
basebandit May 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 43 additions & 43 deletions cluster/formatted.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -393,27 +393,27 @@ func formatServiceList(services *corev1.ServiceList, includeNamespace bool) stri

// Format the output
if includeNamespace {
result.WriteString(fmt.Sprintf("• %s/%s: Type=%s, ClusterIP=%s, ExternalIP=%s, Ports=%s, Age=%s\n",
fmt.Fprintf(&result, "• %s/%s: Type=%s, ClusterIP=%s, ExternalIP=%s, Ports=%s, Age=%s\n",
svc.Namespace,
svc.Name,
serviceType,
clusterIP,
externalIP,
portsStr,
formatDuration(age)))
formatDuration(age))
} else {
result.WriteString(fmt.Sprintf("• %s: Type=%s, ClusterIP=%s, ExternalIP=%s, Ports=%s, Age=%s\n",
fmt.Fprintf(&result, "• %s: Type=%s, ClusterIP=%s, ExternalIP=%s, Ports=%s, Age=%s\n",
svc.Name,
serviceType,
clusterIP,
externalIP,
portsStr,
formatDuration(age)))
formatDuration(age))
}
}

// Add total count
result.WriteString(fmt.Sprintf("\nTotal: %d service(s)", len(services.Items)))
fmt.Fprintf(&result, "\nTotal: %d service(s)", len(services.Items))

return result.String()
}
Expand Down Expand Up @@ -612,7 +612,7 @@ func formatNamespaceList(namespaces *corev1.NamespaceList, labelSelector string)
var result strings.Builder

if labelSelector != "" {
result.WriteString(fmt.Sprintf("Namespaces matching label selector '%s':\n", labelSelector))
fmt.Fprintf(&result, "Namespaces matching label selector '%s':\n", labelSelector)
} else {
result.WriteString("Namespaces:\n")
}
Expand All @@ -624,18 +624,18 @@ func formatNamespaceList(namespaces *corev1.NamespaceList, labelSelector string)
status = "Active"
}

result.WriteString(fmt.Sprintf("• %s: Status=%s, Age=%s",
ns.Name, status, formatDuration(age)))
fmt.Fprintf(&result, "• %s: Status=%s, Age=%s",
ns.Name, status, formatDuration(age))

if len(ns.Labels) > 0 {
labelCount := len(ns.Labels)
result.WriteString(fmt.Sprintf(" - Labels: %d", labelCount))
fmt.Fprintf(&result, " - Labels: %d", labelCount)
}

result.WriteString("\n")
}

result.WriteString(fmt.Sprintf("\nTotal: %d namespace(s)", len(namespaces.Items)))
fmt.Fprintf(&result, "\nTotal: %d namespace(s)", len(namespaces.Items))

return result.String()
}
Expand Down Expand Up @@ -688,21 +688,21 @@ func formatConfigMapList(configMaps *corev1.ConfigMapList, includeNamespace bool
dataCount := len(cm.Data) + len(cm.BinaryData)

if includeNamespace {
result.WriteString(fmt.Sprintf("• %s/%s: Data=%d, Age=%s",
cm.Namespace, cm.Name, dataCount, formatDuration(age)))
fmt.Fprintf(&result, "• %s/%s: Data=%d, Age=%s",
cm.Namespace, cm.Name, dataCount, formatDuration(age))
} else {
result.WriteString(fmt.Sprintf("• %s: Data=%d, Age=%s",
cm.Name, dataCount, formatDuration(age)))
fmt.Fprintf(&result, "• %s: Data=%d, Age=%s",
cm.Name, dataCount, formatDuration(age))
}

if len(cm.Labels) > 0 {
result.WriteString(fmt.Sprintf(" - Labels: %d", len(cm.Labels)))
fmt.Fprintf(&result, " - Labels: %d", len(cm.Labels))
}

result.WriteString("\n")
}

result.WriteString(fmt.Sprintf("\nTotal: %d ConfigMap(s)", len(configMaps.Items)))
fmt.Fprintf(&result, "\nTotal: %d ConfigMap(s)", len(configMaps.Items))

return result.String()
}
Expand Down Expand Up @@ -745,21 +745,21 @@ func formatSecretList(secrets *corev1.SecretList, includeNamespace bool) string
dataCount := len(secret.Data)

if includeNamespace {
result.WriteString(fmt.Sprintf("• %s/%s: Type=%s, Data=%d, Age=%s",
secret.Namespace, secret.Name, secret.Type, dataCount, formatDuration(age)))
fmt.Fprintf(&result, "• %s/%s: Type=%s, Data=%d, Age=%s",
secret.Namespace, secret.Name, secret.Type, dataCount, formatDuration(age))
} else {
result.WriteString(fmt.Sprintf("• %s: Type=%s, Data=%d, Age=%s",
secret.Name, secret.Type, dataCount, formatDuration(age)))
fmt.Fprintf(&result, "• %s: Type=%s, Data=%d, Age=%s",
secret.Name, secret.Type, dataCount, formatDuration(age))
}

if len(secret.Labels) > 0 {
result.WriteString(fmt.Sprintf(" - Labels: %d", len(secret.Labels)))
fmt.Fprintf(&result, " - Labels: %d", len(secret.Labels))
}

result.WriteString("\n")
}

result.WriteString(fmt.Sprintf("\nTotal: %d Secret(s)", len(secrets.Items)))
fmt.Fprintf(&result, "\nTotal: %d Secret(s)", len(secrets.Items))

return result.String()
}
Expand Down Expand Up @@ -812,7 +812,7 @@ func formatJobList(jobs *batchv1.JobList, includeNamespace bool) string {
if includeNamespace {
result.WriteString("Jobs across all namespaces:\n")
} else {
result.WriteString(fmt.Sprintf("Jobs in namespace %q:\n", jobs.Items[0].Namespace))
fmt.Fprintf(&result, "Jobs in namespace %q:\n", jobs.Items[0].Namespace)
}

for _, job := range jobs.Items {
Expand All @@ -832,21 +832,21 @@ func formatJobList(jobs *batchv1.JobList, includeNamespace bool) string {
}

if includeNamespace {
result.WriteString(fmt.Sprintf("• %s/%s: %s - Age: %s",
job.Namespace, job.Name, status, formatDuration(age)))
fmt.Fprintf(&result, "• %s/%s: %s - Age: %s",
job.Namespace, job.Name, status, formatDuration(age))
} else {
result.WriteString(fmt.Sprintf("• %s: %s - Age: %s",
job.Name, status, formatDuration(age)))
fmt.Fprintf(&result, "• %s: %s - Age: %s",
job.Name, status, formatDuration(age))
}

if len(job.Labels) > 0 {
result.WriteString(fmt.Sprintf(" - Labels: %d", len(job.Labels)))
fmt.Fprintf(&result, " - Labels: %d", len(job.Labels))
}

result.WriteString("\n")
}

result.WriteString(fmt.Sprintf("\nTotal: %d Job(s)", len(jobs.Items)))
fmt.Fprintf(&result, "\nTotal: %d Job(s)", len(jobs.Items))

return result.String()
}
Expand Down Expand Up @@ -946,7 +946,7 @@ func formatCronJobList(cronJobs *batchv1.CronJobList, includeNamespace bool) str
result.WriteString("CronJobs across all namespaces:\n")
} else {
if len(cronJobs.Items) > 0 {
result.WriteString(fmt.Sprintf("CronJobs in namespace %q:\n", cronJobs.Items[0].Namespace))
fmt.Fprintf(&result, "CronJobs in namespace %q:\n", cronJobs.Items[0].Namespace)
} else {
result.WriteString("CronJobs in namespace:\n")
}
Expand All @@ -966,21 +966,21 @@ func formatCronJobList(cronJobs *batchv1.CronJobList, includeNamespace bool) str
}

if includeNamespace {
result.WriteString(fmt.Sprintf("• %s/%s: Schedule=%s, %s, LastSchedule=%s, Active=%d, Age=%s",
cronJob.Namespace, cronJob.Name, cronJob.Spec.Schedule, suspended, lastSchedule, len(cronJob.Status.Active), formatDuration(age)))
fmt.Fprintf(&result, "• %s/%s: Schedule=%s, %s, LastSchedule=%s, Active=%d, Age=%s",
cronJob.Namespace, cronJob.Name, cronJob.Spec.Schedule, suspended, lastSchedule, len(cronJob.Status.Active), formatDuration(age))
} else {
result.WriteString(fmt.Sprintf("• %s: Schedule=%s, %s, LastSchedule=%s, Active=%d, Age=%s",
cronJob.Name, cronJob.Spec.Schedule, suspended, lastSchedule, len(cronJob.Status.Active), formatDuration(age)))
fmt.Fprintf(&result, "• %s: Schedule=%s, %s, LastSchedule=%s, Active=%d, Age=%s",
cronJob.Name, cronJob.Spec.Schedule, suspended, lastSchedule, len(cronJob.Status.Active), formatDuration(age))
}

if len(cronJob.Labels) > 0 {
result.WriteString(fmt.Sprintf(" - Labels: %d", len(cronJob.Labels)))
fmt.Fprintf(&result, " - Labels: %d", len(cronJob.Labels))
}

result.WriteString("\n")
}

result.WriteString(fmt.Sprintf("\nTotal: %d CronJob(s)", len(cronJobs.Items)))
fmt.Fprintf(&result, "\nTotal: %d CronJob(s)", len(cronJobs.Items))

return result.String()
}
Expand Down Expand Up @@ -1095,7 +1095,7 @@ func formatIngressList(ingresses *networkingv1.IngressList, includeNamespace boo
result.WriteString("Ingresses across all namespaces:\n")
} else {
if len(ingresses.Items) > 0 {
result.WriteString(fmt.Sprintf("Ingresses in namespace %q:\n", ingresses.Items[0].Namespace))
fmt.Fprintf(&result, "Ingresses in namespace %q:\n", ingresses.Items[0].Namespace)
} else {
result.WriteString("Ingresses in namespace:\n")
}
Expand Down Expand Up @@ -1134,11 +1134,11 @@ func formatIngressList(ingresses *networkingv1.IngressList, includeNamespace boo
}

if includeNamespace {
result.WriteString(fmt.Sprintf("• %s/%s: Class=%s, Hosts=%s, Address=%s, Age=%s",
ingress.Namespace, ingress.Name, ingressClass, hostStr, address, formatDuration(age)))
fmt.Fprintf(&result, "• %s/%s: Class=%s, Hosts=%s, Address=%s, Age=%s",
ingress.Namespace, ingress.Name, ingressClass, hostStr, address, formatDuration(age))
} else {
result.WriteString(fmt.Sprintf("• %s: Class=%s, Hosts=%s, Address=%s, Age=%s",
ingress.Name, ingressClass, hostStr, address, formatDuration(age)))
fmt.Fprintf(&result, "• %s: Class=%s, Hosts=%s, Address=%s, Age=%s",
ingress.Name, ingressClass, hostStr, address, formatDuration(age))
}

// TLS indicator
Expand All @@ -1147,13 +1147,13 @@ func formatIngressList(ingresses *networkingv1.IngressList, includeNamespace boo
}

if len(ingress.Labels) > 0 {
result.WriteString(fmt.Sprintf(" - Labels: %d", len(ingress.Labels)))
fmt.Fprintf(&result, " - Labels: %d", len(ingress.Labels))
}

result.WriteString("\n")
}

result.WriteString(fmt.Sprintf("\nTotal: %d Ingress(es)", len(ingresses.Items)))
fmt.Fprintf(&result, "\nTotal: %d Ingress(es)", len(ingresses.Items))

return result.String()
}
42 changes: 35 additions & 7 deletions cluster/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,17 +32,43 @@ type Manager struct {
contexts map[string]*kai.ContextInfo
currentContext string
currentNamespace string
requestTimeout time.Duration
}

// New creates a new cluster Manager
func New() *Manager {
return &Manager{
// Option configures a Manager.
type Option func(*Manager)

// WithRequestTimeout sets the per-request timeout applied to every Kubernetes
// API client created by the Manager. Zero or negative values are ignored and
// the 30 s default is kept.
func WithRequestTimeout(d time.Duration) Option {
return func(cm *Manager) {
if d > 0 {
cm.requestTimeout = d
}
}
}

// New creates a new cluster Manager. Without options the default request
// timeout is 30 seconds.
func New(opts ...Option) *Manager {
cm := &Manager{
kubeconfigs: make(map[string]string),
clients: make(map[string]kubernetes.Interface),
dynamicClients: make(map[string]dynamic.Interface),
contexts: make(map[string]*kai.ContextInfo),
currentNamespace: "default",
requestTimeout: 30 * time.Second,
}
for _, opt := range opts {
opt(cm)
}
return cm
}

// RequestTimeout returns the per-request timeout configured on this Manager.
func (cm *Manager) RequestTimeout() time.Duration {
return cm.requestTimeout
}

// LoadKubeConfig loads a kubeconfig file into the manager
Expand All @@ -69,7 +95,7 @@ func (cm *Manager) LoadKubeConfig(name, path string) error {
return err
}

clientset, dynamicClient, err := createClients(resolvedPath)
clientset, dynamicClient, err := cm.createClients(resolvedPath)
if err != nil {
return err
}
Expand Down Expand Up @@ -406,14 +432,16 @@ func (cm *Manager) updateKubeconfigCurrentContext(contextName, configPath string
return nil
}

// createClients creates Kubernetes clients from the kubeconfig file
func createClients(path string) (kubernetes.Interface, dynamic.Interface, error) {
// createClients builds Kubernetes typed and dynamic clients from a kubeconfig
// path. The per-request timeout is taken from the Manager so the user-facing
// --request-timeout flag is honored end-to-end.
func (cm *Manager) createClients(path string) (kubernetes.Interface, dynamic.Interface, error) {
config, err := clientcmd.BuildConfigFromFlags("", path)
if err != nil {
return nil, nil, fmt.Errorf("error building config from flags: %w", err)
}

config.Timeout = 30 * time.Second
config.Timeout = cm.requestTimeout

clientset, err := kubernetes.NewForConfig(config)
if err != nil {
Expand Down
21 changes: 16 additions & 5 deletions cmd/kai/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,8 @@

flag.StringVar(&kubeconfig, "kubeconfig", defaultKubeconfig, "Path to kubeconfig file")
flag.StringVar(&contextName, "context", "local", "Name for the loaded context")
flag.StringVar(&transport, "transport", "stdio", "Transport mode: stdio (default) or sse")
flag.StringVar(&sseAddr, "sse-addr", ":8080", "Address for SSE server (only used with -transport=sse)")
flag.StringVar(&transport, "transport", "stdio", "Transport mode: stdio (default), streamable-http, or sse-legacy. \"sse\" is accepted as a deprecated alias of \"sse-legacy\".")
flag.StringVar(&sseAddr, "sse-addr", ":8080", "Address for the HTTP listener (used with streamable-http or sse-legacy). The flag name is kept for backwards compatibility.")
flag.StringVar(&logFormat, "log-format", "json", "Log format: json (default) or text")
flag.StringVar(&logLevel, "log-level", "info", "Log level: debug, info, warn, error")
flag.StringVar(&tlsCert, "tls-cert", "", "Path to TLS certificate file (enables HTTPS for SSE)")
Expand All @@ -63,7 +63,7 @@
}

// Initialize cluster manager
cm := cluster.New()
cm := cluster.New(cluster.WithRequestTimeout(requestTimeout))

if err := cm.LoadKubeConfig(contextName, kubeconfig); err != nil {
logger.Error("failed to load kubeconfig",
Expand Down Expand Up @@ -105,17 +105,28 @@

go func() {
switch transport {
case "streamable-http", "http":
logger.Info("starting server",

Check failure on line 109 in cmd/kai/main.go

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Define a constant instead of duplicating this literal "starting server" 3 times. 


See more on https://sonarcloud.io/project/issues?id=basebandit_kai&issues=AZ4blpwK8bS0tmnOVIBt&open=AZ4blpwK8bS0tmnOVIBt&pullRequest=123
slog.String("transport", "streamable-http"),
slog.String("address", sseAddr),
)
errChan <- s.ServeStreamableHTTP(sseAddr)
case "sse":
logger.Warn("transport \"sse\" is deprecated; use \"sse-legacy\" or migrate to \"streamable-http\"")
fallthrough
case "sse-legacy":
logger.Info("starting server",
slog.String("transport", "sse"),
slog.String("transport", "sse-legacy"),
slog.String("address", sseAddr),
)
errChan <- s.ServeSSE(sseAddr)
default:
case "stdio", "":
logger.Info("starting server",
slog.String("transport", "stdio"),
)
errChan <- s.Serve()
default:
errChan <- fmt.Errorf("unknown transport %q (valid: stdio, streamable-http, sse-legacy)", transport)
}
}()

Expand Down
Loading
Loading