Custom Actions

Custom actions allow you to execute Python code within your guardrails flows. Actions can perform external API calls, process data, update context, or trigger custom business logic.

The @action Decorator

Use the @action decorator to mark a function as an action that can be called from Colang flows.

Basic Action

From nemoguardrails/actions/core.py:

from nemoguardrails.actions.actions import action, ActionResult
from typing import Optional

@action(is_system_action=True)
async def create_event(
    event: dict,
    context: Optional[dict] = None,
):
    """Creates an event for the bot based on the provided data.

    Args:
        event (dict): The input event data.
        context (Optional[dict]): The context for the action. Defaults to None.

    Returns:
        ActionResult: An action result containing the created event.
    """
    event_dict = {
        "_type": event["_type"],
        **{k: v for k, v in event.items() if k != "_type"}
    }

    # Support for referring variables as values
    for k, v in event_dict.items():
        if isinstance(v, str) and v[0] == "$":
            event_dict[k] = context.get(v[1:], None) if context else None

    return ActionResult(events=[event_dict])

Action Decorator Parameters

is_system_action

boolean

default:"False"

Flag indicating if the action is a system action (internal to NeMo Guardrails)

name

string

Custom name for the action. If not provided, uses the function name

execute_async

boolean

default:"False"

Whether the function should be executed in async mode

output_mapping

function

A function to interpret the action’s result. Accepts the return value and returns True if the output is not safe

ActionResult Class

Actions can return an ActionResult object to provide more control over the execution flow.

return_value

any

The value returned by the action

events

list[dict]

Events to be added to the event stream

context_updates

dict

Updates made to the context by this action

Example: Wolfram Alpha Integration

From nemoguardrails/actions/math.py:

import logging
import os
from typing import Optional
from urllib import parse
import aiohttp

from nemoguardrails.actions import action
from nemoguardrails.actions.actions import ActionResult
from nemoguardrails.utils import new_event_dict

log = logging.getLogger(__name__)

APP_ID = os.environ.get("WOLFRAM_ALPHA_APP_ID")
API_URL_BASE = f"https://api.wolframalpha.com/v2/result?appid={APP_ID}"

@action(name="wolfram alpha request")
async def wolfram_alpha_request(
    query: Optional[str] = None,
    context: Optional[dict] = None
):
    """Makes a request to the Wolfram Alpha API.

    Args:
        query (Optional[str]): The query for Wolfram Alpha. Defaults to None.
        context (Optional[dict]): The context for the execution of the action.

    Returns:
        ActionResult or str: The result of the Wolfram Alpha request.

    Raises:
        Exception: If no query is provided to Wolfram Alpha.
    """
    # If we don't have an explicit query, we take the last user message
    if query is None and context is not None:
        query = context.get("last_user_message") or "2+3"

    if query is None:
        raise Exception("No query was provided to Wolfram Alpha.")

    if APP_ID is None:
        return ActionResult(
            return_value=False,
            events=[
                new_event_dict(
                    "BotIntent",
                    intent="inform wolfram alpha app id not set"
                ),
                new_event_dict(
                    "StartUtteranceBotAction",
                    script="Wolfram Alpha app ID is not set."
                ),
                new_event_dict("BotIntent", intent="stop"),
            ],
        )

    url = API_URL_BASE + "&" + parse.urlencode({"i": query})
    log.info(f"Wolfram Alpha: executing request for: {query}")

    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            if resp.status != 200:
                log.info(f"Wolfram Alpha request failed : {query}")
                return ActionResult(
                    return_value=False,
                    events=[
                        new_event_dict(
                            "BotIntent",
                            intent="inform wolfram alpha not working"
                        ),
                        new_event_dict(
                            "StartUtteranceBotAction",
                            script="Apologies, but I cannot answer this question."
                        ),
                        new_event_dict("BotIntent", intent="stop"),
                    ],
                )

            result = await resp.text()
            log.info(f"Wolfram Alpha: the result was {result}.")
            return result

Creating Custom Actions

Create actions.py

Create an actions.py file in your configuration directory:

config/
├── config.yml
├── rails.co
└── actions.py

Define Your Action

from nemoguardrails.actions import action
from typing import Optional

@action(name="check_database")
async def check_database(
    user_id: str,
    context: Optional[dict] = None
):
    """Check user information in database."""
    # Your custom logic here
    user_data = await fetch_user_from_db(user_id)
    return user_data

Call from Colang

Reference the action in your .co files:

define flow verify user
  user provide user id
  $user_data = execute check_database(user_id=$user_id)
  
  if $user_data
    bot confirm user verified
  else
    bot inform user not found

Action Patterns

Simple Return Value
With Context Updates
With Events
External API Call

@action()
async def get_current_time(context: Optional[dict] = None):
    """Returns the current time."""
    from datetime import datetime
    return datetime.now().strftime("%H:%M:%S")

