✨ Introducing Embedded Agents! Drop your Datazone agents into any website, app, or third-party tool with a single line of code. See Embedding Agents for details.
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@actiondef 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" }
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.
Access workspace variables securely within actions.
# Access a variableapi_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.
from datazone.actions import action, context, execute_query@actiondef 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 }
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).
employees = KnowledgeObject("Employee") # main branchemployees = 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@actiondef 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}
from datazone.actions import action, KnowledgeObject@actiondef 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 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.
from datazone.actions import action, context, Variablefrom datazone.actions.clients import CloudFeedClientimport json@actiondef 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}
Actions can raise exceptions - they’ll be captured and returned:
@actiondef validate_input(value: int): """Validate input value.""" if value < 0: raise ValueError("Value must be positive") return {"validated": True, "value": value}
from datazone.actions import action, context@actiondef 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 }
