Skip to content

Commit 23cfc27

Browse files
freeznetfreeznet
authored
mcp: export SNCloud cluster list primitive (#96)
1 parent 203abaf commit 23cfc27

2 files changed

Lines changed: 89 additions & 24 deletions

File tree

‎pkg/mcp/prompts.go‎

Lines changed: 44 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,8 @@ const (
6969
sncloudClusterTypeKafka = "KafkaCluster"
7070
)
7171

72-
type sncloudClusterListEntry struct {
72+
// SNCloudClusterListEntry describes one StreamNative Cloud cluster in list responses.
73+
type SNCloudClusterListEntry struct {
7374
ClusterType string
7475
InstanceName string
7576
ClusterName string
@@ -118,14 +119,14 @@ func getSNCloudAPIClientAndOrganization(ctx context.Context) (*sncloud.APIClient
118119
return apiClient, session.Ctx.Organization, nil
119120
}
120121

121-
func listSNCloudPulsarClusterEntries(ctx context.Context, apiClient *sncloud.APIClient, organization string) ([]sncloudClusterListEntry, error) {
122+
func listSNCloudPulsarClusterEntries(ctx context.Context, apiClient *sncloud.APIClient, organization string) ([]SNCloudClusterListEntry, error) {
122123
clusters, clustersBody, err := apiClient.CloudStreamnativeIoV1alpha1Api.ListCloudStreamnativeIoV1alpha1NamespacedPulsarCluster(ctx, organization).Execute()
123124
if err != nil {
124125
return nil, fmt.Errorf("failed to list pulsar clusters: %v", err)
125126
}
126127
defer func() { _ = clustersBody.Body.Close() }()
127128

128-
entries := make([]sncloudClusterListEntry, 0, len(clusters.Items))
129+
entries := make([]SNCloudClusterListEntry, 0, len(clusters.Items))
129130
for _, cluster := range clusters.Items {
130131
displayName := cluster.Spec.DisplayName
131132
if displayName == nil || *displayName == "" {
@@ -142,7 +143,7 @@ func listSNCloudPulsarClusterEntries(ctx context.Context, apiClient *sncloud.API
142143
clusterName = *cluster.Metadata.Name
143144
}
144145

145-
entries = append(entries, sncloudClusterListEntry{
146+
entries = append(entries, SNCloudClusterListEntry{
146147
ClusterType: sncloudClusterTypePulsar,
147148
InstanceName: cluster.Spec.InstanceName,
148149
ClusterName: clusterName,
@@ -155,14 +156,14 @@ func listSNCloudPulsarClusterEntries(ctx context.Context, apiClient *sncloud.API
155156
return entries, nil
156157
}
157158

158-
func listSNCloudKafkaClusterEntries(ctx context.Context, apiClient *sncloud.APIClient, organization string) ([]sncloudClusterListEntry, error) {
159+
func listSNCloudKafkaClusterEntries(ctx context.Context, apiClient *sncloud.APIClient, organization string) ([]SNCloudClusterListEntry, error) {
159160
clusters, clustersBody, err := apiClient.CloudStreamnativeIoV1alpha1Api.ListCloudStreamnativeIoV1alpha1NamespacedKafkaCluster(ctx, organization).Execute()
160161
if err != nil {
161162
return nil, fmt.Errorf("failed to list kafka clusters: %v", err)
162163
}
163164
defer func() { _ = clustersBody.Body.Close() }()
164165

165-
entries := make([]sncloudClusterListEntry, 0, len(clusters.Items))
166+
entries := make([]SNCloudClusterListEntry, 0, len(clusters.Items))
166167
for _, cluster := range clusters.Items {
167168
displayName := cluster.Spec.DisplayName
168169
clusterName := ""
@@ -173,7 +174,7 @@ func listSNCloudKafkaClusterEntries(ctx context.Context, apiClient *sncloud.APIC
173174
displayName = cluster.Metadata.Name
174175
}
175176

176-
entries = append(entries, sncloudClusterListEntry{
177+
entries = append(entries, SNCloudClusterListEntry{
177178
ClusterType: sncloudClusterTypeKafka,
178179
InstanceName: cluster.Spec.InstanceName,
179180
ClusterName: clusterName,
@@ -197,7 +198,7 @@ func sncloudClusterReadinessStatus(status *sncloud.ComGithubStreamnativeCloudApi
197198
return "Not Ready"
198199
}
199200

200-
func buildSNCloudClusterPromptMessages(summary string, entries []sncloudClusterListEntry) []mcp.PromptMessage {
201+
func buildSNCloudClusterPromptMessages(summary string, entries []SNCloudClusterListEntry) []mcp.PromptMessage {
201202
messages := make([]mcp.PromptMessage, 0, len(entries)+1)
202203
messages = append(messages, mcp.PromptMessage{
203204
Content: mcp.TextContent{
@@ -231,6 +232,29 @@ func buildSNCloudClusterPromptMessages(summary string, entries []sncloudClusterL
231232
return messages
232233
}
233234

235+
// ListSNCloudClusterEntries lists all StreamNative Cloud clusters visible in the current session.
236+
func ListSNCloudClusterEntries(ctx context.Context) ([]SNCloudClusterListEntry, string, error) {
237+
apiClient, organization, err := getSNCloudAPIClientAndOrganization(ctx)
238+
if err != nil {
239+
return nil, "", err
240+
}
241+
242+
pulsarEntries, err := listSNCloudPulsarClusterEntries(ctx, apiClient, organization)
243+
if err != nil {
244+
return nil, "", err
245+
}
246+
247+
kafkaEntries, err := listSNCloudKafkaClusterEntries(ctx, apiClient, organization)
248+
if err != nil {
249+
return nil, "", err
250+
}
251+
252+
entries := make([]SNCloudClusterListEntry, 0, len(pulsarEntries)+len(kafkaEntries))
253+
entries = append(entries, pulsarEntries...)
254+
entries = append(entries, kafkaEntries...)
255+
return entries, organization, nil
256+
}
257+
234258
// NewBuildSNCloudServerlessClusterPrompt creates the reusable serverless cluster build prompt definition.
235259
func NewBuildSNCloudServerlessClusterPrompt() mcp.Prompt {
236260
return mcp.NewPrompt("build-sncloud-serverless-cluster",
@@ -243,31 +267,27 @@ func NewBuildSNCloudServerlessClusterPrompt() mcp.Prompt {
243267

244268
// HandleListSNCloudClusters handles listing StreamNative Cloud clusters.
245269
func HandleListSNCloudClusters(ctx context.Context, _ mcp.GetPromptRequest) (*mcp.GetPromptResult, error) {
246-
apiClient, organization, err := getSNCloudAPIClientAndOrganization(ctx)
247-
if err != nil {
248-
return nil, err
249-
}
250-
251-
pulsarEntries, err := listSNCloudPulsarClusterEntries(ctx, apiClient, organization)
270+
entries, organization, err := ListSNCloudClusterEntries(ctx)
252271
if err != nil {
253272
return nil, err
254273
}
255-
256-
kafkaEntries, err := listSNCloudKafkaClusterEntries(ctx, apiClient, organization)
257-
if err != nil {
258-
return nil, err
274+
pulsarCount := 0
275+
kafkaCount := 0
276+
for _, entry := range entries {
277+
switch entry.ClusterType {
278+
case sncloudClusterTypePulsar:
279+
pulsarCount++
280+
case sncloudClusterTypeKafka:
281+
kafkaCount++
282+
}
259283
}
260-
261-
entries := make([]sncloudClusterListEntry, 0, len(pulsarEntries)+len(kafkaEntries))
262-
entries = append(entries, pulsarEntries...)
263-
entries = append(entries, kafkaEntries...)
264284
messages := buildSNCloudClusterPromptMessages(
265285
fmt.Sprintf(
266286
"There are %d StreamNative Cloud clusters in organization %s (%d PulsarCluster, %d KafkaCluster):",
267287
len(entries),
268288
organization,
269-
len(pulsarEntries),
270-
len(kafkaEntries),
289+
pulsarCount,
290+
kafkaCount,
271291
),
272292
entries,
273293
)

‎pkg/mcp/streamnative_cloud_primitives_test.go‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -826,6 +826,51 @@ func TestHandleListSNCloudClustersIncludesClusterTypes(t *testing.T) {
826826
}
827827
}
828828

829+
func TestListSNCloudClusterEntriesIncludesPulsarAndKafkaClusters(t *testing.T) {
830+
t.Parallel()
831+
832+
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
833+
switch r.URL.Path {
834+
case "/apis/cloud.streamnative.io/v1alpha1/namespaces/session-org/pulsarclusters":
835+
w.Header().Set("Content-Type", "application/json")
836+
_, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"pc-test","annotations":{"cloud.streamnative.io/engine":"ursa"}},"spec":{"instanceName":"inst-p","displayName":"Pulsar Display"},"status":{"broker":{"readyReplicas":1},"conditions":[{"type":"Ready","status":"True"}]}}]}`))
837+
case "/apis/cloud.streamnative.io/v1alpha1/namespaces/session-org/kafkaclusters":
838+
w.Header().Set("Content-Type", "application/json")
839+
_, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"kc-test"},"spec":{"instanceName":"inst-k","displayName":"Kafka Display","location":"use1"},"status":{"conditions":[{"type":"Ready","status":"True"}]}}]}`))
840+
default:
841+
t.Fatalf("unexpected path %s", r.URL.Path)
842+
}
843+
}))
844+
defer server.Close()
845+
846+
session, err := config.NewSNCloudSession(config.SNCloudContext{
847+
JWTToken: "token",
848+
APIURL: server.URL,
849+
LogAPIURL: server.URL,
850+
Organization: "session-org",
851+
})
852+
if err != nil {
853+
t.Fatalf("failed to create session: %v", err)
854+
}
855+
856+
ctx := context.Background()
857+
ctx = WithSNCloudSession(ctx, session)
858+
859+
entries, organization, err := ListSNCloudClusterEntries(ctx)
860+
if err != nil {
861+
t.Fatalf("expected no list error, got %v", err)
862+
}
863+
if organization != "session-org" {
864+
t.Fatalf("expected organization session-org, got %q", organization)
865+
}
866+
if len(entries) != 2 {
867+
t.Fatalf("expected two cluster entries, got %#v", entries)
868+
}
869+
if entries[0].ClusterType != "PulsarCluster" || entries[1].ClusterType != "KafkaCluster" {
870+
t.Fatalf("expected typed pulsar and kafka entries, got %#v", entries)
871+
}
872+
}
873+
829874
func TestBuildSNCloudContextClusterPromptResultUsesPulsarClustersOnly(t *testing.T) {
830875
t.Parallel()
831876

0 commit comments

Comments
 (0)