nemoguardrails.server.api | NVIDIA NeMo Guardrails Library Developer Guide

Module Contents

Classes

Name	Description
`ChunkError`	A terminal error frame pushed into the stream by the guardrails runtime.
`ChunkErrorMetadata`	-
`GuardrailsApp`	Custom FastAPI subclass with additional attributes for Guardrails server.
`GuardrailsConfigurationError`	Exception raised for errors in the configuration.

Functions

Name	Description
`_add_cors_middleware`	-
`_build_main_model`	Build the main model for a request, on top of the configured one when there is one.
`_configured_main_model`	Return the model the config declares as “main”, if it declares one.
`_format_streaming_response`	Format streaming chunks from LLMRails.stream_async() as SSE events.
`_generate_cache_key`	Generates a cache key for the given config ids and model name.
`_get_rails`	Returns the rails instance for the given config id and model.
`_has_config_file`	Check if a directory (or its ‘config’ subdirectory) contains a config.yml/yaml.
`_inject_model`	Inject the request’s model into a RailsConfig, keeping the configured main model’s fields.
`_inline_reasoning_as_think_tags`	Move `reasoning_content` into the assistant message as a <think> prefix and clear the field.
`_map_rail_status`	Map internal RailStatus to API status string.
`_prepend_stream_chunk`	-
`_resolve_main_model_engine`	Resolve the engine for an injected main model. Priority order:
`_resolve_main_model_parameters`	Resolve the parameters for an injected main model, preferring MAIN_MODEL_BASE_URL.
`_update_models_in_config`	Update the main model in the RailsConfig.
`_validated_main_model`	Build a main model from its fields, reporting a rejected field as a configuration error.
`chat_completion`	Chat completion for the provided conversation.
`get_challenges`	Returns the list of available challenges for red teaming.
`get_rails_configs`	Returns the list of available rails configurations.
`guardrail_check`	Guardrail check request.
`health`	Return HTTP 200 while the server process is running and able to serve requests.
`lifespan`	Register any additional challenges, if available at startup.
`list_models`	Return the list of models available from the configured provider.
`process_chunk`	Processes a single chunk from the stream.
`register_challenges`	Register additional challenges
`register_datastore`	Registers a DataStore to be used by the server.
`register_logger`	Register an additional logger
`root_handler`	-
`root_redirect`	-
`set_default_config_id`	-
`start_auto_reload_monitoring`	Start a thread that monitors the config folder for changes.

Data

ALLOWED_ORIGINS

ENABLE_CORS

_EXCEPTION_HANDLERS

api_description

api_request_headers

llm_rails_events_history_cache

llm_rails_instances

API

class nemoguardrails.server.api.ChunkError()

Bases: BaseModel

A terminal error frame pushed into the stream by the guardrails runtime.

type is restricted to the internal markers so that model output which merely looks like an OpenAI error object is streamed on as ordinary content rather than ending the stream. Without output rails nothing else inspects a chunk before it reaches process_chunk, so this is the last gate.

error

ChunkErrorMetadata

class nemoguardrails.server.api.ChunkErrorMetadata()

Bases: BaseModel

code

Union[str, int, None] = None

message

str

param

Optional[str] = None

type

Literal['generation_error', 'downstream_error', 'guardrails_violation']

class nemoguardrails.server.api.GuardrailsApp(
    args = (),
    kwargs = {}
)

Bases: FastAPI

Custom FastAPI subclass with additional attributes for Guardrails server.

auto_reload

bool = False

default_config_id

Optional[str] = None

disable_chat_ui

bool

loop

Optional[AbstractEventLoop] = None

rails_config_path

str = ''

single_config_id

Optional[str] = None

single_config_mode

bool = False

stop_signal

bool = False

task

Optional[Future] = None

class nemoguardrails.server.api.GuardrailsConfigurationError()

Exception

Bases: Exception

Exception raised for errors in the configuration.

nemoguardrails.server.api._add_cors_middleware(
    application: fastapi.FastAPI,
    origins: typing.List[str]
) -> None

nemoguardrails.server.api._build_main_model(
    model_name: str,
    configured_model: typing.Optional[nemoguardrails.rails.llm.config.Model]
) -> nemoguardrails.rails.llm.config.Model

Build the main model for a request, on top of the configured one when there is one.

nemoguardrails.server.api._configured_main_model(
    config: nemoguardrails.RailsConfig
) -> typing.Optional[nemoguardrails.rails.llm.config.Model]

Return the model the config declares as “main”, if it declares one.

nemoguardrails.server.api._format_streaming_response(
    stream_iterator: typing.AsyncIterator[typing.Union[str, dict]],
    model_name: str
) -> typing.AsyncIterator[str]

async

Format streaming chunks from LLMRails.stream_async() as SSE events.

Args: stream_iterator: AsyncIterator from stream_async() that yields str or dict chunks model_name: The model name to include in the chunks

Yields: SSE-formatted strings (data: {…}

)

nemoguardrails.server.api._generate_cache_key(
    config_ids: typing.List[str],
    model_name: typing.Optional[str] = None
) -> str

Generates a cache key for the given config ids and model name.

nemoguardrails.server.api._get_rails(
    config_ids: typing.List[str],
    model_name: typing.Optional[str] = None
) -> nemoguardrails.LLMRails

async

Returns the rails instance for the given config id and model.

Parameters:

config_ids

List[str]

List of configuration IDs to load

model_name

Optional[str]Defaults to None

The model name from the request (overrides config’s main model)

nemoguardrails.server.api._has_config_file(
    path: str
) -> bool

Check if a directory (or its ‘config’ subdirectory) contains a config.yml/yaml.

nemoguardrails.server.api._inject_model(
    config: nemoguardrails.RailsConfig,
    model_name: str
) -> nemoguardrails.RailsConfig

