Skip to content

Runtime

Helpers for loading configuration bundles and running guardrails.

This module is the bridge between configuration and runtime execution for guardrail validation. It provides pure helpers for loading config bundles, instantiating guardrails from registry specs, and orchestrating parallel execution of guardrail checks. All logic is pure and side-effect-free except for I/O during config reading.

ConfiguredGuardrail dataclass

Bases: Generic[TContext, TIn, TCfg]

A configured, executable guardrail.

This class binds a GuardrailSpec definition to a validated configuration object. The resulting instance is used to run guardrail logic in production pipelines. It supports both sync and async check functions.

Attributes:

Name Type Description
definition GuardrailSpec[TContext, TIn, TCfg]

The immutable guardrail specification.

config TCfg

Validated user configuration for this instance.

Source code in src/guardrails/runtime.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@dataclass(frozen=True, slots=True)
class ConfiguredGuardrail(Generic[TContext, TIn, TCfg]):
    """A configured, executable guardrail.

    This class binds a `GuardrailSpec` definition to a validated configuration
    object. The resulting instance is used to run guardrail logic in production
    pipelines. It supports both sync and async check functions.

    Attributes:
        definition (GuardrailSpec[TContext, TIn, TCfg]): The immutable guardrail specification.
        config (TCfg): Validated user configuration for this instance.
    """

    definition: GuardrailSpec[TContext, TIn, TCfg]
    config: TCfg

    async def _ensure_async(
        self,
        fn: Callable[P, MaybeAwaitableResult],
        *a: P.args,
        **kw: P.kwargs,
    ) -> GuardrailResult:
        """Ensure a guardrail function is executed asynchronously.

        If the function is sync, runs it in a thread for compatibility with async flows.
        If already async, simply awaits it. Used internally to normalize execution style.

        Args:
            fn: Guardrail check function (sync or async).
            *a: Positional arguments for the check function.
            **kw: Keyword arguments for the check function.

        Returns:
            GuardrailResult: The result of the check function.
        """
        result = fn(*a, **kw)
        if inspect.isawaitable(result):
            return await result
        return cast("GuardrailResult", await asyncio.to_thread(lambda: result))

    async def run(self, ctx: TContext, data: TIn) -> GuardrailResult:
        """Run the guardrail's check function with the provided context and data.

        Main entry point for executing guardrails. Supports both sync and async
        functions, ensuring results are always awaited.

        Args:
            ctx (TContext): Runtime context for the guardrail.
            data (TIn): Input value to be checked.

        Returns:
            GuardrailResult: The outcome of the guardrail logic.
        """
        return await self._ensure_async(
            self.definition.check_fn, ctx, data, self.config
        )

run async

run(ctx: TContext, data: TIn) -> GuardrailResult

Run the guardrail's check function with the provided context and data.

Main entry point for executing guardrails. Supports both sync and async functions, ensuring results are always awaited.

Parameters:

Name Type Description Default
ctx TContext

Runtime context for the guardrail.

required
data TIn

Input value to be checked.

required

Returns:

Name Type Description
GuardrailResult GuardrailResult

The outcome of the guardrail logic.

Source code in src/guardrails/runtime.py
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def run(self, ctx: TContext, data: TIn) -> GuardrailResult:
    """Run the guardrail's check function with the provided context and data.

    Main entry point for executing guardrails. Supports both sync and async
    functions, ensuring results are always awaited.

    Args:
        ctx (TContext): Runtime context for the guardrail.
        data (TIn): Input value to be checked.

    Returns:
        GuardrailResult: The outcome of the guardrail logic.
    """
    return await self._ensure_async(
        self.definition.check_fn, ctx, data, self.config
    )

GuardrailConfig

Bases: BaseModel

Configuration for a single guardrail instance.

Used for serializing, deserializing, and validating guardrail configs as part of a bundle.

Attributes:

Name Type Description
name str

The registry name used to look up the guardrail spec.

config dict[str, Any]

Raw user configuration for this guardrail.

Source code in src/guardrails/runtime.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class GuardrailConfig(BaseModel):
    """Configuration for a single guardrail instance.

    Used for serializing, deserializing, and validating guardrail configs as part
    of a bundle.

    Attributes:
        name (str): The registry name used to look up the guardrail spec.
        config (dict[str, Any]): Raw user configuration for this guardrail.
    """

    name: str
    config: dict[str, Any]

ConfigBundle

Bases: BaseModel

