Skip to content

Dynamic Bots - Usage Documentation

Note

Available for RDA versions 8.2 and Above.

Overview

The Mako Code Executor is a dynamic bot execution framework that allows user's to create, store, and execute custom data processing logic using Mako templates. Bot definitions are stored in a persistent stream (pstream) and can be invoked dynamically with parameters.

Table of Contents

  • Basic Usage
  • Bot Configuration
  • Allowed Imports
  • Secure Credential Access with Wrapper Modules
  • Variables Available in Mako Templates
  • Setting Output
  • Using as a Data Source
  • Using as a Sink (with Input DataFrame)
  • Parameter Validation with JSON Schema
  • Credentials Management
  • Complete Example

1. Basic Usage

1.1 As a Source (get_data)

@dm:dynamic-bot bot_name="all-pstream-stats" secret_name is "api-cred-131-35" host is "192.168.131.35"

1.2 As a Sink (update_data) - Processing Input DataFrame

@dm:recall name is "input_df"
    --> @dm:dynamic-bot bot_name="data_transformer"

2. Bot Configuration

Bot definitions must be stored in a pstream (default: cfx_bots_registry) with the following structure:

2.1 Required Fields

Field Type Description
bot_name string Unique identifier for the bot
mako_template string The Mako template code to execute
version string Version identifier (e.g., "1.0.0")
default_version boolean Set to true for the default version

2.2 Optional Fields

Field Type Description
json_schema object/string JSON Schema for parameter validation
imports array List of Python modules to import
allowed_credential_types array List of credential types the bot can access

3. Allowed Imports

The imports field in your bot configuration controls which Python modules are available in your Mako template. This is a security feature that restricts what code can be executed.

3.1 Import Restrictions

Important

Only the following modules are allowed by the MakoCodeExecutor framework:

Module Description Usage
pandas Data manipulation and analysis Data processing, DataFrames
numpy Numerical computing Array operations, mathematical functions
re Regular expressions Pattern matching, text processing
json JSON encoding/decoding Parse/generate JSON data
datetime Date and time handling Timestamps, date operations
rda_requests Secure HTTP client wrapper API calls with credential management
rda_paramiko Secure SSH client wrapper SSH connections with credential management
rda_logging RDAF logging framework Structured logging
rda_pathlib Restricted filesystem path operations Sandboxed file access with RestrictedPath

Any other imports will be rejected during bot verification.

Note

More modules may be allowed in future after review.

3.2 Import Configuration Format

The imports field is an array of import specifications:

"imports": [
    {"module": "module_name"},
    {"module": "module_name", "as": "alias"},
    {"module": "package.submodule"},
    {"module": "package.submodule", "as": "alias"}
]

Remember

Only modules from the allowed list above can be imported. For example:

✅ Valid: {"module": "pandas", "as": "pd"}
✅ Valid: {"module": "rda_requests", "as": "requests"}
✅ Valid: {"module": "rda_pathlib"} with {"items": {"RestrictedPath": "Path"}}
❌ Invalid: {"module": "requests"} (not in allowed list, use rda_requests)
❌ Invalid: {"module": "pathlib"} (not in allowed list, use rda_pathlib)
❌ Invalid: {"module": "os"} (not in allowed list)

3.3 Examples

3.3.1 Basic Import

"imports": [
    {"module": "json"},
    {"module": "rda_logging.rda_logger", "as": "logging"}
]

In your template, these are available directly:

<%
logger = logging.getLogger("my_bot")
data = json.loads('{"key": "value"}')
%>

3.3.2 Import with Alias

"imports": [
    {"module": "pandas", "as": "pd"},
    {"module": "numpy", "as": "np"}
]

In your template:

<%
df = pd.DataFrame([{"col1": 1, "col2": 2}])
arr = np.array([1, 2, 3, 4, 5])
%>

"imports": [
    {"module": "rda_requests", "as": "requests"},
    {"module": "rda_paramiko", "as": "paramiko"},
    {"module": "rda_pathlib", "items": {"RestrictedPath": "Path"}}
]

In your template:

<%
# Secure HTTP requests
session = requests.Session()
response = session.get(url, rda_secret_name=secret_name)

