Actions - Datazone

Overview

Actions allow you to deploy serverless Python functions that can be triggered on-demand by endpoints or used as tools by AI agents. Think of actions as lambda-like functions that run in isolated environments.

from datazone.actions import action

@action
def send_email(to: str, subject: str, body: str):
  """
  Send an email to a recipient.
  """
  # Your email sending logic here
  print(f"Sending email to {to}")

  # Return structured result
  return {
      "status": "sent",
      "recipient": to,
      "timestamp": "2026-02-11T10:00:00Z"
  }

Project Structure

my-project

config.yml

actions

send_mail_action.py

What Are Actions?

Actions are custom Python functions that run in isolated environments, accept parameters, return structured results, and can be triggered via endpoints or used as agent tools.

Use Cases

Send notifications
Process data
Call external APIs
Generate reports
Automate workflows
Extend AI agent capabilities with custom logic

Action with Logging

Use context to log information during execution:

from datazone.actions import action, context

@action
def process_data(dataset_id: str, operation: str = "transform"):
    """
    Process dataset with specified operation.

    Args:
        dataset_id: ID of dataset to process
        operation: Type of operation (default: transform)
    """
    context.log_info(f"Starting {operation} on dataset {dataset_id}")

    try:
        # Your processing logic
        result_count = 42
        context.log_info(f"Processed {result_count} records")

        return {
            "status": "success",
            "dataset_id": dataset_id,
            "records_processed": result_count
        }
    except Exception as e:
        context.log_error(f"Processing failed: {str(e)}")
        raise

Action with Optional Parameters

from datazone.actions import action, context

@action
def generate_report(report_type: str, format: str = "pdf", email: str = None):
    """
    Generate and optionally email a report.

    Args:
        report_type: Type of report to generate
        format: Output format (default: pdf)
        email: Optional email address to send report
    """
    context.log_info(f"Generating {report_type} report in {format} format")

    report_url = f"https://example.com/reports/{report_type}.{format}"

    if email:
        context.log_info(f"Sending report to {email}")
        # Send email logic

    return {
        "report_url": report_url,
        "format": format,
        "emailed": email is not None
    }

Actions SDK Reference

Decorator

from datazone.actions import action

@action
def my_function():
    """Your function logic"""
    pass

The @action decorator marks a function as executable by Datazone.

Context API

The context object provides logging capabilities during action execution:

from datazone.actions import context

context.log_info

function

Log informational messages during action execution.

context.log_info("Processing started")
context.log_info(f"Processed {count} records")

context.log_warning

function

Log warning messages for potential issues.

context.log_warning("Potential issue detected")
context.log_warning(f"Unusual data pattern in {field}")

context.log_error

function

Log error messages when operations fail.

context.log_error("Operation failed")
context.log_error(f"Failed to connect: {str(e)}")

All logs are collected and returned with the action response.

Variable API

The Variable class allows actions to access secure variables stored in Datazone:

from datazone.actions import Variable

Variable

class

Access workspace variables securely within actions.

# Access a variable
api_key = Variable("API_KEY")

# Use as a string (Variable stringifies to its value)
headers = {"Authorization": f"Bearer {Variable('AUTH_TOKEN')}"}

Variables are workspace-scoped and can be marked as secret to encrypt sensitive data like API keys, passwords, and tokens. Learn more in the Variables documentation.

Example: Using Variables

from datazone.actions import action, Variable

@action
def send_notification(message: str):
    """Send notification using API key from Variables."""
    api_key = str(Variable("API_KEY"))

    print(f"Sending notification with key: {api_key[:4]}...")

    return {"status": "sent", "message": message}

Security Best Practice: Always store sensitive credentials as Variables rather than hardcoding them in your action code.

execute_query

The execute_query function runs SQL queries against your Datazone datasets and returns structured results:

from datazone.actions import execute_query

execute_query

function

Execute a SQL query against Datazone datasets.

result = execute_query("SELECT * FROM my_dataset LIMIT 10")
print(result.data)    # list of row dicts
print(result.status)  # query status string

Returns a QueryResult object with:

data — list of row dictionaries
status — execution status string

Example: Query and Transform

from datazone.actions import action, context, execute_query

@action
def summarize_orders(min_amount: float = 100.0):
    """Return order totals above a threshold."""
    context.log_info(f"Querying orders with amount >= {min_amount}")

    result = execute_query(f"""
        SELECT customer_id, SUM(amount) AS total
        FROM orders
        WHERE amount >= {min_amount}
        GROUP BY customer_id
        ORDER BY total DESC
        LIMIT 20
    """)

    context.log_info(f"Found {len(result.data)} customers")

    return {
        "status": result.status,
        "row_count": len(result.data),
        "rows": result.data
    }

