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 modules. 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 policies. 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.
suppress_enforcement	bool	When False a trigger from this guardrail raises :exc:GuardrailEnforcementTriggered (block mode). Defaults to True (alert / non-blocking).

Source code in src/mendguardrails/runtime.py

@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
    policies. 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.
        suppress_enforcement (bool): When ``False`` a trigger from this guardrail
            raises :exc:`GuardrailEnforcementTriggered` (block mode). Defaults
            to ``True`` (alert / non-blocking).
    """

    definition: GuardrailSpec[TContext, TIn, TCfg]
    config: TCfg
    suppress_enforcement: bool = True

    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/mendguardrails/runtime.py

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.
suppress_enforcement	bool	When False a trigger from this guardrail raises :exc:GuardrailEnforcementTriggered (block mode). Defaults to True (alert / non-blocking).

Source code in src/mendguardrails/runtime.py

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.
        suppress_enforcement (bool): When ``False`` a trigger from this guardrail
            raises :exc:`GuardrailEnforcementTriggered` (block mode).  Defaults
            to ``True`` (alert / non-blocking).
    """

    name: str
    config: dict[str, Any] = Field(default_factory=dict)
    suppress_enforcement: bool = True

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.
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_enforcement (bool): If True (default), don't raise exceptions on enforcement

Source code in src/mendguardrails/runtime.py

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.
        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_enforcement (bool): If True (default), don't raise exceptions on enforcement
    """

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

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

Policy

Bases: BaseModel

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

This class groups together guardrail configurations for the three main policy 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
policy = Policy(
    pre_flight=load_config_bundle(PRE_FLIGHT),
    input=load_config_bundle(INPUT_BUNDLE),
    output=load_config_bundle(OUTPUT_BUNDLE),
)

# Just output stage
policy = Policy(
    output=load_config_bundle(OUTPUT_BUNDLE),
)

# Print active stages
for stage in policy.stages():
    print(stage.version)

Source code in src/mendguardrails/runtime.py

class Policy(BaseModel):
    """Three-stage collection of validated ConfigBundles for an LLM policy.

    This class groups together guardrail configurations for the three main
    policy 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
        policy = Policy(
            pre_flight=load_config_bundle(PRE_FLIGHT),
            input=load_config_bundle(INPUT_BUNDLE),
            output=load_config_bundle(OUTPUT_BUNDLE),
        )

        # Just output stage
        policy = Policy(
            output=load_config_bundle(OUTPUT_BUNDLE),
        )

        # Print active stages
        for stage in policy.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")

    @property
    def is_empty(self) -> bool:
        """Return True when all stages are None (guardrails explicitly disabled)."""
        return not any(getattr(self, stage) is not None for stage in self._STAGE_ORDER)

    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)

is_empty property

is_empty: bool

Return True when all stages are None (guardrails explicitly disabled).

stages

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

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

Source code in src/mendguardrails/runtime.py

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/mendguardrails/runtime.py

@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

default_policy

default_policy() -> Policy

Return the default guardrail policy (basic PII and content moderation).

Used when no config is provided to clients. Includes: - pre_flight: PII (with block) and Moderation (hate, violence categories) - input: none (add checks via custom policy if needed) - output: Moderation (default category set from empty config)

Returns:

Name	Type	Description
Policy	Policy	A validated policy with default guardrails.

Source code in src/mendguardrails/runtime.py

def default_policy() -> Policy:
    """Return the default guardrail policy (basic PII and content moderation).

    Used when no config is provided to clients. Includes:
    - pre_flight: PII (with block) and Moderation (hate, violence categories)
    - input: none (add checks via custom policy if needed)
    - output: Moderation (default category set from empty config)

    Returns:
        Policy: A validated policy with default guardrails.
    """
    pre_flight_bundle = ConfigBundle(
        guardrails=[
            GuardrailConfig(name="PII", config={"block": True, "entities": ["US_SSN", "PHONE_NUMBER", "EMAIL_ADDRESS"]}),
            GuardrailConfig(name="HarmfulContent", config={"categories": ["hate", "violence"]}),
        ],
    )
    output_bundle = ConfigBundle(
        guardrails=[GuardrailConfig(name="HarmfulContent", config={})],
    )
    return Policy(pre_flight=pre_flight_bundle, input=None, output=output_bundle)

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/mendguardrails/runtime.py

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_policy

load_policy(source: PolicySource) -> Policy

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