# Secure SSH connections
ssh = paramiko.SSHClient()
ssh.connect(host, rda_secret_name=secret_name)

# Secure file operations (sandboxed to /var/dynamic_bots)
file_path = Path("data/output.txt")
file_path.write_text("Hello, World!")
%>

3.4 Using rda_pathlib for Safe File Operations

The rda_pathlib module provides RestrictedPath - a sandboxed filesystem interface that prevents directory traversal attacks and restricts file operations to a safe sandbox.

3.4.1 Key Features

  • Sandboxed Access: All file operations are restricted to /var/dynamic_bots/
  • Auto-Creation: Sandbox directory is automatically created if it doesn't exist
  • Path Traversal Prevention: Attempts to access parent directories or use .. are blocked
  • Symlink Protection: Symlinks are not followed by default
  • Similar API to pathlib: Familiar Path-like interface
  • Explicit Blocking: Dangerous operations like rename(), chmod(), and symlinks are explicitly blocked

3.4.2 Configuration

{
  "imports": [
    {
      "module": "rda_pathlib",
      "items": {
        "RestrictedPath": "Path"
      }
    }
  ]
}

3.4.3 Usage Examples

<%
# Create a path within the sandbox
file_path = Path("output/data.json")

# Write data
data = {"result": "success", "count": 100}
file_path.write_text(json.dumps(data))

# Read data back
content = file_path.read_text()
loaded_data = json.loads(content)

output["value"] = [loaded_data]
%>

<%
config_file = Path("config/settings.json")

if config_file.exists():
    settings = json.loads(config_file.read_text())
else:
    settings = {"default": True}

output["value"] = [settings]
%>

<%
# Build paths safely
base = Path("reports")
date_path = base / "2025" / "01" / "report.csv"

# Get path info
logger.info(f"Path name: {date_path.name}")
logger.info(f"Path parent: {date_path.parent}")
logger.info(f"Path suffix: {date_path.suffix}")

# Check file/directory type
if date_path.exists():
    if date_path.is_file():
        logger.info("It's a file")
    elif date_path.is_dir():
        logger.info("It's a directory")

output["value"] = [{"path": str(date_path)}]
%>

<%
# Create directory structure
reports_dir = Path("reports/2025/01")
reports_dir.mkdir(parents=True, exist_ok=True)

# List directory contents
data_dir = Path("data")
if data_dir.exists() and data_dir.is_dir():
    files = []
    for item in data_dir.iterdir():
        files.append({
            "name": item.name,
            "is_file": item.is_file(),
            "is_dir": item.is_dir()
        })
    output["value"] = files
else:
    output["value"] = []
%>

<%
# Get file stats
file_path = Path("data/input.json")
if file_path.exists():
    stats = file_path.stat()
    info = {
        "size": stats.st_size,
        "modified": stats.st_mtime,
        "is_file": file_path.is_file()
    }

    # Delete file if needed
    # file_path.unlink()  # Removes the file

    output["value"] = [info]
else:
    output["value"] = [{"error": "File not found"}]
%>

3.4.4 Security Restrictions

Allowed Operations Blocked Operations
• Read/write files within /var/dynamic_bots/
• Create directories with mkdir()
• List directory contents with iterdir()
• Check status with exists(), is_file(), is_dir()
• Get file statistics with stat()
• Delete files with unlink()
• Path manipulation (join, name, parent, etc.)
• Read/write text and bytes		 • Access to files outside /var/dynamic_bots/ sandbox
• Path traversal with .. or absolute paths
• Following symlinks (disabled by default)
• rename() - explicitly blocked
• replace() - explicitly blocked
• chmod() - explicitly blocked
• symlink_to() - explicitly blocked
• hardlink_to() - explicitly blocked
• Direct use of os or shutil modules

3.4.5 Security Error Examples

<%
# Path traversal - raises PermissionError
dangerous_path = Path("../../etc/passwd")
dangerous_path.read_text()  # ❌ Error: Access outside sandbox not allowed
%>

Error message: 

PermissionError: Access outside sandbox (/var/dynamic_bots) is not allowed: /etc/passwd


<%
# Blocked operation - raises PermissionError
file_path = Path("data/file.txt")
file_path.rename("data/newname.txt")  # ❌ Error: rename() is not allowed
%>

Error message: 