from nemoguardrails.actions.actions import ActionResult

@action()
async def update_user_preferences(
    preferences: dict,
    context: Optional[dict] = None
):
    """Updates user preferences in context."""
    return ActionResult(
        return_value=True,
        context_updates={"user_preferences": preferences}
    )

from nemoguardrails.actions.actions import ActionResult
from nemoguardrails.utils import new_event_dict

@action()
async def trigger_notification(
    message: str,
    context: Optional[dict] = None
):
    """Triggers a notification event."""
    return ActionResult(
        return_value=True,
        events=[
            new_event_dict(
                "NotificationEvent",
                message=message,
                timestamp=datetime.now().isoformat()
            )
        ]
    )

import aiohttp
from nemoguardrails.actions.actions import ActionResult

@action(name="fetch_weather")
async def fetch_weather(
    city: str,
    context: Optional[dict] = None
):
    """Fetches weather data from external API."""
    api_key = os.environ.get("WEATHER_API_KEY")
    url = f"https://api.weather.com/v1/city/{city}?key={api_key}"
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            if resp.status == 200:
                data = await resp.json()
                return data["temperature"]
            else:
                return ActionResult(
                    return_value=None,
                    context_updates={"weather_error": True}
                )

Accessing Context

The context parameter provides access to the conversation state:

@action()
async def personalized_greeting(context: Optional[dict] = None):
    """Generate a personalized greeting."""
    if context is None:
        return "Hello!"
    
    user_name = context.get("user_name", "there")
    last_visit = context.get("last_visit_date")
    
    if last_visit:
        return f"Welcome back, {user_name}! Last visit: {last_visit}"
    else:
        return f"Hello, {user_name}! Nice to meet you."

Error Handling

@action()
async def safe_api_call(endpoint: str, context: Optional[dict] = None):
    """Makes an API call with error handling."""
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(endpoint) as resp:
                if resp.status == 200:
                    return await resp.json()
                else:
                    log.warning(f"API call failed: {resp.status}")
                    return ActionResult(
                        return_value=None,
                        context_updates={"api_error": resp.status}
                    )
    except Exception as e:
        log.error(f"Exception during API call: {e}")
        return ActionResult(
            return_value=None,
            context_updates={"api_exception": str(e)}
        )

Best Practices

Use Async Functions

Always use async def for actions, even if they don’t make async calls

@action()
async def my_action():
    # Your code here
    pass

Include Type Hints

Use type hints for better code clarity and IDE support

@action()
async def typed_action(
    param1: str,
    param2: int,
    context: Optional[dict] = None
) -> str:
    return f"Result: {param1} - {param2}"

Document Your Actions

Include docstrings explaining parameters and return values

Handle None Context

Always check if context is None before accessing it

if context is None:
    context = {}

Registering Actions

Actions in actions.py are automatically registered when the configuration is loaded. No additional registration is needed.

# config/actions.py
from nemoguardrails.actions import action

@action()
async def custom_action_1():
    """First custom action."""
    pass

@action()
async def custom_action_2():
    """Second custom action."""
    pass

# Both actions are automatically available in your rails

Testing Actions

Direct Testing
Integration Testing

import asyncio
from config.actions import custom_action

async def test_action():
    result = await custom_action(
        param="test",
        context={"user_id": "123"}
    )
    print(f"Result: {result}")

asyncio.run(test_action())

from nemoguardrails import RailsConfig, LLMRails

config = RailsConfig.from_path("./config")
rails = LLMRails(config)

response = rails.generate(
    messages=[{"role": "user", "content": "Trigger action"}]
)
print(response)

Get Started

Core Concepts

Configuration

Guardrails Library

Built-in Guardrails

Usage

Deployment

Evaluation

The @action Decorator

Basic Action

Action Decorator Parameters

ActionResult Class

Example: Wolfram Alpha Integration

Creating Custom Actions

Action Patterns

Accessing Context

Error Handling

Best Practices

Registering Actions

Testing Actions

Next Steps

Rails Definition

Guardrails Library

Build docs developers (and LLMs) love

Get Started

Core Concepts

Configuration

Guardrails Library

Built-in Guardrails

Usage

Deployment

Evaluation

​The @action Decorator

​Basic Action

​Action Decorator Parameters

​ActionResult Class

​Example: Wolfram Alpha Integration

​Creating Custom Actions

​Action Patterns

​Accessing Context

​Error Handling

​Best Practices

​Registering Actions

​Testing Actions

​Next Steps

Rails Definition

Guardrails Library

Build docs developers (and LLMs) love

The @action Decorator

Basic Action

Action Decorator Parameters

ActionResult Class

Example: Wolfram Alpha Integration

Creating Custom Actions

Action Patterns

Accessing Context

Error Handling

Best Practices

Registering Actions

Testing Actions

Next Steps