Versioned collection of configured guardrails.

Represents a serializable "bundle" of guardrails to be run as a unit. Suitable for JSON storage and loading.

Attributes:

Name Type Description
guardrails list[GuardrailConfig]

The configured guardrails.

version int

Format version for forward/backward compatibility.

stage_name str

User-defined name for the pipeline stage this bundle is for. This can be any string that helps identify which part of your pipeline triggered the guardrail (e.g., "user_input_validation", "content_generation", "pre_processing", etc.). It will be included in GuardrailResult info for easy identification.

config dict[str, Any]

Execution configuration for this bundle. Optional fields include: - concurrency (int): Maximum number of guardrails to run in parallel (default: 10) - suppress_tripwire (bool): If True, don't raise exceptions on tripwires (default: False)

Source code in src/guardrails/runtime.py
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class ConfigBundle(BaseModel):
    """Versioned collection of configured guardrails.

    Represents a serializable "bundle" of guardrails to be run as a unit.
    Suitable for JSON storage and loading.

    Attributes:
        guardrails (list[GuardrailConfig]): The configured guardrails.
        version (int): Format version for forward/backward compatibility.
        stage_name (str): User-defined name for the pipeline stage this bundle is for.
            This can be any string that helps identify which part of your pipeline
            triggered the guardrail (e.g., "user_input_validation", "content_generation",
            "pre_processing", etc.). It will be included in GuardrailResult info for
            easy identification.
        config (dict[str, Any]): Execution configuration for this bundle.
            Optional fields include:
            - concurrency (int): Maximum number of guardrails to run in parallel (default: 10)
            - suppress_tripwire (bool): If True, don't raise exceptions on tripwires (default: False)
    """

    guardrails: list[GuardrailConfig]
    version: int = 1
    stage_name: str = "unnamed"
    config: dict[str, Any] = {}

    model_config = ConfigDict(frozen=True, extra="forbid")

PipelineBundles

Bases: BaseModel

Three-stage collection of validated ConfigBundles for an LLM pipeline.

This class groups together guardrail configurations for the three main pipeline stages:

  • pre_flight: Checks that run before the LLM request is issued.
  • input: Checks on the user's prompt (may run concurrently with LLM call).
  • output: Checks on incremental LLM output.

At least one stage must be provided, but all stages are optional.

Attributes:

Name Type Description
pre_flight ConfigBundle | None

Guardrails to run before the LLM request.

input ConfigBundle | None

Guardrails to run on user input.

output ConfigBundle | None

Guardrails to run on generated output.

version int

Schema version for the envelope itself. Defaults to 1.

Example
# All stages
pipeline = PipelineBundles(
    pre_flight=load_config_bundle(PRE_FLIGHT),
    input=load_config_bundle(INPUT_BUNDLE),
    output=load_config_bundle(OUTPUT_BUNDLE),
)

# Just output stage
pipeline = PipelineBundles(
    output=load_config_bundle(OUTPUT_BUNDLE),
)

# Print active stages
for stage in pipeline.stages():
    print(stage.version)
Source code in src/guardrails/runtime.py
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class PipelineBundles(BaseModel):
    """Three-stage collection of validated ConfigBundles for an LLM pipeline.

    This class groups together guardrail configurations for the three main
    pipeline stages:

    - pre_flight: Checks that run before the LLM request is issued.
    - input: Checks on the user's prompt (may run concurrently with LLM call).
    - output: Checks on incremental LLM output.

    At least one stage must be provided, but all stages are optional.

    Attributes:
        pre_flight (ConfigBundle | None): Guardrails to run before the LLM request.
        input (ConfigBundle | None): Guardrails to run on user input.
        output (ConfigBundle | None): Guardrails to run on generated output.
        version (int): Schema version for the envelope itself. Defaults to 1.

    Example:
        ```python
        # All stages
        pipeline = PipelineBundles(
            pre_flight=load_config_bundle(PRE_FLIGHT),
            input=load_config_bundle(INPUT_BUNDLE),
            output=load_config_bundle(OUTPUT_BUNDLE),
        )

        # Just output stage
        pipeline = PipelineBundles(
            output=load_config_bundle(OUTPUT_BUNDLE),
        )

        # Print active stages
        for stage in pipeline.stages():
            print(stage.version)
        ```
    """

    pre_flight: ConfigBundle | None = None
    input: ConfigBundle | None = None
    output: ConfigBundle | None = None
    version: int = 1

    model_config = ConfigDict(frozen=True, extra="forbid")

    _STAGE_ORDER: Final[tuple[str, ...]] = ("pre_flight", "input", "output")

    def model_post_init(self, __context: Any) -> None:
        """Validate that at least one stage is provided."""
        if not any(getattr(self, stage) is not None for stage in self._STAGE_ORDER):
            raise ValueError(
                "At least one stage (pre_flight, input, or output) must be provided"
            )

    def stages(self) -> tuple[ConfigBundle, ...]:
        """Return non-None bundles in execution order (pre_flight → input → output)."""
        return tuple(
            bundle
            for name in self._STAGE_ORDER
            if (bundle := getattr(self, name)) is not None
        )