PermissionError: rename() is not allowed in RestrictedPath

3.5 Commonly Used Module Combinations

Here are commonly used module combinations for different bot purposes:

3.5.1 Data Processing Bots

"imports": [
    {"module": "pandas", "as": "pd"},
    {"module": "numpy", "as": "np"},
    {"module": "json"},
    {"module": "datetime"}
]

3.5.2 API Integration Bots

"imports": [
    {"module": "rda_requests", "as": "requests"},
    {"module": "json"},
    {"module": "rda_logging.rda_logger", "as": "logging"}
]

Note

Use rda_requests instead of plain requests for secure credential handling via rda_secret_name.

3.5.3 SSH Automation Bots

"imports": [
    {"module": "rda_paramiko", "as": "paramiko"},
    {"module": "json"},
    {"module": "rda_logging.rda_logger", "as": "logging"}
]

3.5.4 Data Transformation Bots

"imports": [
    {"module": "pandas", "as": "pd"},
    {"module": "json"},
    {"module": "re"},
    {"module": "datetime"}
]

3.6 Security Enforcement

The MakoCodeExecutor framework enforces strict import restrictions:

Allowed Modules (Whitelist) Blocked Modules
Only these 9 modules can be imported:
• pandas, numpy, re, json, datetime
• rda_requests, rda_paramiko, rda_logging, rda_pathlib		 All other modules are blocked, including:
• Process Execution: subprocess, os.system, commands
• Direct HTTP: requests, httpx, urllib3 (use rda_requests)
• Direct SSH: paramiko (use rda_paramiko)
• File System: os, shutil, glob, pathlib (use rda_pathlib)
• Code Execution: eval, exec, compile, importlib
• Network: socket, ftplib, telnetlib, urllib
• System Access: sys, platform, ctypes
• Cloud SDKs: boto3, google-cloud-*, azure-*
• Databases: pymongo, psycopg2, mysql-*

3.5.5 Verification Error Example

If you try to import a non-allowed module:

"imports": [
    {"module": "os"}  // ❌ Not allowed
]

You'll receive an error during bot verification:

Exception: Module import is not allowed for these: os, only allowed are pandas, numpy, re, json, datetime, rda_requests, rda_paramiko, rda_logging, rda_pathlib

3.5.6 Best Practices

  1. Only import what you need: Don't import modules you don't use
  2. Use secure wrappers: Always use rda_requests and rda_paramiko instead of direct alternatives
  3. Validate your imports: Test bot verification before deployment
  4. Keep it simple: The restricted import list is designed for safe data processing
  5. Check the whitelist: Always verify your module is in the allowed list before using it

3.7 Import Error Handling

If a module import fails during template verification, you'll receive an error:

Error: Module Not in Whitelist 

Exception: Module import is not allowed for these: requests, pathlib, only allowed are pandas, numpy, re, json, datetime, rda_requests, rda_paramiko, rda_logging, rda_pathlib

Solutions:

  1. ✅ Use the allowed alternative: Replace requests with rda_requests, pathlib with rda_pathlib
  2. ✅ Check the module name spelling
  3. ✅ Verify the module is in the whitelist (see table above)
  4. ❌ You cannot add new modules to the whitelist - this is a security restriction

3.8 Complete Import Examples

{
  "bot_name": "api_collector",
  "imports": [
    { "module": "rda_requests", "as": "requests" },
    { "module": "json" },
    { "module": "rda_logging.rda_logger", "as": "logging" },
    { "module": "datetime" }
  ]
}

Template usage:

<%
logger = logging.getLogger("api_collector")

# Create session - uses rda_requests wrapper
session = requests.Session()

# Make API request with secure credential loading
response = session.get(
    f"https://{host}/api/data",
    rda_secret_name=secret_name,
    timeout=30,
    verify=False
)

# Parse response
data = response.json()

# Add timestamp
data['collected_at'] = datetime.datetime.now().isoformat()

# Return results
output["value"] = data
%>

{
  "bot_name": "data_normalizer",
  "imports": [
    { "module": "pandas", "as": "pd" },
    { "module": "numpy", "as": "np" },
    { "module": "re" }
  ]
}

Template usage:

<%
# Use pandas for data manipulation
df_clean = df.copy()

