search-topics-by-name
Find specific Kafka topics in your cluster by name using the Schema Registry REST API. Streamline topic identification for Confluent Kafka or Confluent Cloud environments.
Instructions
List all topics in the Kafka cluster matching the specified name.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| baseUrl | No | The base URL of the Schema Registry REST API. | |
| topicName | Yes | The topic name to search for |
Input Schema (JSON Schema)
{
"$schema": "http://json-schema.org/draft-07/schema#",
"additionalProperties": false,
"properties": {
"baseUrl": {
"default": "",
"description": "The base URL of the Schema Registry REST API.",
"format": "uri",
"type": "string"
},
"topicName": {
"description": "The topic name to search for",
"type": "string"
}
},
"required": [
"topicName"
],
"type": "object"
}
Implementation Reference
- The handle method of SearchTopicsByNameHandler that executes the tool: parses args, calls Schema Registry API to search topics by name, and returns matching topic names.async handle( clientManager: ClientManager, toolArguments: Record<string, unknown>, ): Promise<CallToolResult> { const { topicName, baseUrl } = searchTopicsByNameArguments.parse(toolArguments); if (baseUrl !== undefined && baseUrl !== "") { clientManager.setConfluentCloudSchemaRegistryEndpoint(baseUrl); } const pathBasedClient = wrapAsPathBasedClient( clientManager.getConfluentCloudSchemaRegistryRestClient(), ); const { data: response, error } = await pathBasedClient[ "/catalog/v1/search/basic?types=kafka_topic&query={topicName}" ].GET({ params: { path: { topicName: topicName, }, }, }); if (error) { return this.createResponse( `Failed to search for topics by name: ${JSON.stringify(error)}`, true, ); } return this.createResponse( response?.entities ?.map((entity) => entity.attributes?.qualifiedName) .filter(Boolean) .join(", ") || "No matching topics found", ); }
- Zod input schema defining parameters: topicName (required string) and optional baseUrl.const searchTopicsByNameArguments = z.object({ baseUrl: z .string() .describe("The base URL of the Schema Registry REST API.") .url() .default(() => env.SCHEMA_REGISTRY_ENDPOINT ?? "") .optional(), topicName: z.string().describe("The topic name to search for"), });
- src/confluent/tools/tool-factory.ts:62-62 (registration)Registration of the SearchTopicsByNameHandler in the ToolFactory's handlers Map using the tool name constant.[ToolName.SEARCH_TOPICS_BY_NAME, new SearchTopicsByNameHandler()],
- ToolName enum constant defining the string identifier 'search-topics-by-name' used for registration and configuration.SEARCH_TOPICS_BY_NAME = "search-topics-by-name",
- getToolConfig method providing the tool name, description, and inputSchema for MCP tool registration.getToolConfig(): ToolConfig { return { name: ToolName.SEARCH_TOPICS_BY_NAME, description: "List all topics in the Kafka cluster matching the specified name.", inputSchema: searchTopicsByNameArguments.shape, }; }