> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/NVIDIA-NeMo/Guardrails/llms.txt
> Use this file to discover all available pages before exploring further.

# 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`:

```python theme={null}
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

<ParamField path="is_system_action" type="boolean" default="False">
  Flag indicating if the action is a system action (internal to NeMo Guardrails)
</ParamField>

<ParamField path="name" type="string">
  Custom name for the action. If not provided, uses the function name
</ParamField>

<ParamField path="execute_async" type="boolean" default="False">
  Whether the function should be executed in async mode
</ParamField>

<ParamField path="output_mapping" type="function">
  A function to interpret the action's result. Accepts the return value and returns True if the output is not safe
</ParamField>

## ActionResult Class

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

<ParamField path="return_value" type="any">
  The value returned by the action
</ParamField>

<ParamField path="events" type="list[dict]">
  Events to be added to the event stream
</ParamField>

<ParamField path="context_updates" type="dict">
  Updates made to the context by this action
</ParamField>

## Example: Wolfram Alpha Integration

From `nemoguardrails/actions/math.py`:

```python theme={null}
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

<Steps>
  <Step title="Create actions.py">
    Create an `actions.py` file in your configuration directory:

    ```
    config/
    ├── config.yml
    ├── rails.co
    └── actions.py
    ```
  </Step>

  <Step title="Define Your Action">
    ```python theme={null}
    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
    ```
  </Step>

  <Step title="Call from Colang">
    Reference the action in your `.co` files:

    ```colang theme={null}
    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
    ```
  </Step>
</Steps>

## Action Patterns

<Tabs>
  <Tab title="Simple Return Value">
    ```python theme={null}
    @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")
    ```
  </Tab>

  <Tab title="With Context Updates">
    ```python theme={null}
    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}
        )
    ```
  </Tab>

  <Tab title="With Events">
    ```python theme={null}
    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()
                )
            ]
        )
    ```
  </Tab>

  <Tab title="External API Call">
    ```python theme={null}
    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}
                    )
    ```
  </Tab>
</Tabs>

## Accessing Context

The `context` parameter provides access to the conversation state:

```python theme={null}
@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

<CodeGroup>
  ```python Graceful Errors theme={null}
  @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)}
          )
  ```

  ```python With Logging theme={null}
  import logging

  log = logging.getLogger(__name__)

  @action()
  async def logged_action(param: str, context: Optional[dict] = None):
      """Action with comprehensive logging."""
      log.info(f"Action called with param: {param}")
      
      try:
          result = perform_operation(param)
          log.info(f"Action succeeded: {result}")
          return result
      except Exception as e:
          log.error(f"Action failed: {e}")
          raise
  ```
</CodeGroup>

## Best Practices

<Steps>
  <Step title="Use Async Functions">
    Always use `async def` for actions, even if they don't make async calls

    ```python theme={null}
    @action()
    async def my_action():
        # Your code here
        pass
    ```
  </Step>

  <Step title="Include Type Hints">
    Use type hints for better code clarity and IDE support

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

  <Step title="Document Your Actions">
    Include docstrings explaining parameters and return values
  </Step>

  <Step title="Handle None Context">
    Always check if context is None before accessing it

    ```python theme={null}
    if context is None:
        context = {}
    ```
  </Step>
</Steps>

## Registering Actions

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

```python theme={null}
# 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

<Tabs>
  <Tab title="Direct Testing">
    ```python theme={null}
    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())
    ```
  </Tab>

  <Tab title="Integration Testing">
    ```python theme={null}
    from nemoguardrails import RailsConfig, LLMRails

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

    response = rails.generate(
        messages=[{"role": "user", "content": "Trigger action"}]
    )
    print(response)
    ```
  </Tab>
</Tabs>

## Next Steps

<CardGroup cols={2}>
  <Card title="Rails Definition" icon="shield" href="/configuration/rails-definition">
    Learn how to call actions from Colang flows
  </Card>

  <Card title="Guardrails Library" icon="books" href="/guardrails/overview">
    Explore built-in actions and rails
  </Card>
</CardGroup>