fix: Tool Input Array JSON Schema Validation (#31)

sijie · web-flow · commit f7390a05578f · 2025-05-15T09:20:52.000+08:00
We don't provide the item schema in array types. It fails when using
OpenAI and Gemini models
diff --git a/pkg/mcp/kafka_admin_topics_tools.go b/pkg/mcp/kafka_admin_topics_tools.go
@@ -113,6 +113,12 @@ func KafkaAdminAddTopicTools(s *server.MCPServer, readOnly bool, features []stri
 				"Each entry should be in the format 'key=value'. "+
 				"Example configs include 'cleanup.policy', 'retention.ms', 'compression.type'. "+
 				"Optional."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "config entry",
+				},
+			),
 		),
 	)
 	s.AddTool(kafkaTopicTool, handleKafkaTopicTool(readOnly))
diff --git a/pkg/mcp/kafka_client_produce_tools.go b/pkg/mcp/kafka_client_produce_tools.go
@@ -93,6 +93,12 @@ func KafkaClientAddProduceTools(s *server.MCPServer, readOnly bool, features []s
 			mcp.Description("Message headers in the format of [{\"key\": \"value\"}]. "+
 				"Optional. Headers allow you to attach metadata to messages without modifying the payload. "+
 				"They are passed along with the message to consumers."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "header entry",
+				},
+			),
 		),
 		mcp.WithBoolean("sync",
 			mcp.Description("Whether to wait for server acknowledgment before returning. "+
diff --git a/pkg/mcp/pulsar_admin_cluster_tools.go b/pkg/mcp/pulsar_admin_cluster_tools.go
@@ -84,9 +84,21 @@ func PulsarAdminAddClusterTools(s *server.MCPServer, readOnly bool, features []s
 			mcp.Description("List of clusters to be registered as peer-clusters, used when:\n"+
 				"- resource=cluster and operation is create or update\n"+
 				"- resource=peer_clusters and operation is update"),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "peer cluster name",
+				},
+			),
 		),
 		mcp.WithArray("brokers",
 			mcp.Description("List of broker names to include in a failure domain, required when resource=failure_domain and operation is create or update"),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "broker",
+				},
+			),
 		),
 	)
 	s.AddTool(clusterTool, handleClusterTool(readOnly))
