diff --git a/internal/impl/pure/bloblang_string.go b/internal/impl/pure/bloblang_string.go
index 55ced15ce..7c70c0951 100644
--- a/internal/impl/pure/bloblang_string.go
+++ b/internal/impl/pure/bloblang_string.go
@@ -1,15 +1,18 @@
 package pure
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"net/url"
 
 	"github.com/redpanda-data/benthos/v4/internal/bloblang/query"
+	"github.com/redpanda-data/benthos/v4/internal/bundle"
+	"github.com/redpanda-data/benthos/v4/internal/config"
+	"github.com/redpanda-data/benthos/v4/internal/docs"
 	"github.com/redpanda-data/benthos/v4/public/bloblang"
 )
 
-// var compressAlgorithms = map[string]
-
 func init() {
 	if err := bloblang.RegisterMethodV2("parse_form_url_encoded",
 		bloblang.NewPluginSpec().
@@ -32,6 +35,71 @@ func init() {
 		}); err != nil {
 		panic(err)
 	}
+
+	if err := bloblang.RegisterMethodV2("lint_yaml_config",
+		bloblang.NewPluginSpec().
+			Category(query.MethodCategoryParsing).
+			Version("4.30.1").
+			Beta().
+			Description(`Lints a yaml configuration and returns the linting errors if any.`).
+			Param(bloblang.NewBoolParam("deprecated").Description("Emit linting errors for the presence of deprecated fields.").Default(false)).
+			Param(bloblang.NewBoolParam("require_labels").Description("Emit linting errors when components do not have labels.").Default(false)).
+			Param(bloblang.NewBoolParam("skip_env_var_check").Description("Suppress lint errors when environment interpolations exist without defaults within configs but aren't defined.").Default(false)).
+			Example("", `root = content().lint_yaml_config()`,
+				[2]string{
+					`input:
+  generate:
+    count: 1
+`,
+					`["(3,1) field mapping is required"]`,
+				},
+			),
+		func(args *bloblang.ParsedParams) (bloblang.Method, error) {
+			linterConf := docs.NewLintConfig(bundle.GlobalEnvironment)
+
+			if deprecated, err := args.GetOptionalBool("deprecated"); err != nil {
+				return nil, err
+			} else {
+				linterConf.RejectDeprecated = *deprecated
+			}
+			if requireLabels, err := args.GetOptionalBool("require_labels"); err != nil {
+				return nil, err
+			} else {
+				linterConf.RequireLabels = *requireLabels
+			}
+
+			skipEnvVarCheck, err := args.GetOptionalBool("skip_env_var_check")
+			if err != nil {
+				return nil, err
+			}
+
+			return bloblang.BytesMethod(func(data []byte) (any, error) {
+				var outputLints []any
+				if !*skipEnvVarCheck {
+					if _, err := config.NewReader("", nil).ReplaceEnvVariables(context.Background(), data); err != nil {
+						var errEnvMissing *config.ErrMissingEnvVars
+						if errors.As(err, &errEnvMissing) {
+							outputLints = append(outputLints, docs.NewLintError(1, docs.LintMissingEnvVar, err).Error())
+						} else {
+							return nil, fmt.Errorf("failed to replace env vars: %w", err)
+						}
+					}
+				}
+
+				configLints, err := config.LintYAMLBytes(linterConf, data)
+				if err != nil {
+					return nil, fmt.Errorf("failed to parse yaml: %w", err)
+				}
+
+				for _, lint := range configLints {
+					outputLints = append(outputLints, lint.Error())
+				}
+
+				return outputLints, nil
+			}), nil
+		}); err != nil {
+		panic(err)
+	}
 }
 
 func urlValuesToMap(values url.Values) map[string]any {
diff --git a/internal/impl/pure/bloblang_string_test.go b/internal/impl/pure/bloblang_string_test.go
index e99bd2914..59657cb35 100644
--- a/internal/impl/pure/bloblang_string_test.go
+++ b/internal/impl/pure/bloblang_string_test.go
@@ -63,3 +63,108 @@ func TestParseUrlencoded(t *testing.T) {
 		})
 	}
 }
+
+func TestLintYAMLConfig(t *testing.T) {
+	testCases := []struct {
+		name     string
+		method   string
+		target   any
+		args     []any
+		exp      any
+		expError string
+	}{
+		{
+			name:   "lints yaml configs",
+			method: "lint_yaml_config",
+			target: `input:
+  generate:
+    count: 1
+    mapping: root.foo = "bar"
+`,
+			args: []any{},
+			exp:  []any(nil),
+		},
+		{
+			name:   "rejects invalid yaml configs with both spaces and tabs as indentation",
+			method: "lint_yaml_config",
+			target: `input:
+  generate:
+    count: 1
+	mapping: root.foo = "bar"
+`,
+			args:     []any{},
+			exp:      nil,
+			expError: "failed to parse yaml: yaml: line 3: found a tab character that violates indentation",
+		},
+		{
+			name:   "lints yaml configs with deprecated bloblang methods",
+			method: "lint_yaml_config",
+			target: `input:
+  generate:
+    count: 1
+    mapping: root.ts = 666.format_timestamp()
+`,
+			args: []any{true},
+			exp:  []any(nil), // TODO: THIS SHOULD FAIL!
+		},
+		{
+			name:   "lints yaml configs with missing labels",
+			method: "lint_yaml_config",
+			target: `input:
+  generate:
+    count: 1
+    mapping: root.foo = "bar"
+`,
+			args: []any{false, true},
+			exp:  []any{"(2,1) label is required for generate"},
+		},
+		{
+			name:   "lints yaml configs with unset environment variables",
+			method: "lint_yaml_config",
+			target: `input:
+  generate:
+    count: ${TESTVAR}
+    mapping: root.foo = "bar"
+`,
+			args: []any{false, false, false},
+			exp:  []any{"(1,1) required environment variables were not set: [TESTVAR]"},
+		},
+		{
+			name:   "lints yaml configs with unset environment variables which have a default value",
+			method: "lint_yaml_config",
+			target: `input:
+  generate:
+    count: ${TESTVAR:blobfish}
+    mapping: root.foo = "bar"
+`,
+			args: []any{false, false, false},
+			exp:  []any(nil),
+		},
+	}
+
+	for _, test := range testCases {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			targetClone := value.IClone(test.target)
+			argsClone := value.IClone(test.args).([]any)
+
+			fn, err := query.InitMethodHelper(test.method, query.NewLiteralFunction("", targetClone), argsClone...)
+			require.NoError(t, err)
+
+			res, err := fn.Exec(query.FunctionContext{
+				Maps:     map[string]query.Function{},
+				Index:    0,
+				MsgBatch: nil,
+			})
+			if test.expError != "" {
+				require.ErrorContains(t, err, test.expError)
+			} else {
+				require.NoError(t, err)
+			}
+
+			assert.Equal(t, test.exp, res)
+			assert.Equal(t, test.target, targetClone)
+			assert.Equal(t, test.args, argsClone)
+		})
+	}
+}