model_post_init

model_post_init(__context: Any) -> None

Validate that at least one stage is provided.

Source code in src/guardrails/runtime.py
185
186
187
188
189
190
def model_post_init(self, __context: Any) -> None:
    """Validate that at least one stage is provided."""
    if not any(getattr(self, stage) is not None for stage in self._STAGE_ORDER):
        raise ValueError(
            "At least one stage (pre_flight, input, or output) must be provided"
        )

stages

stages() -> tuple[ConfigBundle, ...]

Return non-None bundles in execution order (pre_flight → input → output).

Source code in src/guardrails/runtime.py
192
193
194
195
196
197
198
def stages(self) -> tuple[ConfigBundle, ...]:
    """Return non-None bundles in execution order (pre_flight → input → output)."""
    return tuple(
        bundle
        for name in self._STAGE_ORDER
        if (bundle := getattr(self, name)) is not None
    )

JsonString dataclass

Explicit wrapper to mark a string as a raw JSON config.

Used to distinguish JSON string inputs from other config sources (path/dict).

Attributes:

Name Type Description
content str

The raw JSON string.

Source code in src/guardrails/runtime.py
201
202
203
204
205
206
207
208
209
210
211
@dataclass(frozen=True, slots=True)
class JsonString:
    """Explicit wrapper to mark a string as a raw JSON config.

    Used to distinguish JSON string inputs from other config sources (path/dict).

    Attributes:
        content (str): The raw JSON string.
    """

    content: str

load_config_bundle

load_config_bundle(source: ConfigSource) -> ConfigBundle

Load a ConfigBundle from a path, dict, JSON string, or already-parsed object.

Supported sources
  • ConfigBundle: Already validated bundle.
  • dict: Raw data, parsed as ConfigBundle.
  • Path: JSON file on disk.
  • JsonString: Raw JSON string.
Example usage
bundle = load_config_bundle(JsonString('{"guardrails": [...]}'))
bundle = load_config_bundle({"guardrails": [...]})
bundle = load_config_bundle(Path("./config.json"))

Parameters:

Name Type Description Default
source ConfigSource

Config bundle input.

required

Raises:

Type Description
ConfigError

If loading fails.

ValidationError

If model validation fails.

FileNotFoundError

If file doesn't exist.

Returns:

Name Type Description
ConfigBundle ConfigBundle

The loaded and validated configuration bundle.

Source code in src/guardrails/runtime.py
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def load_config_bundle(source: ConfigSource) -> ConfigBundle:
    """Load a ConfigBundle from a path, dict, JSON string, or already-parsed object.

    Supported sources:
      - ConfigBundle: Already validated bundle.
      - dict: Raw data, parsed as ConfigBundle.
      - Path: JSON file on disk.
      - JsonString: Raw JSON string.

    Example usage:
        ```python
        bundle = load_config_bundle(JsonString('{"guardrails": [...]}'))
        bundle = load_config_bundle({"guardrails": [...]})
        bundle = load_config_bundle(Path("./config.json"))
        ```

    Args:
        source (ConfigSource): Config bundle input.

    Raises:
        ConfigError: If loading fails.
        ValidationError: If model validation fails.
        FileNotFoundError: If file doesn't exist.

    Returns:
        ConfigBundle: The loaded and validated configuration bundle.
    """
    return _load_bundle(source, ConfigBundle)

load_pipeline_bundles

load_pipeline_bundles(
    source: PipelineSource,
) -> PipelineBundles

Load a PipelineBundles from a path, dict, JSON string, or already-parsed object.

Supported sources
  • PipelineBundles: Already validated pipeline bundles.
  • dict: Raw data, parsed as PipelineBundles.
  • Path: JSON file on disk.
  • JsonString: Raw JSON string.