FileContainerClient

FileContainerClient provides access to files stored in the Datazone file container (LakeFS/S3-compatible storage). It is automatically initialized as file_client in agent code execution environments — no import is required.

file_client is pre-initialized in agent code execution contexts. It is not available inside @action decorated functions.

file_client.read

method

Fetch raw bytes for a file from the file container.

# path is relative to the file container root
raw = file_client.read("reports/data.csv")

path — path relative to the file container root
Returns raw bytes
Raises ValueError if the file exceeds 10 MB or cannot be fetched

Examples: Reading Different File Types

# CSV / plain text
text = file_client.read("data.csv").decode("utf-8")
print(text[:500])

# JSON
import json
config = json.loads(file_client.read("config.json"))
print(config["setting"])

# Excel
import io
import openpyxl
wb = openpyxl.load_workbook(io.BytesIO(file_client.read("report.xlsx")))
ws = wb.active
for row in ws.iter_rows(values_only=True):
    print(row)

# Agent-uploaded file
raw = file_client.read(f"agent-uploads/{agent_id}/{upload_id}/{filename}")

CloudFeedClient

CloudFeedClient is a simplified SAP connector for reading tables and executing function modules from within actions:

from datazone.actions.clients import CloudFeedClient

CloudFeedClient

class

SAP CloudFeed client for table access and function execution.Constructor parameters:

base_url — base URL of the SAP system
username — SAP username
password — SAP password
timeout (optional) — request timeout in seconds (default: 30)
max_retries (optional) — max retries on failure (default: 10)
backoff_factor (optional) — exponential backoff factor (default: 0.3)

Method	Description
`get_table_list(filter_name, filter_desc, rows)`	List available SAP tables, supports wildcard filters
`get_table_schema(table_name)`	Get field definitions for a table
`preview_table(table_name, rows)`	Preview table contents
`execute_function(obj_name, input_data, output_fields, commit)`	Execute a SAP function module

Example: Query SAP Sales Orders

from datazone.actions import action, context, Variable
from datazone.actions.clients import CloudFeedClient

@action
def fetch_sap_orders(table_name: str = "VBAP", preview_rows: int = 10):
    """Fetch sales order data from SAP."""
    client = CloudFeedClient(
        base_url=str(Variable("SAP_BASE_URL")),
        username=str(Variable("SAP_USERNAME")),
        password=str(Variable("SAP_PASSWORD")),
    )

    context.log_info(f"Fetching schema for {table_name}")
    schema = client.get_table_schema(table_name)
    fields = [f["FIELDNAME"] for f in schema]

    context.log_info(f"Previewing {preview_rows} rows")
    preview = client.preview_table(table_name, rows=preview_rows)

    return {
        "table": table_name,
        "fields": fields,
        "preview": preview
    }

Example: Execute SAP Function Module

from datazone.actions import action, context, Variable
from datazone.actions.clients import CloudFeedClient
import json

@action
def create_sap_order(material: str, quantity: int, plant: str):
    """Create a purchase order in SAP via function module."""
    with CloudFeedClient(
        base_url=str(Variable("SAP_BASE_URL")),
        username=str(Variable("SAP_USERNAME")),
        password=str(Variable("SAP_PASSWORD")),
    ) as client:
        input_data = json.dumps({
            "MATERIAL": material,
            "QUANTITY": quantity,
            "PLANT": plant
        })

        context.log_info(f"Creating order for {material} x{quantity} at {plant}")
        result = client.execute_function(
            obj_name="Z_CREATE_PO",
            input_data=input_data,
            commit=True
        )

        context.log_info("Order created successfully")
        return {"result": result, "material": material, "quantity": quantity}

Example: Search and Inspect SAP Tables

from datazone.actions import action, context, Variable
from datazone.actions.clients import CloudFeedClient

@action
def search_sap_tables(name_filter: str = "VBAP*"):
    """List SAP tables matching a filter and return their schemas."""
    client = CloudFeedClient(
        base_url=str(Variable("SAP_BASE_URL")),
        username=str(Variable("SAP_USERNAME")),
        password=str(Variable("SAP_PASSWORD")),
    )

    tables = client.get_table_list(filter_name=name_filter, rows=20)
    context.log_info(f"Found {len(tables)} tables matching '{name_filter}'")

    result = []
    for table in tables[:5]:  # inspect first 5
        name = table["TABNAME"]
        schema = client.get_table_schema(name)
        result.append({"table": name, "field_count": len(schema)})

    return result

Configuration

Add to config.yaml

project_name: my-project
project_id: proj_abc123

actions:
  - path: actions/send_email.py
  - path: actions/process_data.py
  - path: actions/generate_report.py
  - path: workflows/cleanup_data.py

Repository Structure

config.yml

