Configure GuardrailsCustom Actions

Creating Custom Actions

View as Markdown

This section describes how to create custom actions in the actions.py file.

The @action Decorator

Use the @action decorator from nemoguardrails.actions to define custom actions:

1from nemoguardrails.actions import action
2
3@action()
4async def my_custom_action():
5    """A simple custom action."""
6    return "result"

Decorator Parameters

ParameterTypeDescriptionDefault
namestrCustom name for the actionFunction name
is_system_actionboolAlways run locally, bypassing the actions serverFalse
execute_asyncboolContinue event processing while the action runs (Colang 2.x only)False

Custom Action Name

Override the default action name:

1@action(name="validate_user_input")
2async def check_input(text: str):
3    """Validates user input."""
4    return len(text) > 0

Call from Colang:

$is_valid = execute validate_user_input(text=$user_message)

System Actions

When is_system_action=True, the action always runs locally, even when an actions_server_url is configured. This is important for actions that need access to special parameters like context, llm, config, and events, which are only injected for locally-run actions.

When no actions_server_url is configured, all actions run locally and receive special parameters regardless of the is_system_action setting. The flag only affects behavior when an actions server is in use.

1@action(is_system_action=True)
2async def check_policy_compliance(context: Optional[dict] = None):
3    """Check if message complies with policy."""
4    message = context.get("last_user_message", "")
5    # Validation logic
6    return True

Async Execution

When execute_async=True, the event processing loop does not wait for the action to complete before continuing. The action runs in the background, and the event processing loop retrieves the result later by polling. Use this setting for long-running operations when you do not need the result immediately.

This flag is only supported in the Colang 2.x runtime. In the Colang 1.0 runtime, it is stored in metadata but has no effect.

1from nemoguardrails.actions import action
2from nemoguardrails.http import HTTPClient, http_call
3
4
5@action(execute_async=True)
6async def call_external_api(
7    endpoint: str,
8    http_client: HTTPClient | None = None,
9):
10    """Call an external API without blocking event processing."""
11    response = await http_call(http_client, "GET", endpoint)
12    return response.json()

Rail Decisions

The @action decorator does not interpret an action’s return value as a safety decision. Ordinary custom actions can return strings, booleans, numbers, dictionaries, or other Python values for a Colang flow to consume explicitly.

When the action itself makes a rail decision, return a RailOutcome. It carries an explicit allow, block, or transform decision without relying on implicit boolean or numeric conventions.

If you previously used the removed output_mapping decorator parameter, follow the migration guide.

Function Parameters

Actions can accept parameters of the following types:

TypeExample
str"hello"
int42
float3.14
boolTrue
list["a", "b", "c"]
dict{"key": "value"}

Basic Parameters

1@action()
2async def greet_user(name: str, formal: bool = False):
3    """Generate a greeting."""
4    if formal:
5        return f"Good day, {name}."
6    return f"Hello, {name}!"

Call from Colang:

$greeting = execute greet_user(name="Alice", formal=True)

Optional Parameters with Defaults

1@action()
2async def search_documents(
3    query: str,
4    max_results: int = 10,
5    include_metadata: bool = False
6):
7    """Search documents with optional parameters."""
8    results = perform_search(query, limit=max_results)
9    if include_metadata:
10        return {"results": results, "count": len(results)}
11    return results

Return Values

Actions can return various types:

Manifest-backed rail actions are the exception. They must return a RailOutcome.

Simple Return

1@action()
2async def get_status():
3    return "active"

Dictionary Return

1@action()
2async def get_user_info(user_id: str):
3    return {
4        "id": user_id,
5        "name": "John Doe",
6        "role": "admin"
7    }

Boolean Return for Flow Logic

A boolean remains application data until a Colang flow interprets it explicitly:

1from typing import Optional
2
3from nemoguardrails.actions import action
4
5
6@action(is_system_action=True)
7async def is_safe_content(context: Optional[dict] = None):
8    content = context.get("bot_message", "")
9    return not contains_harmful_content(content)

Branch on the value in the flow:

$is_safe = execute is_safe_content
if not $is_safe
  bot refuse to respond
  stop

Error Handling

Handle errors gracefully within actions:

1from nemoguardrails.actions import action
2from nemoguardrails.http import HTTPClient, HTTPTimeoutError, http_call
3
4
5@action()
6async def fetch_data(
7    url: str,
8    http_client: HTTPClient | None = None,
9):
10    """Fetch data with error handling."""
11    try:
12        response = await http_call(http_client, "GET", url)
13        return response.json()
14    except HTTPTimeoutError as error:
15        raise RuntimeError("External data service timed out") from error

Example Actions

Input Validation Action

1from typing import Optional
2
3from nemoguardrails.actions import action
4from nemoguardrails.actions.rail_outcome import RailOutcome
5
6
7@action(is_system_action=True)
8async def check_input_length(context: Optional[dict] = None):
9    """Ensure user input is not too long."""
10    user_message = context.get("user_message", "")
11    max_length = 1000
12
13    if len(user_message) > max_length:
14        return RailOutcome.block(reason="The input exceeds the length limit.")
15
16    return RailOutcome.allow()

Add a Colang 1.0 flow to rails.co that consumes the decision:

define subflow check input length
  $result = execute check_input_length
  if $result.is_blocked
    bot refuse to respond
    stop

Enable the flow as an input rail in config.yml:

1rails:
2  input:
3    flows:
4      - check input length

Output Filtering Action

1import re
2from typing import Optional
3
4from nemoguardrails.actions import action
5from nemoguardrails.actions.rail_outcome import RailOutcome
6
7
8@action(is_system_action=True)
9async def filter_sensitive_data(context: Optional[dict] = None):
10    """Check for sensitive data in bot response."""
11    bot_response = context.get("bot_message", "")
12
13    sensitive_patterns = [
14        r"\b\d{3}-\d{2}-\d{4}\b",  # SSN pattern
15        r"\b\d{16}\b",  # Credit card pattern
16    ]
17
18    for pattern in sensitive_patterns:
19        if re.search(pattern, bot_response):
20            return RailOutcome.block(reason="The response contains sensitive data.")
21
22    return RailOutcome.allow()

Add a Colang 1.0 flow to rails.co that consumes the decision:

define subflow filter sensitive data
  $result = execute filter_sensitive_data
  if $result.is_blocked
    bot refuse to respond
    stop

Enable the flow as an output rail in config.yml:

1rails:
2  output:
3    flows:
4      - filter sensitive data

External API Action

1from nemoguardrails.actions import action
2from nemoguardrails.http import HTTPClient, http_call
3
4
5@action(execute_async=True)
6async def query_knowledge_base(
7    query: str,
8    top_k: int = 5,
9    http_client: HTTPClient | None = None,
10):
11    """Query an external knowledge base API."""
12    response = await http_call(
13        http_client,
14        "POST",
15        "https://api.example.com/search",
16        json={"query": query, "limit": top_k},
17    )
18    return response.json().get("results", [])