mcp-confluent

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/confluent/tools/handlers/connect/create-connector-handler.ts:78-141 (handler)

  The `handle` method in `CreateConnectorHandler` class that parses arguments, validates environment and cluster IDs, makes a POST request to the Confluent Cloud REST API to create the connector, and returns the result or error.
  async handle( clientManager: ClientManager, toolArguments: Record<string, unknown> | undefined, ): Promise<CallToolResult> { const { clusterId, environmentId, connectorName, connectorConfig, baseUrl, } = createConnectorArguments.parse(toolArguments); const environment_id = getEnsuredParam( "KAFKA_ENV_ID", "Environment ID is required", environmentId, ); const kafka_cluster_id = getEnsuredParam( "KAFKA_CLUSTER_ID", "Kafka Cluster ID is required", clusterId, ); if (baseUrl !== undefined && baseUrl !== "") { clientManager.setConfluentCloudRestEndpoint(baseUrl); } const pathBasedClient = wrapAsPathBasedClient( clientManager.getConfluentCloudRestClient(), ); const { data: response, error } = await pathBasedClient[ "/connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors" ].POST({ params: { path: { environment_id: environment_id, kafka_cluster_id: kafka_cluster_id, }, }, body: { name: connectorName, config: { name: connectorName, "kafka.api.key": getEnsuredParam( "KAFKA_API_KEY", "Kafka API Key is required to create the connector. Check if env vars are properly set", ), "kafka.api.secret": getEnsuredParam( "KAFKA_API_SECRET", "Kafka API Secret is required to create the connector. Check if env vars are properly set", ), ...connectorConfig, }, }, }); if (error) { return this.createResponse( `Failed to create connector ${connectorName}: ${JSON.stringify(error)}`, true, ); } return this.createResponse( `${connectorName} created: ${JSON.stringify(response)}`, ); }
• src/confluent/tools/handlers/connect/create-connector-handler.ts:14-75 (schema)

  Zod schema defining the input parameters for the create-connector tool, including baseUrl, environmentId, clusterId, connectorName, and connectorConfig with specific fields for managed and custom connectors.
  const createConnectorArguments = z.object({ baseUrl: z .string() .trim() .describe("The base URL of the Kafka Connect REST API.") .url() .default(() => env.CONFLUENT_CLOUD_REST_ENDPOINT ?? "") .optional(), environmentId: z .string() .optional() .describe( "The unique identifier for the environment this resource belongs to.", ), clusterId: z .string() .optional() .describe("The unique identifier for the Kafka cluster."), connectorName: z .string() .nonempty() .describe("The name of the connector to create."), connectorConfig: z .object({ // Required fields "connector.class": z .string() .describe( "Required for Managed Connector, Ignored for Custom Connector. The connector class name, e.g., BigQuerySink, GcsSink, etc.", ), // Optional fields "confluent.connector.type": z .string() .default("MANAGED") .describe("Required for Custom Connector. The connector type") .optional(), "confluent.custom.plugin.id": z .string() .describe( "Required for Custom Connector. The custom plugin id of custom connector", ) .optional(), "confluent.custom.connection.endpoints": z .string() .describe( "Optional for Custom Connector. Egress endpoint(s) for the connector", ) .optional(), "confluent.custom.schema.registry.auto": z .string() .default("FALSE") .describe( "Optional for Custom Connector. Automatically add required schema registry properties", ) .optional(), }) // Allow additional string properties .catchall(z.string()), });
• src/confluent/tools/tool-factory.ts:53-53 (registration)

  Registration of the CreateConnectorHandler instance in the ToolFactory's static handlers Map under the CREATE_CONNECTOR key.
  [ToolName.CREATE_CONNECTOR, new CreateConnectorHandler()],
• src/confluent/tools/tool-name.ts:13-13 (helper)

  Enum value defining the string name 'create-connector' for the tool.
  CREATE_CONNECTOR = "create-connector",