Example usage
bundle = load_pipeline_bundles(JsonString('{"pre_flight": {...}, ...}'))
bundle = load_pipeline_bundles({"pre_flight": {...}, ...})
bundle = load_pipeline_bundles(Path("./pipeline.json"))

Parameters:

Name Type Description Default
source PipelineSource

Pipeline bundles input.

required

Raises:

Type Description
ConfigError

If loading fails.

ValidationError

If model validation fails.

FileNotFoundError

If file doesn't exist.

Returns:

Name Type Description
PipelineBundles PipelineBundles

The loaded and validated pipeline configuration bundle.

Source code in src/guardrails/runtime.py
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def load_pipeline_bundles(source: PipelineSource) -> PipelineBundles:
    """Load a PipelineBundles from a path, dict, JSON string, or already-parsed object.

    Supported sources:
      - PipelineBundles: Already validated pipeline bundles.
      - dict: Raw data, parsed as PipelineBundles.
      - Path: JSON file on disk.
      - JsonString: Raw JSON string.

    Example usage:
        ```python
        bundle = load_pipeline_bundles(JsonString('{"pre_flight": {...}, ...}'))
        bundle = load_pipeline_bundles({"pre_flight": {...}, ...})
        bundle = load_pipeline_bundles(Path("./pipeline.json"))
        ```

    Args:
        source (PipelineSource): Pipeline bundles input.

    Raises:
        ConfigError: If loading fails.
        ValidationError: If model validation fails.
        FileNotFoundError: If file doesn't exist.

    Returns:
        PipelineBundles: The loaded and validated pipeline configuration bundle.
    """
    return _load_bundle(source, PipelineBundles)

instantiate_guardrails

instantiate_guardrails(
    bundle: ConfigBundle,
    registry: GuardrailRegistry | None = None,
) -> list[ConfiguredGuardrail[Any, Any, Any]]

Instantiate all configured guardrails in a bundle as executable objects.

This function validates each guardrail configuration, retrieves the spec from the registry, and returns a list of fully configured guardrails.

Parameters:

Name Type Description Default
bundle ConfigBundle

The validated configuration bundle.

required
registry GuardrailRegistry

Registry mapping names to specs. If not provided, defaults to default_spec_registry.

None

Raises:

Type Description
ConfigError

If any individual guardrail config is invalid.

Returns:

Type Description
list[ConfiguredGuardrail[Any, Any, Any]]

list[ConfiguredGuardrail[Any, Any, Any]]: All configured/runnable guardrail objects.

Source code in src/guardrails/runtime.py
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def instantiate_guardrails(
    bundle: ConfigBundle,
    registry: GuardrailRegistry | None = None,
) -> list[ConfiguredGuardrail[Any, Any, Any]]:
    """Instantiate all configured guardrails in a bundle as executable objects.

    This function validates each guardrail configuration, retrieves the spec
    from the registry, and returns a list of fully configured guardrails.

    Args:
        bundle (ConfigBundle): The validated configuration bundle.
        registry (GuardrailRegistry, optional): Registry mapping names to specs.
            If not provided, defaults to `default_spec_registry`.

    Raises:
        ConfigError: If any individual guardrail config is invalid.

    Returns:
        list[ConfiguredGuardrail[Any, Any, Any]]: All configured/runnable guardrail objects.
    """
    registry = registry or default_spec_registry
    logger.debug("Instantiating guardrails using registry %s", registry)
    out: list[ConfiguredGuardrail[Any, Any, Any]] = []
    for item in bundle.guardrails:
        logger.debug("Configuring guardrail '%s'", item.name)
        grd_spec = registry.get(item.name)
        # Validate guardrail-specific config and wrap errors
        try:
            cfg = grd_spec.config_schema.model_validate(item.config)
        except Exception as e:
            logger.error("Guardrail '%s' config validation failed: %s", item.name, e)
            raise ConfigError(
                f"Configuration for guardrail '{item.name}' invalid: {e}",
            ) from e
        out.append(grd_spec.instantiate(config=cfg))
    logger.info("Instantiated %d guardrails", len(out))
    return out

run_guardrails async

run_guardrails(
    ctx: TContext,
    data: TIn,
    media_type: str,
    guardrails: Iterable[
        ConfiguredGuardrail[TContext, TIn, Any]
    ],
    *,
    concurrency: int = 10,
    result_handler: Callable[
        [GuardrailResult], Coroutine[None, None, None]
    ]
    | None = None,
    suppress_tripwire: bool = False,
    stage_name: str | None = None,
    raise_guardrail_errors: bool = False,
) -> list[GuardrailResult]