actions

send_email.py

process_data.py

generate_report.py

workflows

Each file should contain one @action decorated function. Action function names must be unique within your project.

Return Values

Actions should return structured data (dict, list, or primitives):

@action
def my_action():
    # Good - structured dictionary
    return {
        "status": "success",
        "data": [1, 2, 3],
        "metadata": {"count": 3}
    }

    # Good - simple values
    return 42

    # Good - lists
    return [{"id": 1}, {"id": 2}]

Error Handling

Actions can raise exceptions - they’ll be captured and returned:

@action
def validate_input(value: int):
    """Validate input value."""
    if value < 0:
        raise ValueError("Value must be positive")

    return {"validated": True, "value": value}

Using Actions

1. In Endpoints

Connect actions to API endpoints for webhook-style triggers:

endpoints:
  - name: send-notification
    type: action
    config:
      action_id: "507f1f77bcf86cd799439011"

When the endpoint is called, the action executes automatically. Learn more in the Endpoints documentation.

2. In Agents

Enable actions as tools for AI agents: When creating an agent:

Select “Action” in the tools section
Choose which actions the agent can use
The agent will automatically call actions when needed

The AI agent decides when and how to use actions based on user questions. Learn more in the Agents documentation.

Examples

Slack Notification

from datazone.actions import action, context
import requests

@action
def send_slack_message(channel: str, message: str, webhook_url: str):
    """
    Send message to Slack channel.

    Args:
        channel: Slack channel name
        message: Message to send
        webhook_url: Slack webhook URL
    """
    context.log_info(f"Sending message to {channel}")

    payload = {
        "channel": channel,
        "text": message
    }

    response = requests.post(webhook_url, json=payload)

    if response.status_code == 200:
        context.log_info("Message sent successfully")
        return {"status": "sent", "channel": channel}
    else:
        context.log_error(f"Failed to send: {response.text}")
        raise Exception(f"Slack API error: {response.status_code}")

Data Validation

from datazone.actions import action, context

@action
def validate_customer_data(customer_id: str, email: str, age: int):
    """
    Validate customer data before processing.

    Args:
        customer_id: Customer identifier
        email: Customer email address
        age: Customer age
    """
    errors = []

    # Validate email
    if "@" not in email:
        errors.append("Invalid email format")

    # Validate age
    if age < 18 or age > 120:
        errors.append("Age must be between 18 and 120")

    if errors:
        context.log_error(f"Validation failed: {', '.join(errors)}")
        return {
            "valid": False,
            "errors": errors
        }

    context.log_info("Validation passed")
    return {
        "valid": True,
        "customer_id": customer_id
    }

API Integration

from datazone.actions import action, context, Variable
import requests

@action
def fetch_weather(city: str):
    """
    Fetch current weather for a city.

    Args:
        city: City name
    """
    context.log_info(f"Fetching weather for {city}")

    api_key = str(Variable("WEATHER_API_KEY"))
    url = f"https://api.weather.com/data?city={city}&key={api_key}"

    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()

        data = response.json()
        context.log_info("Weather data retrieved")

        return {
            "city": city,
            "temperature": data["temp"],
            "conditions": data["conditions"],
            "humidity": data["humidity"]
        }
    except requests.exceptions.RequestException as e:
        context.log_error(f"API request failed: {str(e)}")
        raise

Next Steps

Create Action Endpoints
Use Actions in Agents
Manage Variables - Store secure credentials and configuration
Learn About Projects

​Overview

​What Are Actions?

​Use Cases

​Action with Logging

​Action with Optional Parameters

​Actions SDK Reference

​Decorator

​Context API

​Variable API

​Example: Using Variables

​execute_query

​Example: Query and Transform

​FileContainerClient

​Examples: Reading Different File Types

​CloudFeedClient

​Example: Query SAP Sales Orders

​Example: Execute SAP Function Module

​Example: Search and Inspect SAP Tables

​Configuration

​Add to config.yaml

​Repository Structure

​Return Values

​Error Handling

​Using Actions

​1. In Endpoints

​2. In Agents

​Examples

​Slack Notification

​Data Validation

​API Integration

​Next Steps

Overview

What Are Actions?

Use Cases

Action with Logging

Action with Optional Parameters

Actions SDK Reference

Decorator

Context API

Variable API

Example: Using Variables

execute_query

Example: Query and Transform

FileContainerClient

Examples: Reading Different File Types

CloudFeedClient

Example: Query SAP Sales Orders

Example: Execute SAP Function Module

Example: Search and Inspect SAP Tables

Configuration

Add to config.yaml

Repository Structure

Return Values

Error Handling

Using Actions

1. In Endpoints

2. In Agents

Examples

Slack Notification

Data Validation

API Integration

Next Steps