diff --git a/pkg/mcp/pulsar_admin_functions_tools.go b/pkg/mcp/pulsar_admin_functions_tools.go
@@ -90,7 +90,14 @@ func PulsarAdminAddFunctionsTools(s *server.MCPServer, readOnly bool, features [
 			mcp.Description("The input topics for the function (array of strings). Optional for 'create' and 'update' operations. "+
 				"Topics must be specified in the format 'persistent://tenant/namespace/topic'. "+
 				"Functions can consume from multiple topics, each with potentially different serialization types. "+
-				"All input topics should exist before the function is created.")),
+				"All input topics should exist before the function is created."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "input topic",
+				},
+			),
+		),
 		mcp.WithString("output",
 			mcp.Description("The output topic for the function results. Optional for 'create' and 'update' operations. "+
 				"Specified in the format 'persistent://tenant/namespace/topic'. "+
diff --git a/pkg/mcp/pulsar_admin_namespace_policy_tools.go b/pkg/mcp/pulsar_admin_namespace_policy_tools.go
@@ -172,12 +172,30 @@ func PulsarAdminAddNamespacePolicyTools(s *server.MCPServer, readOnly bool, feat
 			),
 			mcp.WithArray("actions",
 				mcp.Description("Actions to grant for permission policies (e.g., produce, consume)"),
+				mcp.Items(
+					map[string]interface{}{
+						"type":        "string",
+						"description": "action",
+					},
+				),
 			),
 			mcp.WithArray("clusters",
 				mcp.Description("List of clusters for replication policies"),
+				mcp.Items(
+					map[string]interface{}{
+						"type":        "string",
+						"description": "cluster",
+					},
+				),
 			),
 			mcp.WithArray("roles",
 				mcp.Description("List of roles for subscription permission policies"),
+				mcp.Items(
+					map[string]interface{}{
+						"type":        "string",
+						"description": "role",
+					},
+				),
 			),
 			mcp.WithString("ttl",
 				mcp.Description("Message TTL in seconds (or 0 to disable TTL)"),
diff --git a/pkg/mcp/pulsar_admin_namespace_tools.go b/pkg/mcp/pulsar_admin_namespace_tools.go
@@ -64,6 +64,12 @@ func PulsarAdminAddNamespaceTools(s *server.MCPServer, readOnly bool, features [
 		),
 		mcp.WithArray("clusters",
 			mcp.Description("List of clusters to assign when creating a namespace. Used with 'create' operation."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "Cluster name",
+				},
+			),
 		),
 		mcp.WithString("subscription",
 			mcp.Description("Subscription name. Required for 'unsubscribe' operation, optional for 'clear_backlog'."),
diff --git a/pkg/mcp/pulsar_admin_nsisolationpolicy_tools.go b/pkg/mcp/pulsar_admin_nsisolationpolicy_tools.go
@@ -70,12 +70,30 @@ func PulsarAdminAddNsIsolationPolicyTools(s *server.MCPServer, readOnly bool, fe
 		),
 		mcp.WithArray("namespaces",
 			mcp.Description("List of namespaces to apply the isolation policy. Required for policy.set"),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "namespace",
+				},
+			),
 		),
 		mcp.WithArray("primary",
 			mcp.Description("List of primary brokers for the namespaces. Required for policy.set"),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "primary broker",
+				},
+			),
 		),
 		mcp.WithArray("secondary",
 			mcp.Description("List of secondary brokers for the namespaces. Optional for policy.set"),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "secondary broker",
+				},
+			),
 		),
 		mcp.WithString("autoFailoverPolicyType",
 			mcp.Description("Auto failover policy type (e.g., min_available). Optional for policy.set"),
diff --git a/pkg/mcp/pulsar_admin_sinks_tools.go b/pkg/mcp/pulsar_admin_sinks_tools.go
@@ -94,7 +94,14 @@ func PulsarAdminAddSinksTools(s *server.MCPServer, readOnly bool, features []str
 				"Topics must be specified in the format 'persistent://tenant/namespace/topic'. "+
 				"Sinks can consume from multiple topics, but they should have compatible schemas. "+
 				"All input topics should exist before the sink is created. "+
-				"Either inputs or topics-pattern must be specified.")),
+				"Either inputs or topics-pattern must be specified."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "input topic",
+				},
+			),
+		),
 		mcp.WithString("topics-pattern",
 			mcp.Description("TopicsPattern to consume from list of topics that match the pattern. Optional for 'create' and 'update' operations. "+
 				"Specified as a regular expression, e.g., 'persistent://tenant/namespace/prefix.*'. "+
diff --git a/pkg/mcp/pulsar_admin_tenant_tools.go b/pkg/mcp/pulsar_admin_tenant_tools.go
@@ -77,12 +77,24 @@ func PulsarAdminAddTenantTools(s *server.MCPServer, readOnly bool, features []st
 				"namespaces within the tenant, and can manage topic configurations. "+
 				"Format: array of role strings, e.g., ['admin1', 'orgAdmin']. "+
 				"Use empty array [] to remove all admin roles."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "role",
+				},
+			),
 		),
 		mcp.WithArray("allowedClusters",
 			mcp.Description("List of clusters that this tenant can access. Required for 'create' and 'update' operations. "+
 				"Restricts the tenant to only use specified clusters, enabling geographic or infrastructure isolation. "+
 				"Format: array of cluster names, e.g., ['us-west', 'us-east']. "+
 				"An empty list means no clusters are accessible to this tenant."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "cluster",
+				},
+			),
 		),
 	)
 	s.AddTool(tenantTool, handleTenantTool(readOnly))
diff --git a/pkg/mcp/pulsar_admin_topic_policy_tools.go b/pkg/mcp/pulsar_admin_topic_policy_tools.go
@@ -178,6 +178,12 @@ func PulsarAdminAddTopicPolicyTools(s *server.MCPServer, readOnly bool, features
 			),
 			mcp.WithArray("actions",
 				mcp.Description("Actions to grant for permission policies. Valid values are: produce, consume"),
+				mcp.Items(
+					map[string]interface{}{
+						"type":        "string",
+						"description": "action",
+					},
+				),
 			),
 			mcp.WithString("ttl",
 				mcp.Description("Message TTL in seconds (or 0 to disable TTL)"),
diff --git a/pkg/mcp/pulsar_client_produce_tools.go b/pkg/mcp/pulsar_client_produce_tools.go
@@ -47,6 +47,12 @@ func PulsarClientAddProducerTools(s *server.MCPServer, _ bool, features []string
 		),
 		mcp.WithArray("messages",
 			mcp.Description("Messages to send. Specify multiple times for multiple messages. IMPORTANT: Use this parameter to provide message content."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "message",
+				},
+			),
 		),
 		mcp.WithNumber("num-produce",
 			mcp.Description("Number of times to send message(s) (default: 1)"),
@@ -65,6 +71,12 @@ func PulsarClientAddProducerTools(s *server.MCPServer, _ bool, features []string
 		),
 		mcp.WithArray("properties",
 			mcp.Description("Properties to add, key=value format. Specify multiple times for multiple properties."),
+			mcp.Items(
+				map[string]interface{}{
+					"type":        "string",
+					"description": "property",
+				},
+			),
 		),
 		mcp.WithString("key",
 			mcp.Description("Partitioning key to add to each message"),