Run a set of configured guardrails concurrently and collect their results.

Validates context requirements for each guardrail, filters guardrails by the specified media type, and runs each check function concurrently, up to the specified concurrency limit. Results for all executed guardrails are collected and returned in order. If any guardrail triggers a tripwire, the function will raise a GuardrailTripwireTriggered exception unless tripwire suppression is enabled.

An optional asynchronous result handler can be provided to perform side effects (e.g., logging, custom result processing) for each guardrail result as it becomes available.

Parameters:

Name Type Description Default
ctx TContext

Context object passed to all guardrail checks. Must satisfy all required fields specified by each guardrail's context schema.

required
data TIn

The input to be validated by the guardrails.

required
media_type str

MIME type used to filter which guardrails to execute. (e.g. "text/plain")

required
guardrails Iterable[ConfiguredGuardrail[TContext, TIn, Any]]

Iterable of configured guardrails to run.

required
concurrency int

Maximum number of guardrails to run in parallel. Defaults to 10.

10
result_handler Callable[[GuardrailResult], Awaitable[None]]

Asynchronous callback function to be invoked for each guardrail result as it is produced. Defaults to None.

None
suppress_tripwire bool

If True, tripwire-triggered results are included in the returned list and no exception is raised. If False (default), the function aborts and raises GuardrailTripwireTriggered on the first tripwire.

False
stage_name str | None

Name of the pipeline stage (e.g., "pre_flight", "input", "output"). If provided, this will be included in the GuardrailResult info. Defaults to None.

None
raise_guardrail_errors bool

If True, raise exceptions when guardrails fail to execute. If False (default), treat guardrail execution errors as safe and continue execution.

False

Returns:

Type Description
list[GuardrailResult]

list[GuardrailResult]: List of results for all executed guardrails. If tripwire suppression is disabled (default) and a tripwire is triggered, only results up to the first tripwire may be included.

Raises:

Type Description
GuardrailTripwireTriggered

Raised if a guardrail tripwire is triggered and suppress_tripwire is False.

ContextValidationError

Raised if the provided context does not meet requirements for any guardrail being executed.