# Remove special characters using regex
df_clean['name'] = df_clean['name'].apply(lambda x: re.sub(r'[^a-zA-Z0-9\s]', '', x))

# Normalize numeric columns using numpy
numeric_cols = df_clean.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
    df_clean[col] = (df_clean[col] - df_clean[col].mean()) / df_clean[col].std()

# Return normalized data
output["value"] = df_clean
%>

{
  "bot_name": "ssh_command_executor",
  "imports": [
    { "module": "rda_paramiko", "as": "paramiko" },
    { "module": "json" },
    { "module": "rda_logging.rda_logger", "as": "logging" }
  ]
}

Template usage:

<%
logger = logging.getLogger("ssh_executor")

# SSH connection with secure credential loading
ssh = paramiko.SSHClient()
ssh.connect(host, rda_secret_name=secret_name)

# Execute command
stdin, stdout, stderr = ssh.exec_command(command)
output_data = stdout.read().decode()
error_data = stderr.read().decode()

ssh.close()

# Return results
result = {
    "output": output_data,
    "error": error_data,
    "command": command
}

output["value"] = [result]
%>

{
  "bot_name": "file_processor",
  "imports": [
    { "module": "rda_pathlib", "items": {"RestrictedPath": "Path"} },
    { "module": "pandas", "as": "pd" },
    { "module": "json" }
  ]
}

Template usage:

<%
# Process input data and save to file
input_file = Path("input") / f"{file_name}.json"
output_file = Path("output") / f"{file_name}_processed.json"

# Read input if exists
if input_file.exists():
    data = json.loads(input_file.read_text())
else:
    data = {"status": "new_file"}

# Process data
data["processed"] = True
data["timestamp"] = datetime.datetime.now().isoformat()

# Write output (creates directory if needed)
output_file.write_text(json.dumps(data, indent=2))

# Return result
output["value"] = [{
    "input_file": str(input_file),
    "output_file": str(output_file),
    "status": "success"
}]
%>

4. Secure Credential Access with Wrapper Modules

4.1 Why Use Wrapper Modules?

The new architecture prevents templates from: - Accessing raw credentials directly - Performing unrestricted file system operations - Escaping sandbox restrictions

Instead, secure wrapper modules (rda_requests, rda_paramiko, and rda_pathlib) provide controlled, sandboxed access.

4.2 Key Benefits

  1. Enhanced Security: Templates never see raw credential data or access files outside sandbox
  2. Simplified Code: No manual credential parsing or path validation
  3. Consistent Pattern: Use rda_secret_name for credentials, RestrictedPath for files
  4. Vault Integration: Automatic credential loading from the vault
  5. Sandbox Protection: File operations restricted to safe directories

4.3 Quick Start

For HTTP/API calls, use rda_requests:

"imports": [
  { "module": "rda_requests", "as": "requests" }
]

<%
session = requests.Session()
response = session.get(url, rda_secret_name=secret_name, verify=False)
%>

For SSH connections, use rda_paramiko:

"imports": [
  { "module": "rda_paramiko", "as": "paramiko" }
]

<%
client = paramiko.SSHClient()
client.connect(host, rda_secret_name=secret_name)
%>

For file operations, use rda_pathlib:

"imports": [
  { "module": "rda_pathlib", "items": {"RestrictedPath": "Path"} }
]

<%
file_path = Path("data/output.json")
file_path.write_text(json.dumps(data))
%>

See the Credentials Management section for detailed usage.

4. Variables Available in Mako Templates

When your Mako template executes, the following variables are available in the execution context:

4.1 Template Parameters (from JSON Schema)

Any parameters defined in your JSON schema and passed during invocation are available as variables:

<%
# If you pass host="192.168.131.35" and secret_name="api-cred-131-35"
# These are directly accessible as variables:
url = f"https://{host}/api/v2"  # host is available
# secret_name is available to pass to rda_requests/rda_paramiko

# To use credentials securely:
session = requests.Session()  # requires rda_requests import
response = session.get(url, rda_secret_name=secret_name)
%>

4.2 df - Input DataFrame (Sink Mode Only)

When used as a sink (with update_data), the input DataFrame is available as df:

<%
# Access the input dataframe
row_count = len(df)
column_names = df.columns.tolist()