Supported sources

• Policy: Already validated policy.
• dict: Raw data, parsed as Policy.
• Path: JSON file on disk.
• JsonString: Raw JSON string.

Example usage

policy = load_policy(JsonString('{"pre_flight": {...}, ...}'))
policy = load_policy({"pre_flight": {...}, ...})
policy = load_policy(Path("./policy.json"))

Parameters:

Name	Type	Description	Default
source	PolicySource	Policy input.	required

Raises:

Type	Description
ConfigError	If loading fails.
ValidationError	If model validation fails.
FileNotFoundError	If file doesn't exist.

Returns:

Name	Type	Description
Policy	Policy	The loaded and validated policy configuration.

Source code in src/mendguardrails/runtime.py

def load_policy(source: PolicySource) -> Policy:
    """Load a Policy from a path, dict, JSON string, or already-parsed object.

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

    Example usage:
        ```python
        policy = load_policy(JsonString('{"pre_flight": {...}, ...}'))
        policy = load_policy({"pre_flight": {...}, ...})
        policy = load_policy(Path("./policy.json"))
        ```

    Args:
        source (PolicySource): Policy input.

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

    Returns:
        Policy: The loaded and validated policy configuration.
    """
    return _load_bundle(source, Policy)

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/mendguardrails/runtime.py

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, suppress_enforcement=item.suppress_enforcement))
    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_enforcement: bool | None = None,
    stage_name: str | None = None,
    raise_guardrail_errors: bool = False,
    guardrail_id: str | None = None,
    mend_key: str | None = None,
    inference_model: str | None = None,
    offline: bool = False,
    audit_queue: Any | None = None,
) -> 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 enforcement, the function will raise a GuardrailEnforcementTriggered exception unless enforcement 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 modules. 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_enforcement	bool | None	Controls enforcement behaviour. None (default) — defer to each guardrail's own suppress_enforcement flag (per-guardrail policy). True — suppress all enforcement regardless of per-guardrail flags. False — hard-block on the first triggered guardrail regardless of per-guardrail flags.	None
stage_name	str | None	Name of the policy 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
guardrail_id	str | None	guardrailId / clientConnectionId for audit; forwarded to the audit handler for trace correlation.	None
mend_key	str | None	Resolved Mend license key used to derive the audit base URL when GUARDRAILS_AUDIT_URL is not set. Falls back to :envvar:MEND_KEY when None.	None
inference_model	str | None	Name of the AI inference model used in the surrounding API call (e.g. "gpt-4o"). Included as model in every audit event so the server distinguishes the LLM from the guardrail check model.	None
offline	bool	When True audit events are suppressed regardless of whether an audit URL is configured. Defaults to False.	False
audit_queue	EventQueue | None	Pre-started :class:~guardrails.utils.events.EventQueue for non-blocking event delivery. When provided, audit payloads are enqueued instead of awaited inline, keeping audit I/O off the guardrail execution hot-path. When None (default), inline delivery is used.	None

Returns:

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

Raises:

Type	Description
GuardrailEnforcementTriggered	Raised if a guardrail enforcement is triggered and suppress_enforcement 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_enforcement=True,
    stage_name="input",
)