Example
results = await run_guardrails(
    ctx=my_ctx,
    data="example input",
    media_type="text/plain",
    guardrails=my_guardrails,
    concurrency=4,
    suppress_tripwire=True,
    stage_name="input",
)
Source code in src/guardrails/runtime.py
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
async def run_guardrails(
    ctx: TContext,
    data: TIn,
    media_type: str,
    guardrails: Iterable[ConfiguredGuardrail[TContext, TIn, Any]],
    *,
    concurrency: int = 10,
    result_handler: (
        Callable[[GuardrailResult], Coroutine[None, None, None]] | None
    ) = None,
    suppress_tripwire: bool = False,
    stage_name: str | None = None,
    raise_guardrail_errors: bool = False,
) -> list[GuardrailResult]:
    """Run a set of configured guardrails concurrently and collect their results.

    Validates context requirements for each guardrail, filters guardrails by the specified media type,
    and runs each check function concurrently, up to the specified concurrency limit. Results for all
    executed guardrails are collected and returned in order. If any guardrail triggers a tripwire,
    the function will raise a `GuardrailTripwireTriggered` exception unless tripwire suppression is enabled.

    An optional asynchronous result handler can be provided to perform side effects (e.g., logging,
    custom result processing) for each guardrail result as it becomes available.

    Args:
        ctx (TContext): Context object passed to all guardrail checks. Must satisfy all required
            fields specified by each guardrail's context schema.
        data (TIn): The input to be validated by the guardrails.
        media_type (str): MIME type used to filter which guardrails to execute. (e.g. "text/plain")
        guardrails (Iterable[ConfiguredGuardrail[TContext, TIn, Any]]): Iterable of configured
            guardrails to run.
        concurrency (int, optional): Maximum number of guardrails to run in parallel.
            Defaults to 10.
        result_handler (Callable[[GuardrailResult], Awaitable[None]], optional): Asynchronous
            callback function to be invoked for each guardrail result as it is produced.
            Defaults to None.
        suppress_tripwire (bool, optional): If True, tripwire-triggered results are included in
            the returned list and no exception is raised. If False (default), the function aborts
            and raises `GuardrailTripwireTriggered` on the first tripwire.
        stage_name (str | None, optional): Name of the pipeline stage (e.g., "pre_flight", "input", "output").
            If provided, this will be included in the GuardrailResult info. Defaults to None.
        raise_guardrail_errors (bool, optional): If True, raise exceptions when guardrails fail to execute.
            If False (default), treat guardrail execution errors as safe and continue execution.

    Returns:
        list[GuardrailResult]: List of results for all executed guardrails. If tripwire suppression
            is disabled (default) and a tripwire is triggered, only results up to the first tripwire
            may be included.

    Raises:
        GuardrailTripwireTriggered: Raised if a guardrail tripwire is triggered and
            `suppress_tripwire` is False.
        ContextValidationError: Raised if the provided context does not meet requirements for any
            guardrail being executed.

    Example:
        ```python
        results = await run_guardrails(
            ctx=my_ctx,
            data="example input",
            media_type="text/plain",
            guardrails=my_guardrails,
            concurrency=4,
            suppress_tripwire=True,
            stage_name="input",
        )
        ```
    """  # noqa: E501
    selected = [g for g in guardrails if g.definition.media_type == media_type]
    logger.debug(
        "Running %d guardrails for media-type '%s' with concurrency=%d",
        len(selected),
        media_type,
        concurrency,
    )
    if not selected:
        return []

    for g in selected:
        validate_guardrail_context(g, ctx)

    semaphore = asyncio.Semaphore(concurrency)
    results: list[GuardrailResult] = []
    tripwire_results: list[GuardrailResult] = []

    async def _run_one(
        g: ConfiguredGuardrail[TContext, TIn, Any],
    ) -> GuardrailResult:
        async with semaphore:
            logger.debug("Running guardrail '%s'", g.definition.name)
            try:
                result = await g.run(ctx, data)

                # Always add stage_name to the result info while preserving all fields
                result = GuardrailResult(
                    tripwire_triggered=result.tripwire_triggered,
                    execution_failed=result.execution_failed,
                    original_exception=result.original_exception,
                    info={**result.info, "stage_name": stage_name or "unnamed"}
                )

            except Exception as exc:
                logger.error("Guardrail '%s' failed to execute: %s", g.definition.name, exc)

                if raise_guardrail_errors:
                    # Re-raise the exception to stop execution
                    raise exc
                else:
                    # Create a safe result indicating execution error
                    result = GuardrailResult(
                        tripwire_triggered=False,  # Don't trigger tripwire on execution errors
                        execution_failed=True,
                        original_exception=exc,
                        info={
                            "checked_text": str(data),
                            "stage_name": stage_name or "unnamed",
                            "guardrail_name": g.definition.name,
                            "error": str(exc),
                        }
                    )

            # Invoke user-provided handler for each result
            if result_handler:
                try:
                    await result_handler(result)
                except Exception as handler_err:
                    logger.error(
                        "Error in result_handler for guardrail '%s': %s",
                        g.definition.name,
                        handler_err,
                    )
            if result.tripwire_triggered:
                logger.warning("Tripwire triggered by '%s'", g.definition.name)
            return result

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(_run_one(g)) for g in selected]
    results.extend(task.result() for task in tasks)

    # Check for guardrail execution failures and re-raise if configured
    if raise_guardrail_errors:
        execution_failures = [r for r in results if r.execution_failed]

        if execution_failures:
            # Re-raise the first execution failure
            failure = execution_failures[0]
            logger.debug("Re-raising guardrail execution error due to raise_guardrail_errors=True")
            raise failure.original_exception

    tripwire_results = [r for r in results if r.tripwire_triggered]
    if tripwire_results and not suppress_tripwire:
        raise GuardrailTripwireTriggered(tripwire_results[0])

    logger.info("Completed guardrail run; %d results returned", len(results))
    return results

check_plain_text async

check_plain_text(
    text: str,
    bundle_path: ConfigSource = Path("guardrails.json"),
    registry: GuardrailRegistry | None = None,
    ctx: Any = None,
    **kwargs: Any,
) -> list[GuardrailResult]

Validate plain text input against all configured guardrails for the 'text/plain' media type.

This function loads a guardrail configuration bundle, instantiates all guardrails for the specified registry, and runs each guardrail against the provided text input. It is the recommended entry point for validating plain text with one or more guardrails in an async context.