# Transform the dataframe
df_transformed = df.copy()
df_transformed['new_column'] = df_transformed['existing_column'] * 2

# Set as output
output["value"] = df_transformed
%>

4.3 output Dictionary

A special dictionary for returning results (see next section).

4.4 Imported Modules

Any modules specified in the imports configuration are pre-imported:

<%
# If imports includes {"module": "pandas", "as": "pd"}
df_result = pd.DataFrame([{'col1': 1, 'col2': 2}])

# If imports includes {"module": "rda_requests", "as": "requests"}
response = requests.get("https://api.example.com/data")

# If imports includes {"module": "rda_logging.rda_logger", "as": "logging"}
logger = logging.getLogger("my_bot")
logger.info("Processing started")
%>

4.5 Built-in Python Functions

Standard Python built-ins are available: len(), range(), enumerate(), zip(), etc.

5. Setting Output

The correct way to return data is:

<%
# Create your result (DataFrame, list, dict, etc.)
result = [
    {"pstream_name": "stream1", "status": "Success", "count": 100},
    {"pstream_name": "stream2", "status": "Success", "count": 250}
]

# Set the output - THIS IS THE CORRECT WAY
output["value"] = result
%>

5.1 Supported Output Types

The framework automatically converts various Python types to DataFrames:

  1. List of dictionaries (most common):
output["value"] = [{"col1": "a", "col2": 1}, {"col1": "b", "col2": 2}]
  1. Pandas DataFrame:
output["value"] = pd.DataFrame(data)
  1. Single dictionary:
output["value"] = {"col1": "value", "col2": 123}
# Becomes a single-row DataFrame
  1. JSON string: 
    output["value"] = json.dumps([{"key": "value"}])
# Will be parsed and converted to DataFrame

6. Using as a Data Source

When used as a data source, the bot generates data without any input:

@dm:dynamic-bot
    bot_name="all-pstream-stats"
    secret_name is "api-cred-131-35"
    host is "192.168.131.35"

The template receives:

  • host = "192.168.131.35"
  • secret_name = "api-cred-131-35"

Note

The rda_requests wrapper will load the credential securely when rda_secret_name=secret_name is used.

Returns: A DataFrame with execution results

7. Using as a Sink (with Input DataFrame)

When used as a sink, the bot receives a DataFrame and processes it.

@dm:recall name is "input_df"
    --> @dm:dynamic-bot
        bot_name="data_transformer"

The template receives:

  • df = The input DataFrame from @dm:recall

Mako Template Example:

<%
# Access the input DataFrame
logger.info(f"Received {len(df)} rows with columns: {df.columns.tolist()}")

# Transform the data
df_normalized = df.copy()
for col in df_normalized.select_dtypes(include=['float64', 'int64']).columns:
    df_normalized[col] = (df_normalized[col] - df_normalized[col].min()) / (df_normalized[col].max() - df_normalized[col].min())

# Add metadata
df_normalized['transformation'] = operation
df_normalized['processed_at'] = datetime.now().isoformat()

# Return the transformed data
output["value"] = df_normalized
%>

8. Parameter Validation with JSON Schema

The json_schema field defines what parameters your bot accepts and validates them before execution.

8.1 Schema Structure

{
  "type": "object",
  "properties": {
    "param_name": {
      "type": "string|integer|number|boolean|array|object",
      "description": "Human-readable description"
    }
  },
  "required": ["param1", "param2"]
}

{
  "type": "object",
  "properties": {
    "host": {
      "type": "string",
      "description": "Hostname of the RDAF setup"
    },
    "cred_name": {
      "type": "string",
      "description": "Credential name to use for collection"
    },
    "port": {
      "type": "integer",
      "description": "Port number"
    },
    "timeout": {
      "type": "number",
      "description": "Request timeout in seconds"
    },
    "verify_ssl": {
      "type": "boolean",
      "description": "Whether to verify SSL certificates"
    }
  },
  "required": ["host", "cred_name"]
}

8.2 Validation Behavior

  • ✅ Required parameters must be provided
  • ✅ Type checking is enforced (string, int, bool, etc.)
  • ✅ If json_schema is null or empty, NO parameters are allowed
  • ❌ Extra parameters not in schema will cause validation failure
  • ❌ Missing required parameters will cause validation failure

