Custom Actions

Create custom Python actions for your guardrails

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

1. Create actions.py

   Create an actions.py file in your configuration directory:

   config/
   ├── config.yml
   ├── rails.co
   └── actions.py
   
2. 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
   
3. 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

@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")

With Context Updates

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}
    )

With Events

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()
            )
        ]
    )

External API Call

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

Graceful Errors

@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

1. 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
   
2. 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}"
   
3. Document Your Actions

   Include docstrings explaining parameters and return values

4. 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

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())

Integration Testing

from nemoguardrails import RailsConfig, LLMRails

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

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

Next Steps

Rails Definition

Learn how to call actions from Colang flows

Guardrails Library

Explore built-in actions and rails