diff --git a/src/cli.rs b/src/cli.rs
index 6d21421d2a1c1..a8ec39fd8df7b 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -9,7 +9,7 @@ use crate::service;
 use crate::tap;
 #[cfg(feature = "api-client")]
 use crate::top;
-use crate::{config, generate, get_version, graph, list, unit_test, validate};
+use crate::{config, convert_config, generate, get_version, graph, list, unit_test, validate};
 use crate::{generate_schema, signal};
 
 #[derive(Parser, Debug)]
@@ -34,6 +34,7 @@ impl Opts {
             Some(SubCommand::Validate(_))
             | Some(SubCommand::Graph(_))
             | Some(SubCommand::Generate(_))
+            | Some(SubCommand::ConvertConfig(_))
             | Some(SubCommand::List(_))
             | Some(SubCommand::Test(_)) => {
                 if self.root.verbose == 0 {
@@ -241,6 +242,14 @@ pub enum SubCommand {
     /// Validate the target config, then exit.
     Validate(validate::Opts),
 
+    /// Convert a config file from one format to another.
+    /// This command can also walk directories recursively and convert all config files that are discovered.
+    /// Note that this is a best effort conversion due to the following reasons:
+    /// * The comments from the original config file are not preserved.
+    /// * Explicitly set default values in the original implementation might be omitted.
+    /// * Depending on how each source/sink config struct configures serde, there might be entries with null values.
+    ConvertConfig(convert_config::Opts),
+
     /// Generate a Vector configuration containing a list of components.
     Generate(generate::Opts),
 
@@ -290,6 +299,7 @@ impl SubCommand {
     ) -> exitcode::ExitCode {
         match self {
             Self::Config(c) => config::cmd(c),
+            Self::ConvertConfig(opts) => convert_config::cmd(opts),
             Self::Generate(g) => generate::cmd(g),
             Self::GenerateSchema => generate_schema::cmd(),
             Self::Graph(g) => graph::cmd(g),
diff --git a/src/config/format.rs b/src/config/format.rs
index 0c7dd68442ef1..1f21dd18da3e4 100644
--- a/src/config/format.rs
+++ b/src/config/format.rs
@@ -2,6 +2,7 @@
 
 #![deny(missing_docs, missing_debug_implementations)]
 
+use std::fmt;
 use std::path::Path;
 use std::str::FromStr;
 
@@ -35,6 +36,17 @@ impl FromStr for Format {
     }
 }
 
+impl fmt::Display for Format {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        let format = match self {
+            Format::Toml => "toml",
+            Format::Json => "json",
+            Format::Yaml => "yaml",
+        };
+        write!(f, "{}", format)
+    }
+}
+
 impl Format {
     /// Obtain the format from the file path using extension as a hint.
     pub fn from_path<T: AsRef<Path>>(path: T) -> Result<Self, T> {
diff --git a/src/config/mod.rs b/src/config/mod.rs
index 86b7f7dde7238..c65de18da77d4 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -9,6 +9,7 @@ use std::{
 };
 
 use indexmap::IndexMap;
+use serde::Serialize;
 pub use vector_config::component::{GenerateConfig, SinkDescription, TransformDescription};
 use vector_config::configurable_component;
 pub use vector_core::config::{
@@ -100,7 +101,7 @@ impl ConfigPath {
     }
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Serialize)]
 pub struct Config {
     #[cfg(feature = "api")]
     pub api: api::Options,
diff --git a/src/convert_config.rs b/src/convert_config.rs
new file mode 100644
index 0000000000000..516d123a5245c
--- /dev/null
+++ b/src/convert_config.rs
@@ -0,0 +1,291 @@
+use crate::config::{format, ConfigBuilder, Format};
+use clap::Parser;
+use colored::*;
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::str::FromStr;
+
+#[derive(Parser, Debug)]
+#[command(rename_all = "kebab-case")]
+pub struct Opts {
+    /// The input path. It can be a single file or a directory. If this points to a directory,
+    /// all files with a "toml", "yaml" or "json" extension will be converted.
+    pub(crate) input_path: PathBuf,
+
+    /// The output file or directory to be created. This command will fail if the output directory exists.
+    pub(crate) output_path: PathBuf,
+
+    /// The target format to which existing config files will be converted to.
+    #[arg(long, default_value = "yaml")]
+    pub(crate) output_format: Format,
+}
+
+fn check_paths(opts: &Opts) -> Result<(), String> {
+    let in_metadata = fs::metadata(&opts.input_path)
+        .unwrap_or_else(|_| panic!("Failed to get metadata for: {:?}", &opts.input_path));
+
+    if opts.output_path.exists() {
+        return Err(format!(
+            "Output path {:?} already exists. Please provide a non-existing output path.",
+            opts.output_path
+        ));
+    }
+
+    if opts.output_path.extension().is_none() {
+        if in_metadata.is_file() {
+            return Err(format!(
+                "{:?} points to a file but {:?} points to a directory.",
+                opts.input_path, opts.output_path
+            ));
+        }
+    } else if in_metadata.is_dir() {
+        return Err(format!(
+            "{:?} points to a directory but {:?} points to a file.",
+            opts.input_path, opts.output_path
+        ));
+    }
+
+    Ok(())
+}
+
+pub(crate) fn cmd(opts: &Opts) -> exitcode::ExitCode {
+    if let Err(e) = check_paths(opts) {
+        #[allow(clippy::print_stderr)]
+        {
+            eprintln!("{}", e.red());
+        }
+        return exitcode::SOFTWARE;
+    }
+
+    return if opts.input_path.is_file() && opts.output_path.extension().is_some() {
+        if let Some(base_dir) = opts.output_path.parent() {
+            if !base_dir.exists() {
+                fs::create_dir_all(base_dir).unwrap_or_else(|_| {
+                    panic!("Failed to create output dir(s): {:?}", &opts.output_path)
+                });
+            }
+        }
+
+        match convert_config(&opts.input_path, &opts.output_path, opts.output_format) {
+            Ok(_) => exitcode::OK,
+            Err(errors) => {
+                #[allow(clippy::print_stderr)]
+                {
+                    errors.iter().for_each(|e| eprintln!("{}", e.red()));
+                }
+                exitcode::SOFTWARE
+            }
+        }
+    } else {
+        match walk_dir_and_convert(&opts.input_path, &opts.output_path, opts.output_format) {
+            Ok(()) => {
+                #[allow(clippy::print_stdout)]
+                {
+                    println!(
+                        "Finished conversion(s). Results are in {:?}",
+                        opts.output_path
+                    );
+                }
+                exitcode::OK
+            }
+            Err(errors) => {
+                #[allow(clippy::print_stderr)]
+                {
+                    errors.iter().for_each(|e| eprintln!("{}", e.red()));
+                }
+                exitcode::SOFTWARE
+            }
+        }
+    };
+}
+
+fn convert_config(
+    input_path: &Path,
+    output_path: &Path,
+    output_format: Format,
+) -> Result<(), Vec<String>> {
+    if output_path.exists() {
+        return Err(vec![format!("Output path {output_path:?} exists")]);
+    }
+    let input_format = match Format::from_str(
+        input_path
+            .extension()
+            .unwrap_or_else(|| panic!("Failed to get extension for: {input_path:?}"))
+            .to_str()
+            .unwrap_or_else(|| panic!("Failed to convert OsStr to &str for: {input_path:?}")),
+    ) {
+        Ok(format) => format,
+        Err(_) => return Ok(()), // skip irrelevant files
+    };
+
+    if input_format == output_format {
+        return Ok(());
+    }
+
+    #[allow(clippy::print_stdout)]
+    {
+        println!("Converting {input_path:?} config to {output_format:?}.");
+    }
+    let file_contents = fs::read_to_string(input_path).map_err(|e| vec![e.to_string()])?;
+    let builder: ConfigBuilder = format::deserialize(&file_contents, input_format)?;
+    let config = builder.build()?;
+    let output_string =
+        format::serialize(&config, output_format).map_err(|e| vec![e.to_string()])?;
+    fs::write(output_path, output_string).map_err(|e| vec![e.to_string()])?;
+
+    #[allow(clippy::print_stdout)]
+    {
+        println!("Wrote result to {output_path:?}.");
+    }
+    Ok(())
+}
+
+fn walk_dir_and_convert(
+    input_path: &Path,
+    output_dir: &Path,
+    output_format: Format,
+) -> Result<(), Vec<String>> {
+    let mut errors = Vec::new();
+
+    if input_path.is_dir() {
+        for entry in fs::read_dir(input_path)
+            .unwrap_or_else(|_| panic!("Failed to read dir: {input_path:?}"))
+        {
+            let entry_path = entry
+                .unwrap_or_else(|_| panic!("Failed to get entry for dir: {input_path:?}"))
+                .path();
+            let new_output_dir = if entry_path.is_dir() {
+                let last_component = entry_path
+                    .file_name()
+                    .unwrap_or_else(|| panic!("Failed to get file_name for {entry_path:?}"))
+                    .clone();
+                let new_dir = output_dir.join(last_component);
+
+                if !new_dir.exists() {
+                    fs::create_dir_all(&new_dir)
+                        .unwrap_or_else(|_| panic!("Failed to create output dir: {new_dir:?}"));
+                }
+                new_dir
+            } else {
+                output_dir.to_path_buf()
+            };
+
+            if let Err(new_errors) = walk_dir_and_convert(
+                &input_path.join(&entry_path),
+                &new_output_dir,
+                output_format,
+            ) {
+                errors.extend(new_errors);
+            }
+        }
+    } else {
+        let output_path = output_dir.join(
+            input_path
+                .with_extension(output_format.to_string().as_str())
+                .file_name()
+                .ok_or_else(|| {
+                    vec![format!(
+                        "Cannot create output path for input: {input_path:?}"
+                    )]
+                })?,
+        );
+        if let Err(new_errors) = convert_config(input_path, &output_path, output_format) {
+            errors.extend(new_errors);
+        }
+    }
+
+    if errors.is_empty() {
+        Ok(())
+    } else {
+        Err(errors)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::config::{format, ConfigBuilder, Format};
+    use crate::convert_config::{check_paths, walk_dir_and_convert, Opts};
+    use std::path::{Path, PathBuf};
+    use std::str::FromStr;
+    use std::{env, fs};
+    use tempfile::tempdir;
+
+    fn test_data_dir() -> PathBuf {
+        PathBuf::from(env::var_os("CARGO_MANIFEST_DIR").unwrap()).join("tests/data/cmd/config")
+    }
+
+    // Read the contents of the specified `path` and deserialize them into a `ConfigBuilder`.
+    // Finally serialize them a string again. Configs do not implement equality,
+    // so for these tests we will rely on strings for comparisons.
+    fn convert_file_to_config_string(path: &Path) -> String {
+        let files_contents = fs::read_to_string(path).unwrap();
+        let extension = path.extension().unwrap().to_str().unwrap();
+        let file_format = Format::from_str(extension).unwrap();
+        let builder: ConfigBuilder = format::deserialize(&files_contents, file_format).unwrap();
+        let config = builder.build().unwrap();
+
+        format::serialize(&config, file_format).unwrap()
+    }
+
+    #[test]
+    fn invalid_path_opts() {
+        let check_error = |opts, pattern| {
+            let error = check_paths(&opts).unwrap_err();
+            assert!(error.contains(pattern));
+        };
+
+        check_error(
+            Opts {
+                input_path: ["./"].iter().collect(),
+                output_path: ["./"].iter().collect(),
+                output_format: Format::Yaml,
+            },
+            "already exists",
+        );
+
+        check_error(
+            Opts {
+                input_path: ["./"].iter().collect(),
+                output_path: ["./out.yaml"].iter().collect(),
+                output_format: Format::Yaml,
+            },
+            "points to a file.",
+        );
+
+        check_error(
+            Opts {
+                input_path: [test_data_dir(), "config_2.toml".into()].iter().collect(),
+                output_path: ["./another_dir"].iter().collect(),
+                output_format: Format::Yaml,
+            },
+            "points to a directory.",
+        );
+    }
+
+    #[test]
+    fn convert_all_from_dir() {
+        let input_path = test_data_dir();
+        let output_dir = tempdir()
+            .expect("Unable to create tempdir for config")
+            .into_path();
+        walk_dir_and_convert(&input_path, &output_dir, Format::Yaml).unwrap();
+
+        let mut count: usize = 0;
+        let original_config = convert_file_to_config_string(&test_data_dir().join("config_1.yaml"));
+        for entry in fs::read_dir(&output_dir).unwrap() {
+            let entry = entry.unwrap();
+            let path = entry.path();
+            if path.is_file() {
+                let extension = path.extension().unwrap().to_str().unwrap();
+                if extension == Format::Yaml.to_string() {
+                    // Note that here we read the converted string directly.
+                    let converted_config = fs::read_to_string(&output_dir.join(&path)).unwrap();
+                    assert_eq!(converted_config, original_config);
+                    count += 1;
+                }
+            }
+        }
+        // There two non-yaml configs in the input directory.
+        assert_eq!(count, 2);
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index d16cc17ce9a46..c5ab3f1692955 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -67,6 +67,7 @@ pub mod aws;
 #[allow(unreachable_pub)]
 pub mod codecs;
 pub(crate) mod common;
+mod convert_config;
 pub mod encoding_transcode;
 pub mod enrichment_tables;
 #[cfg(feature = "gcp")]
diff --git a/tests/data/cmd/config/config_1.yaml b/tests/data/cmd/config/config_1.yaml
new file mode 100644
index 0000000000000..bb523436f65ce
--- /dev/null
+++ b/tests/data/cmd/config/config_1.yaml
@@ -0,0 +1,36 @@
+data_dir: /var/lib/vector/
+sources:
+  source0:
+    count: 9223372036854775807
+    format: json
+    interval: 1.0
+    type: demo_logs
+    decoding:
+      codec: bytes
+    framing:
+      method: bytes
+transforms:
+  transform0:
+    inputs:
+      - source0
+    drop_on_abort: false
+    drop_on_error: false
+    metric_tag_values: single
+    reroute_dropped: false
+    runtime: ast
+    type: remap
+sinks:
+  sink0:
+    inputs:
+      - transform0
+    target: stdout
+    type: console
+    encoding:
+      codec: json
+    healthcheck:
+      enabled: true
+      uri: null
+    buffer:
+      type: memory
+      max_events: 500
+      when_full: block
diff --git a/tests/data/cmd/config/config_2.toml b/tests/data/cmd/config/config_2.toml
new file mode 100644
index 0000000000000..fc46fc50f8ad0
--- /dev/null
+++ b/tests/data/cmd/config/config_2.toml
@@ -0,0 +1,43 @@
+# Comments are dropped.
+#[api]
+#enabled = true
+#address = "127.0.0.1:8686"
+
+data_dir = "/var/lib/vector/"
+
+[sources.source0]
+count = 9223372036854775807
+format = "json"
+interval = 1.0
+type = "demo_logs"
+
+[sources.source0.decoding]
+codec = "bytes"
+
+[sources.source0.framing]
+method = "bytes"
+
+[transforms.transform0]
+inputs = ["source0"]
+drop_on_abort = false
+drop_on_error = false
+metric_tag_values = "single"
+reroute_dropped = false
+runtime = "ast"
+type = "remap"
+
+[sinks.sink0]
+inputs = ["transform0"]
+target = "stdout"
+type = "console"
+
+[sinks.sink0.encoding]
+codec = "json"
+
+[sinks.sink0.healthcheck]
+enabled = true
+
+[sinks.sink0.buffer]
+type = "memory"
+max_events = 500
+when_full = "block"
diff --git a/tests/data/cmd/config/config_3.json b/tests/data/cmd/config/config_3.json
new file mode 100644
index 0000000000000..3517f16295177
--- /dev/null
+++ b/tests/data/cmd/config/config_3.json
@@ -0,0 +1,51 @@
+{
+  "data_dir": "/var/lib/vector/",
+  "sources": {
+    "source0": {
+      "count": 9223372036854775807,
+      "format": "json",
+      "interval": 1.0,
+      "type": "demo_logs",
+      "decoding": {
+        "codec": "bytes"
+      },
+      "framing": {
+        "method": "bytes"
+      }
+    }
+  },
+  "transforms": {
+    "transform0": {
+      "inputs": [
+        "source0"
+      ],
+      "drop_on_abort": false,
+      "drop_on_error": false,
+      "metric_tag_values": "single",
+      "reroute_dropped": false,
+      "runtime": "ast",
+      "type": "remap"
+    }
+  },
+  "sinks": {
+    "sink0": {
+      "inputs": [
+        "transform0"
+      ],
+      "target": "stdout",
+      "type": "console",
+      "encoding": {
+        "codec": "json"
+      },
+      "healthcheck": {
+        "enabled": true,
+        "uri": null
+      },
+      "buffer": {
+        "type": "memory",
+        "max_events": 500,
+        "when_full": "block"
+      }
+    }
+  }
+}