9. Credentials Management

9.1 Using rda_secret_name

Instead of passing credentials directly, you pass a secret name that the wrapper modules use to securely load credentials from the vault:

@dm:dynamic-bot
    bot_name="my_bot"
    secret_name is "api-cred-131-35"
    host is "192.168.131.35"

9.2 Using rda_requests for HTTP/API Calls

The rda_requests module is a wrapper around the standard requests library that securely loads credentials:

Bot Configuration:

{
  "bot_name": "api_collector",
  "imports": [
    { "module": "rda_requests", "as": "requests" },
    { "module": "json" },
    { "module": "rda_logging.rda_logger", "as": "logging" }
  ]
}

Template Usage:

<%
logger = logging.getLogger("api_collector")

# Create a session - this is the custom rda_requests.Session
session = requests.Session()

# Make requests using rda_secret_name parameter
# The wrapper will automatically load credentials from the vault
response = session.get(
    f"https://{host}/api/v2/data",
    rda_secret_name=secret_name,  # Secret loaded securely
    verify=False,
    timeout=30
)

if response.ok:
    data = response.json()
    output["value"] = data
else:
    output["value"] = [{"status": "Failed", "reason": response.text}]
%>

How rda_secret_name Works:

  • The secret is loaded from the vault using rda_secret_name
  • For HTTP auth secrets (type: httpauth or http-oauth):
  • username and password → Basic Auth
  • headers → Added to request headers
  • payload → Added to request body
  • secure_p1_key/value, secure_p2_key/value → Added to payload
  • secure_h1_key/value, secure_h2_key/value → Added to headers

Bot Name: requests_new_test

JSON Schema:

{
  "type": "object",
  "properties": {
    "ip": {
      "type": "string",
      "description": "RDA Platform IP"
    },
    "secret_name": {
      "type": "string",
      "description": "Name of the secret in the vault to load credential from"
    }
  },
  "required": ["ip"]
}

Imports:

[
  {
    "module": "pathlib",
    "items": {
      "Path": "Path"
    }
  },
  {
    "module": "rda_requests",
    "as": "requests"
  }
]

Mako Template:

<%
SERVER_URL = f'http://{ip}/api/v2/jobs'

session = requests.Session()
resp = session.get(SERVER_URL, rda_secret_name=secret_name, verify=False)
resp.raise_for_status()
resp = resp.json()

output['value'] = resp.get("rda_jobs", []) %>

Invocation:

@dm:dynamic-bot 
    bot_name="requests_new_test" 
    ip="192.168.131.35" 
    secret_name="api-cred-131-35"

9.3 Using rda_paramiko for SSH Connections

The rda_paramiko module wraps the standard paramiko library for secure SSH access:

Bot Configuration:

{
  "bot_name": "ssh_executor",
  "imports": [
    { "module": "rda_paramiko", "as": "paramiko" },
    { "module": "rda_logging.rda_logger", "as": "logging" }
  ]
}

Template Usage:

<%
logger = logging.getLogger("ssh_executor")

# Create SSH client
client = paramiko.SSHClient()

# Connect using rda_secret_name
# The wrapper will automatically load SSH credentials from the vault
client.connect(
    host,
    rda_secret_name=secret_name  # Secret loaded securely
)

# Execute commands
stdin, stdout, stderr = client.exec_command("ls -la")
output_data = stdout.read().decode()

client.close()

output["value"] = [{"command_output": output_data}]
%>

How SSH rda_secret_name Works:

  • The secret is loaded from the vault (type: ssh-cred or device-host-ssh)
  • Automatically sets:
  • username from the secret
  • password (if provided)
  • private_key (if provided)
  • port (defaults to 22)

Bot Name: paramiko_new_test

JSON Schema:

{
  "type": "object",
  "properties": {
    "host": {
      "type": "string",
      "description": "Host to connect to"
    },
    "secret_name": {
      "type": "string",
      "description": "Name of the secret in the vault to load credential from"
    }
  },
  "required": ["host"]
}

Imports:

[
  {
    "module": "pathlib",
    "items": {
      "Path": "Path"
    }
  },
  {
    "module": "rda_paramiko",
    "as": "paramiko"
  }
]

