Skip to main content
Datazone Agents

​
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.

​
Configuration

​
Add to config.yaml

Register your actions in 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
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:
  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.

​
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
import requests

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

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

    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