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_name: my-project
project_id: proj_abc123
actions:
- path: actions/send_mail_action.py

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

Inside an action you can:

Invoke an LLM — call a model through your model account with Agent, optionally with structured output
Run SQL queries — read from your datasets with execute_query
Read and write knowledge objects — get, list, update and batch upsert instances with KnowledgeObject
Use secure variables — access API keys and credentials with Variable, without hardcoding them
Log execution details — emit info, warning and error logs with context
Connect to SAP — read tables and execute function modules with CloudFeedClient
Call any external API — the runtime ships with requests, so any HTTP service is one call away

Which makes actions a good fit for sending notifications, processing data, generating reports, automating workflows, and extending 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

function

Log informational messages during action execution.

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

function

Log warning messages for potential issues.

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

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

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

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
    }

Agent

The Agent class invokes an LLM through one of your Model Accounts, without any provider SDK or API key in your code — the call is proxied by Datazone and billed against your workspace.

from datazone.actions import Agent

class

Create an agent bound to a model account and model.Parameters:

model (optional) — model to use, e.g. CLAUDE_45_SONNET. Defaults to your organisation’s default model.
model_account (optional) — model account id. Defaults to your organisation’s AI settings account.
response_format (optional) — a Pydantic model or JSON schema dict for structured output.

agent = Agent(model="CLAUDE_45_SONNET")
result = agent.invoke("Summarize this in one sentence: ...")
result["content"]

method

Invoke the model and return a result dictionary.Accepts either a plain string or a message list:

agent.invoke("Say hi")

agent.invoke({
    "messages": [
        {"role": "system", "content": "You are concise."},
        {"role": "user", "content": "Say hi"},
    ]
})

Message roles are system, user and assistant. Returns a dict with:

content — the raw text answer (empty when response_format is used)
structured_response — the parsed object when response_format is used, otherwise None
model — the model that produced the answer

Raises RuntimeError if the invocation fails.

Example: Summarize Text

from datazone.actions import action, Agent

@action
def summarize(text: str):
    """Summarize a piece of text in one sentence."""
    agent = Agent(model="CLAUDE_45_SONNET")

    result = agent.invoke(f"Summarize in one sentence:\n\n{text}")

    return {"summary": result["content"]}

Example: Structured Output

Pass a Pydantic model as response_format to get a validated object back instead of raw text:

from typing import Literal

from pydantic import BaseModel, Field

from datazone.actions import action, context, Agent

class TicketTriage(BaseModel):
    """Triage decision for a support ticket."""
    priority: Literal["low", "medium", "high"] = Field(description="Urgency of the ticket")
    team: str = Field(description="Team that should own it, e.g. billing, platform, data")
    summary: str = Field(description="One sentence summary of the problem")

@action
def triage_ticket(subject: str, body: str):
    """Classify an incoming support ticket."""
    agent = Agent(model="CLAUDE_45_SONNET", response_format=TicketTriage)

    result = agent.invoke({
        "messages": [
            {"role": "system", "content": "You triage incoming support tickets. Be concise."},
            {"role": "user", "content": f"Subject: {subject}\n\n{body}"},
        ]
    })

    triage = result["structured_response"]
    context.log_info(f"Triaged as {triage.priority} for {triage.team}")

    return {"priority": triage.priority, "team": triage.team, "summary": triage.summary}

response_format also accepts a plain JSON schema dict — in that case structured_response is returned as a dict rather than a model instance.

Token usage is recorded per invocation and attributed to the action, so LLM costs show up alongside your other workspace usage.

KnowledgeObject

The KnowledgeObject client reads and writes Knowledge Object instances from within an action. Instances are addressed by their opaque _key, and operations run on the action’s project (main branch by default).

from datazone.actions import KnowledgeObject

class

Access instances of a knowledge object by name.

employees = KnowledgeObject("Employee")            # main branch
employees = KnowledgeObject("Employee", branch="feature-x")

Methods:

get(key, add_relationships=False) — fetch one instance (a dict) by its _key.
list(filters=None, page=1, page_size=50, fields=None) — iterate pages of instances (see below).
update(key, payload) — partially update an instance; returns the updated instance.
delete(key) — delete an instance by its _key.
batch_upsert(payloads) — insert/update up to 1000 instances in one call; returns {"created", "updated", "total"}.

Filtering and pagination. list(...) returns an iterator of pages; each page exposes .items and .total_count. filters is a list of {"column", "operator", "value"} objects combined with AND (operators: equal, not_equal, contains, not_contains, greater_than, less_than).

from datazone.actions import action, context, KnowledgeObject

@action
def deactivate_inactive_employees():
    """Mark employees with no recent login as inactive."""
    employees = KnowledgeObject("Employee")
    updated = 0

    for page in employees.list(
        filters=[{"column": "is_active", "operator": "equal", "value": True}]
    ):
        for employee in page.items:
            context.log_info(f"Checking {employee['name']}")
            employees.update(key=employee["_key"], payload={"is_active": False})
            updated += 1

    return {"updated": updated}

Example: Read and upsert

from datazone.actions import action, KnowledgeObject

@action
def sync_employee(employee_id: int, email: str):
    """Read one employee and upsert a batch."""
    employees = KnowledgeObject("Employee")

    # Read a single instance by its key
    existing = employees.get(key="8A3F9C2E1B4D6F7A8A3F9C2E1B4D6F7A")

    # Insert or update many at once
    result = employees.batch_upsert([
        {"id": employee_id, "name": "John", "email": email},
        {"id": employee_id + 1, "name": "Jane", "email": "jane@acme.com"},
    ])

    return {"existing": existing, "upsert": result}

See the Actions SDK page under Knowledge Objects for the full method reference.

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.

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

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

​Agent

​Example: Summarize Text

​Example: Structured Output

​KnowledgeObject

​Example: Read and upsert

​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

Agent

Example: Summarize Text

Example: Structured Output

KnowledgeObject

Example: Read and upsert

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