Mako Template:

<%
ssh = paramiko.SSHClient()
session = ssh.connect(host, rda_secret_name=secret_name)
stdin, stdout, stderr = session.exec_command("ls", timeout=30)
raw_data = stdout.read().decode(encoding="UTF-8")

output['value'] = [{"raw_data": raw_data}] %>

Invocation:

@dm:dynamic-bot 
    bot_name="paramiko_new_test" 
    host="192.168.131.35" 
    secret_name="ssh-cred-131-35"

10. Complete Example

10.1 Bot Configuration

Bot Configuration

10.2 Invocation as Data Source

@dm:dynamic-bot
    bot_name="paramiko_new_test"
    secret_name is "api-cred-131-35"
    host is "192.168.131.35"

10.3 Expected Output

A DataFrame with columns:

  • raw_data: Raw response for command execution
  • execution_status: "Success" or "Failed" (added by framework)
  • execution_timestamp: ISO timestamp (added by framework)

11. Output Status Fields

The framework automatically adds these fields to all output DataFrames:

Field Type Description
execution_status string "Success" if template executed successfully, "Failed" otherwise
reason string Error message if execution failed, empty string otherwise
execution_timestamp string ISO format timestamp (UTC) of execution

12. Error Handling

12.1 Template Execution Errors

If your template raises an exception, the framework returns:

{
    "bot_name": "my_bot",
    "execution_status": "Failed",
    "reason": "Error message details",
    "execution_timestamp": "2025-11-17T10:30:00.000000"
}

12.2 Parameter Validation Errors

If parameters don't match the JSON schema:

Exception: Argument validation failed: 'host' is a required property. Schema path: required

12.3 Best Practices for Error Handling in Templates

<%
try:
    # Your processing logic
    result = perform_operation()

    # Return success
    output["value"] = result
except Exception as e:
    logger.exception(f"Error in bot execution: {e}")

    # Return error status (optional - framework will handle it)
    output["value"] = [{
        "status": "Failed",
        "reason": str(e)
    }]
%>

13. Version Management

13.1 Setting Default Version

{
  "bot_name": "my_bot",
  "version": "1.0.0",
  "default_version": true
}

13.2 Using Specific Version

@dm:dynamic-bot bot_name="my_bot" version="2.0.0" param1="value"

13.3 Using Latest Version (Default)

@dm:dynamic-bot bot_name="my_bot" param1="value"

This uses the version where default_version is true.

14. Advanced Topics

14.1 Custom PStream for Bot Registry

By default, bots are stored in cfx_bots_registry. To use a different pstream:

@dm:dynamic-bot
    bot_name="my_bot"
    pstream_name="my_custom_bot_registry"
    param1="value"

14.2 Combining with Other Data Sources

@dm:recall name is "customer_data"
    --> @dm:dynamic-bot bot_name="enrich_customers" api_key="xyz"
    --> @dm:save name is "enriched_customers"

14.3 Chaining Multiple Bots

@dm:dynamic-bot bot_name="extract_data" source="api"
    --> @dm:dynamic-bot bot_name="transform_data" operation="clean"
    --> @dm:dynamic-bot bot_name="validate_data" schema="strict"

15 Troubleshooting

15.1 Issue: "No bot found with name 'X'"

Solution: Ensure the bot exists in the pstream and default_version is set to true, or specify the version explicitly.

15.2 Issue: "This bot does not accept any parameters"

Solution: The bot has json_schema set to null or empty. Either update the bot configuration or remove the parameters.

15.3 Issue: "No secret with the name 'X' found in the vault"

Solution: Ensure you pass the correct secret name parameter and the secret exists in the vault: secret_name is "my-secret"

15.4 Issue: Template returns text instead of DataFrame

Solution: Use output["value"] = result instead of ${ result }

16 Summary

The Mako Code Executor provides a powerful, flexible way to create dynamic data processing bots:

  1. Define your bot with a Mako template and JSON schema in the Dynamic Bots dashboard under RDA Administration -> Pipelines
  2. Invoke the bot with @dm:dynamic-bot
  3. Access parameters, credentials, and input DataFrames in your template
  4. Return results using output["value"] = your_data
  5. Chain with other data sources for complex workflows