If no context object (ctx) is provided, a minimal default context will be constructed with attributes required by the guardrails' context schema (for example, an OpenAI client and any required fields with safe default values). For advanced use cases, you can supply your own context object.

Keyword arguments are forwarded to run_guardrails, allowing you to control concurrency, provide a result handler, or suppress tripwire exceptions.

Parameters:

Name Type Description Default
text str

The plain text input to validate.

required
bundle_path ConfigSource

Guardrail configuration bundle. This can be a file path, dict, JSON string, or ConfigBundle instance. Defaults to "guardrails.json".

Path('guardrails.json')
registry GuardrailRegistry

Guardrail registry to use for instantiation. If not provided, the default registry is used.

None
ctx Any

Application context object passed to each guardrail. If None (default), a minimal default context will be used.

None
**kwargs Any

Additional keyword arguments forwarded to run_guardrails. Common options include:

  • concurrency (int): Maximum number of guardrails to run concurrently.
  • result_handler (Callable[[GuardrailResult], Awaitable[None]]): Async function invoked for each result.
  • suppress_tripwire (bool): If True, do not raise an exception on tripwire; instead, return all results.
{}

Returns:

Type Description
list[GuardrailResult]

list[GuardrailResult]: Results from all executed guardrails.

Raises:

Type Description
ConfigError

If the configuration bundle cannot be loaded or is invalid.

ContextValidationError

If the context does not meet required fields for the guardrails.

GuardrailTripwireTriggered

If a guardrail tripwire is triggered and suppress_tripwire is False.

Example
from guardrails import check_plain_text

results = await check_plain_text(
    "some text",
    bundle_path="my_guardrails.json",
    concurrency=4,
    suppress_tripwire=True,
)
Source code in src/guardrails/runtime.py
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
async def check_plain_text(
    text: str,
    bundle_path: ConfigSource = Path("guardrails.json"),
    registry: GuardrailRegistry | None = None,
    ctx: Any = None,
    **kwargs: Any,
) -> list[GuardrailResult]:
    """Validate plain text input against all configured guardrails for the 'text/plain' media type.

    This function loads a guardrail configuration bundle, instantiates all guardrails for the
    specified registry, and runs each guardrail against the provided text input. It is the
    recommended entry point for validating plain text with one or more guardrails in an async context.

    If no context object (`ctx`) is provided, a minimal default context will be constructed with
    attributes required by the guardrails' context schema (for example, an OpenAI client and any
    required fields with safe default values). For advanced use cases, you can supply your own
    context object.

    Keyword arguments are forwarded to `run_guardrails`, allowing you to control concurrency,
    provide a result handler, or suppress tripwire exceptions.

    Args:
        text (str): The plain text input to validate.
        bundle_path (ConfigSource, optional): Guardrail configuration bundle. This can be a file path,
            dict, JSON string, or `ConfigBundle` instance. Defaults to `"guardrails.json"`.
        registry (GuardrailRegistry, optional): Guardrail registry to use for instantiation.
            If not provided, the default registry is used.
        ctx (Any, optional): Application context object passed to each guardrail.
            If None (default), a minimal default context will be used.
        **kwargs: Additional keyword arguments forwarded to `run_guardrails`. Common options include:

            - concurrency (int): Maximum number of guardrails to run concurrently.
            - result_handler (Callable[[GuardrailResult], Awaitable[None]]): Async function invoked for each result.
            - suppress_tripwire (bool): If True, do not raise an exception on tripwire; instead, return all results.

    Returns:
        list[GuardrailResult]: Results from all executed guardrails.

    Raises:
        ConfigError: If the configuration bundle cannot be loaded or is invalid.
        ContextValidationError: If the context does not meet required fields for the guardrails.
        GuardrailTripwireTriggered: If a guardrail tripwire is triggered and `suppress_tripwire` is False.

    Example:
        ```python
        from guardrails import check_plain_text

        results = await check_plain_text(
            "some text",
            bundle_path="my_guardrails.json",
            concurrency=4,
            suppress_tripwire=True,
        )
        ```
    """  # noqa: E501
    if ctx is None:
        ctx = _get_default_ctx()
    bundle = load_config_bundle(bundle_path)
    guardrails: list[ConfiguredGuardrail[Any, str, Any]] = instantiate_guardrails(
        bundle, registry=registry
    )
    return await run_guardrails(ctx, text, "text/plain", guardrails, stage_name=bundle.stage_name, **kwargs)