Inject the request’s model into a RailsConfig, keeping the configured main model’s fields.

nemoguardrails.server.api._inline_reasoning_as_think_tags(
    res: nemoguardrails.rails.llm.options.GenerationResponse
) -> nemoguardrails.rails.llm.options.GenerationResponse

Move reasoning_content into the assistant message as a <think> prefix and clear the field.

nemoguardrails.server.api._map_rail_status(
    status: nemoguardrails.rails.llm.options.RailStatus
) -> str

Map internal RailStatus to API status string.

nemoguardrails.server.api._prepend_stream_chunk(
    first_chunk: typing.Union[str, dict],
    stream_iterator: typing.AsyncIterator[typing.Union[str, dict]]
) -> typing.AsyncIterator[typing.Union[str, dict]]

async

nemoguardrails.server.api._resolve_main_model_engine(
    configured_model: typing.Optional[nemoguardrails.rails.llm.config.Model]
) -> str

Resolve the engine for an injected main model. Priority order:

MAIN_MODEL_ENGINE environment variable (warns if mismatch with configured_model.engine)
configured_model.engine if configured_model is provided
Fallback to openai

nemoguardrails.server.api._resolve_main_model_parameters(
    configured_model: typing.Optional[nemoguardrails.rails.llm.config.Model]
) -> dict[str, typing.Any]

Resolve the parameters for an injected main model, preferring MAIN_MODEL_BASE_URL.

nemoguardrails.server.api._update_models_in_config(
    config: nemoguardrails.RailsConfig,
    main_model: nemoguardrails.rails.llm.config.Model
) -> nemoguardrails.RailsConfig

Update the main model in the RailsConfig.

If a model with type=“main” exists, it replaces it. Otherwise, adds it.

nemoguardrails.server.api._validated_main_model(
    model_fields: dict[str, typing.Any]
) -> nemoguardrails.rails.llm.config.Model

Build a main model from its fields, reporting a rejected field as a configuration error.

nemoguardrails.server.api.chat_completion(
    body: nemoguardrails.server.schemas.openai.GuardrailsChatCompletionRequest,
    request: fastapi.Request
)

async

Chat completion for the provided conversation.

TODO: add support for explicit state object.

nemoguardrails.server.api.get_challenges()

async

Returns the list of available challenges for red teaming.

nemoguardrails.server.api.get_rails_configs()

async

Returns the list of available rails configurations.

nemoguardrails.server.api.guardrail_check(
    body: nemoguardrails.server.schemas.openai.GuardrailCheckRequest,
    request: fastapi.Request
)

async

Guardrail check request.

Returns 422 when rail_types includes a type with no configured flows.

nemoguardrails.server.api.health()

async

Return HTTP 200 while the server process is running and able to serve requests.

nemoguardrails.server.api.lifespan(
    app: nemoguardrails.server.api.GuardrailsApp
)

async

nemoguardrails.server.api.list_models(
    request: fastapi.Request
)

async

Return the list of models available from the configured provider.

nemoguardrails.server.api.process_chunk(
    chunk: typing.Any
) -> typing.Union[typing.Any, nemoguardrails.server.api.ChunkError]

Processes a single chunk from the stream.

Parameters:

chunk

Any

A single chunk from the stream (can be str, dict, or other type).

model

The model name (not used in processing but kept for signature consistency).

Returns: Union[Any, ChunkError]

Union[Any, StreamingError]: StreamingError instance for errors or the original chunk.

nemoguardrails.server.api.register_challenges(
    additional_challenges: typing.List[dict]
)

Parameters:

additional_challenges

List[dict]

The new challenges to be registered.

nemoguardrails.server.api.register_datastore(
    datastore_instance: nemoguardrails.server.datastore.datastore.DataStore
)

Registers a DataStore to be used by the server.

nemoguardrails.server.api.register_logger(
    logger: typing.Callable
)

nemoguardrails.server.api.root_handler()

async

nemoguardrails.server.api.root_redirect()

async

nemoguardrails.server.api.set_default_config_id(
    config_id: str
)

nemoguardrails.server.api.start_auto_reload_monitoring()

Start a thread that monitors the config folder for changes.

nemoguardrails.server.api.ALLOWED_ORIGINS = os.getenv('NEMO_GUARDRAILS_SERVER_ALLOWED_ORIGINS', '*')

nemoguardrails.server.api.ENABLE_CORS = os.getenv('NEMO_GUARDRAILS_SERVER_ENABLE_CORS', 'false').lower() == 'true'

nemoguardrails.server.api._EXCEPTION_HANDLERS = ((LLMCallException, llm_call_exception_handler), (ModelEngineError, llm_call_exc...

nemoguardrails.server.api.api_description = 'Guardrails Server API.'

nemoguardrails.server.api.api_request_headers: ContextVar = contextvars.ContextVar('headers')

nemoguardrails.server.api.app = GuardrailsApp(title='Guardrails Server API', description=api_description, versio...

nemoguardrails.server.api.chainlit_app_path = os.path.join(os.path.dirname(__file__), 'app.py')

nemoguardrails.server.api.challenges = []

nemoguardrails.server.api.datastore: Optional[DataStore] = None

nemoguardrails.server.api.llm_rails_events_history_cache: dict[str, dict] = {}

nemoguardrails.server.api.llm_rails_instances: dict[str, LLMRails] = {}

nemoguardrails.server.api.log = logging.getLogger(__name__)

nemoguardrails.server.api.mount_chainlit: Optional[Callable[..., Any]] = _mount_chainlit

nemoguardrails.server.api.origins = ALLOWED_ORIGINS.split(',')

nemoguardrails.server.api.registered_loggers: List[Callable] = []