Source code in src/mendguardrails/runtime.py

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_enforcement: bool | None = None,
    stage_name: str | None = None,
    raise_guardrail_errors: bool = False,
    guardrail_id: str | None = None,
    mend_key: str | None = None,
    inference_model: str | None = None,
    offline: bool = False,
    audit_queue: Any | None = None,
) -> 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 enforcement,
    the function will raise a `GuardrailEnforcementTriggered` exception unless enforcement 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 modules. 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_enforcement (bool | None, optional): Controls enforcement behaviour.
            ``None`` (default) — defer to each guardrail's own
            ``suppress_enforcement`` flag (per-guardrail policy).
            ``True`` — suppress all enforcement regardless of per-guardrail flags.
            ``False`` — hard-block on the first triggered guardrail regardless of
            per-guardrail flags.
        stage_name (str | None, optional): Name of the policy 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.
        guardrail_id (str | None, optional): ``guardrailId`` / ``clientConnectionId`` for audit;
            forwarded to the audit handler for trace correlation.
        mend_key (str | None, optional): Resolved Mend license key used to derive the audit
            base URL when ``GUARDRAILS_AUDIT_URL`` is not set. Falls back to :envvar:`MEND_KEY`
            when ``None``.
        inference_model (str | None, optional): Name of the AI inference model used in the
            surrounding API call (e.g. ``"gpt-4o"``). Included as ``model`` in every audit
            event so the server distinguishes the LLM from the guardrail check model.
        offline (bool): When ``True`` audit events are suppressed regardless of whether
            an audit URL is configured. Defaults to ``False``.
        audit_queue (EventQueue | None, optional): Pre-started
            :class:`~guardrails.utils.events.EventQueue` for non-blocking event delivery.
            When provided, audit payloads are enqueued instead of awaited inline,
            keeping audit I/O off the guardrail execution hot-path.  When ``None``
            (default), inline delivery is used.

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

    Raises:
        GuardrailEnforcementTriggered: Raised if a guardrail enforcement is triggered and
            `suppress_enforcement` 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_enforcement=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)

    # Get audit handler — skipped entirely in offline mode.
    # URL is resolved from mend_key (or MEND_KEY env) when
    # GUARDRAILS_AUDIT_URL is not explicitly set.
    # Audit handler: when suppress is None (per-guardrail mode) pass True so
    # the shared metadata defaults to "alert"; per-result info overrides per event.
    audit_suppress = True if suppress_enforcement is None else suppress_enforcement
    audit_handler = (
        None
        if offline
        else _get_audit_handler(
            guardrail_id,
            mend_key=mend_key,
            inference_model=inference_model,
            suppress_enforcement=audit_suppress,
            queue=audit_queue,
        )
    )

    semaphore = asyncio.Semaphore(concurrency)
    results: list[GuardrailResult] = []
    enforcement_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)

                # Carry stage_name and per-guardrail suppress flag into every result
                # so downstream audit / enforcement logic can act on each guardrail
                # independently without relying on the bundle-level flag.
                result = GuardrailResult(
                    enforcement_triggered=result.enforcement_triggered,
                    execution_failed=result.execution_failed,
                    original_exception=result.original_exception,
                    info={
                        **result.info,
                        "stage_name": stage_name or "unnamed",
                        "suppress_enforcement": g.suppress_enforcement,
                    },
                )

            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(
                        enforcement_triggered=False,  # Don't trigger enforcement 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 and audit 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 audit_handler:
                try:
                    await audit_handler(result)
                except Exception as audit_err:
                    logger.error(
                        "Error in audit handler for guardrail '%s': %s",
                        g.definition.name,
                        audit_err,
                    )
            if result.enforcement_triggered:
                logger.warning("Enforcement 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")
            if failure.original_exception:
                raise failure.original_exception

    # Determine which results constitute a hard block:
    # - suppress_enforcement=False  → block ALL triggered results (explicit override)
    # - suppress_enforcement=True   → block NONE (explicit suppress-all override)
    # - suppress_enforcement=None   → per-guardrail: block only when the result's
    #   own suppress_enforcement flag (injected from ConfiguredGuardrail) is False.
    if suppress_enforcement is False:
        hard_block_results = [r for r in results if r.enforcement_triggered]
    elif suppress_enforcement is True:
        hard_block_results = []
    else:  # None → per-guardrail
        hard_block_results = [
            r for r in results
            if r.enforcement_triggered
            and not r.info.get("suppress_enforcement", True)
        ]
    if hard_block_results:
        raise GuardrailEnforcementTriggered(hard_block_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("mendguardrails.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 enforcement 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 "mendguardrails.json".	Path('mendguardrails.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_enforcement (bool): If True (default), do not raise an exception on enforcement; 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.
GuardrailEnforcementTriggered	If a guardrail enforcement is triggered and suppress_enforcement is False.

Example

from mendguardrails import check_plain_text

results = await check_plain_text(
    "some text",
    bundle_path="my_guardrails.json",
    concurrency=4,
    suppress_enforcement=True,
)

Source code in src/mendguardrails/runtime.py

async def check_plain_text(
    text: str,
    bundle_path: ConfigSource = Path("mendguardrails.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 enforcement 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 `"mendguardrails.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_enforcement (bool): If True (default), do not raise an exception on enforcement; 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.
        GuardrailEnforcementTriggered: If a guardrail enforcement is triggered and `suppress_enforcement` is False.

    Example:
        ```python
        from mendguardrails import check_plain_text

        results = await check_plain_text(
            "some text",
            bundle_path="my_guardrails.json",
            concurrency=4,
            suppress_enforcement=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, **kwargs)