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

Module Contents

Classes

Name	Description
`ChunkError`	-
`ChunkErrorMetadata`	-
`GuardrailsApp`	Custom FastAPI subclass with additional attributes for Guardrails server.
`GuardrailsConfigurationError`	Exception raised for errors in the configuration.

Functions

Name	Description
`_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.
`_raise_invalid_state`	-
`_update_models_in_config`	Update the main model in the RailsConfig.
`_validate_public_state_shape`	Validate request state shape before loading rails config.
`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.
`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

api_description

api_request_headers

llm_rails_events_history_cache

llm_rails_instances

log

origins

registered_loggers

API

class nemoguardrails.server.api.ChunkError()

Bases: BaseModel

error

ChunkErrorMetadata

class nemoguardrails.server.api.ChunkErrorMetadata()

Bases: BaseModel

code

Optional[str] = None

message

str

param

Optional[str] = None

type

Optional[str] = None

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._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._raise_invalid_state(
    detail: str
) -> None

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._validate_public_state_shape(
    state: typing.Optional[dict]
) -> None

Validate request state shape before loading rails config.

At the public HTTP boundary, the only accepted non-empty dict state shape is Colang 1.0 transcript state: {“events”: […]}. Colang 2.0 has no safe public dict state shape.

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.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.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.origins = ALLOWED_ORIGINS.split(',')

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