> ## Documentation Index
> Fetch the complete documentation index at: https://docs.datazone.co/llms.txt
> Use this file to discover all available pages before exploring further.
# Actions
> Deploy serverless Python functions to automate workflows and extend agent capabilities
## 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.
```python send_mail_action.py theme={null}
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"
}
```
```yaml config.yaml theme={null}
project_name: my-project
project_id: proj_abc123
actions:
- path: actions/send_mail_action.py
```
* **Project Structure**
## 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:
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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:
```python theme={null}
from datazone.actions import context
```
Log informational messages during action execution.
```python theme={null}
context.log_info("Processing started")
context.log_info(f"Processed {count} records")
```
Log warning messages for potential issues.
```python theme={null}
context.log_warning("Potential issue detected")
context.log_warning(f"Unusual data pattern in {field}")
```
Log error messages when operations fail.
```python theme={null}
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:
```python theme={null}
from datazone.actions import Variable
```
Access workspace variables securely within actions.
```python theme={null}
# 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](/reference/development/variables).
#### Example: Using Variables
```python theme={null}
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:
```python theme={null}
from datazone.actions import execute_query
```
Execute a SQL query against Datazone datasets.
```python theme={null}
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
```python theme={null}
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
}
```
### KnowledgeObject
The **`KnowledgeObject`** client reads and writes [Knowledge Object](/reference/knowledge-objects/overview) instances from within an action. Instances are addressed by their opaque `_key`, and operations run on the action's project (main branch by default).
```python theme={null}
from datazone.actions import KnowledgeObject
```
Access instances of a knowledge object by name.
```python theme={null}
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`).
```python theme={null}
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
```python theme={null}
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](/reference/knowledge-objects/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.
Fetch raw bytes for a file from the file container.
```python theme={null}
# 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
```python theme={null}
# 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:
```python theme={null}
from datazone.actions.clients import CloudFeedClient
```
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
Register your actions in **`config.yaml`**:
```yaml theme={null}
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
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):
```python theme={null}
@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:
```python theme={null}
@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:
```yaml theme={null}
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](/reference/integration/endpoints#action-based-endpoints).
### 2. In Agents
Enable actions as **tools for AI agents**:
When creating an agent:
1. Select **"Action"** in the tools section
2. Choose which actions the agent can use
3. 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](/reference/agents/overview#action-tools).
## Examples
### Slack Notification
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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](/reference/integration/endpoints#action-based-endpoints)
* [Use Actions in Agents](/reference/agents/overview#action-tools)
* [Manage Variables](/reference/development/variables) - Store secure credentials and configuration
* [Learn About Projects](/reference/development/project)