Skip to main content

​
What are Actions?

Actions are the fundamental building blocks of Ziet. They’re Python functions decorated with @Action that become serverless, scalable units of work with automatic:
  • Retries with exponential backoff
  • Timeout protection
  • Error handling
  • Observability and logging
  • Serverless scaling

​
Basic Action

from ziet import Action, memory

@Action(
    id="process_payment",
    name="Process Payment",
    description="Process payment transaction"
)
def process_payment(amount: int, currency: str = "usd") -> None:
    """Process a payment transaction."""
    # Your logic here
    payment_result = {
        "status": "success",
        "amount": amount,
        "currency": currency
    }
    
    # Store result in memory
    memory.add(key="payment_result", value=payment_result)

​
Action Parameters

Configure action behavior with decorator parameters: 
from ziet import Action, memory

@Action(
    id="custom_action",
    name="Custom Action",
    description="What this does",
    timeout=300,
    retries=3
)
def my_action(param: str) -> None:
    result = {"result": param}
    memory.add(key="action_result", value=result)

​
Parameter Details

​
id
string
Unique identifier for the action. Defaults to module.function_name.
@Action(id="process_payment", name="Process Payment", description="Process payment")
def process_payment(amount: int) -> None:
    # Process and store in memory
    memory.add(key="payment", value={"amount": amount})
​
description
string
required
Human-readable description of what the action does. Used in logs and dashboard.
@Action(
    id="charge_customer",
    name="Charge Customer",
    description="Charge customer via Stripe"
)
def charge_customer(amount: int) -> None:
    # Process payment and store in memory
    memory.add(key="charge", value={"amount": amount})
​
timeout
int
default:"300"
Maximum execution time in seconds. Action is terminated if exceeded.
@Action(
    id="quick_operation",
    name="Quick Operation",
    description="Quick operation",
    timeout=30
)
def quick_operation() -> None:
    # Complete within 30 seconds
    result = perform_operation()
    memory.add(key="operation_result", value=result)
​
retries
int
default:"0"
Number of retry attempts on failure. Uses exponential backoff.
@Action(
    id="api_call",
    name="API Call",
    description="Call external API",
    retries=3
)
def unreliable_api_call() -> None:
    response = call_external_api()
    memory.add(key="api_response", value=response)

​
Retry Behavior

Actions automatically retry on exceptions using exponential backoff: 
from ziet import Action, memory

@Action(
    id="flaky_api_call",
    name="API Call",
    description="Call external API with retries",
    retries=3
)
def flaky_api_call(url: str) -> None:
    response = requests.get(url)
    response.raise_for_status()  # Raises on 4xx/5xx
    
    # Store response in memory
    memory.add(key=f"api_{url}", value=response.json())
Retry schedule:
  • Attempt 1: Immediate execution
  • Attempt 2: Wait 2 seconds
  • Attempt 3: Wait 4 seconds
  • Attempt 4: Wait 8 seconds
  • Max backoff: 30 seconds
What triggers a retry:
  • Any unhandled exception
  • Timeout errors
  • Network errors
What doesn’t trigger a retry:
  • Successful execution (no exception)
  • Explicit return statement

​
Timeout Protection

Prevent actions from running indefinitely: 
from ziet import Action, memory

@Action(
    id="long_running_task",
    name="Long Running Task",
    description="Expensive computation with timeout",
    timeout=60
)
def long_running_task(data: dict) -> None:
    # If this takes > 60s, execution is terminated
    # and a TimeoutError is raised
    result = expensive_computation(data)
    
    # Store result in memory
    memory.add(key="computation_result", value=result)
Important: Set appropriate timeouts based on your action’s expected duration.
  • Too short: Premature termination
  • Too long: Wasted resources on hung operations

​
Error Handling

​
Automatic Error Handling

Actions automatically log errors and provide stack traces: 
from ziet import Action, memory

@Action(
    id="risky_operation",
    name="Risky Operation",
    description="Risky operation with automatic error handling",
    retries=2
)
def risky_operation(data: dict) -> None:
    if not data.get("valid"):
        raise ValueError("Invalid data")
    
    result = process(data)
    memory.add(key="operation_result", value=result)
If this fails:
  1. Error is logged with full stack trace
  2. Action retries (if retries > 0)
  3. Memory stores error details
  4. Dashboard shows failure reason

​
Manual Error Handling

Add custom error handling: 
from ziet import Action, memory

@Action(
    id="safe_operation",
    name="Safe Operation",
    description="Operation with custom error handling"
)
def safe_operation(data: dict) -> None:
    try:
        result = risky_call(data)
        memory.add(key="success", value=result)
        memory.add(key="status", value="success")
    
    except ValueError as e:
        memory.add(key="validation_error", value={"error": str(e)})
        memory.add(key="status", value="validation_error")
    
    except Exception as e:
        memory.add(key="unexpected_error", value={"error": str(e)})
        memory.add(key="status", value="unexpected_error")

​
Using Integrations

Actions can use built-in integrations: 
from ziet import Action, memory
from ziet.integrations import google, openai, stripe

@Action(
    id="research_topic",
    name="Research Topic",
    description="Research and summarize topic",
    timeout=60
)
def research_topic(topic: str) -> None:
    # Search Google
    search_results = google.search(topic, num_results=10)
    
    # Summarize with OpenAI
    prompt = f"Summarize: {search_results}"
    summary = openai.chat([{"role": "user", "content": prompt}])
    
    # Store in memory
    memory.add(key="topic", value=topic)
    memory.add(key="summary", value=summary)
    memory.add(key="search_results", value=search_results)


@Action(
    id="charge_customer",
    name="Charge Customer",
    description="Process payment via Stripe",
    timeout=30,
    retries=2
)
def charge_customer(amount: int, email: str) -> None:
    # Create Stripe payment
    intent = stripe.create_payment_intent(
        amount=amount * 100,  # Convert to cents
        currency="usd",
        metadata={"customer_email": email}
    )
    
    # Store payment info in memory
    memory.add(key="payment_intent_id", value=intent["id"])
    memory.add(key="payment_status", value=intent["status"])
    memory.add(key="payment_amount", value=amount)
View all integrations →

​
Using Memory

Share data between actions using memory: 
from ziet import Action, memory

@Action(
    id="fetch_user",
    name="Fetch User",
    description="Fetch user data from database"
)
def fetch_user(user_id: str) -> None:
    user = database.get_user(user_id)
    
    # Store in memory for other actions
    memory.add(key=f"user_{user_id}", value=user)


@Action(
    id="send_notification",
    name="Send Notification",
    description="Send notification email"
)
def send_notification(user_id: str, message: str) -> None:
    # Retrieve from memory (avoid redundant database call)
    user = memory.get(f"user_{user_id}")
    
    if not user:
        # Fetch if not in memory
        fetch_user(user_id)
        user = memory.get(f"user_{user_id}")
    
    # Send email
    result = send_email(user["email"], message)
    memory.add(key="notification_sent", value=result)
Learn more about Memory →

​
Testing Actions

Test your actions locally before deploying: 
ziet run --local action my_action
This runs your action in local mode with test data. You’ll see:
  • Action execution
  • Memory operations
  • Integration calls (mocked locally)
  • Execution time and logs
Expected output: 
🧪 Running action locally: my_action

[INFO] Starting local run...
[INFO] Action my_action initialized
[INFO] Action executing...
[INFO] Memory added: key=result, value=10
[INFO] Action my_action completed in 0.05s
✅ Local run completed

Result stored in memory: {"result": 10}

​
Best Practices

Each action should do one thing well.✅ Good
@Action(
    id="validate_email",
    name="Validate Email",
    description="Validate email format"
)
def validate_email(email: str) -> None:
    is_valid = "@" in email and "." in email
    memory.add(key="email_valid", value=is_valid)

@Action(
    id="send_welcome_email",
    name="Send Welcome Email",
    description="Send welcome email"
)
def send_welcome_email(email: str) -> None:
    result = sendgrid.send(email, "Welcome!")
    memory.add(key="email_sent", value=result)
❌ Bad
@Action(description="Validate and send email")
def validate_and_send(email: str) -> None:
    # Doing too much in one action
    if "@" not in email:
        memory.add(key="error", value="Invalid email")
        return
    result = sendgrid.send(email, "Welcome!")
    memory.add(key="email_sent", value=result)
Use clear, action-oriented names.✅ Good
  • fetch_user_profile
  • calculate_shipping_cost
  • send_confirmation_email
❌ Bad
  • get_data
  • do_thing
  • process
Add docstrings explaining what the action does.
@Action(
    id="calculate_tax",
    name="Calculate Tax",
    description="Calculate tax amount",
    timeout=10
)
def calculate_tax(amount: float, region: str) -> None:
    """
    Calculate tax for a given amount and region.
    
    Args:
        amount: The subtotal amount in USD
        region: Two-letter region code (e.g., "CA", "NY")
    
    Stores:
        tax_amount: Tax amount in USD
    
    Raises:
        ValueError: If region is invalid
    """
    tax_rates = {"CA": 0.0725, "NY": 0.04, "TX": 0.0625}
    
    if region not in tax_rates:
        raise ValueError(f"Unknown region: {region}")
    
    tax_amount = amount * tax_rates[region]
    memory.add(key="tax_amount", value=tax_amount)
Match timeout to expected duration.
# Quick operations
@Action(
    id="validate_input",
    name="Validate Input",
    description="Validate input data",
    timeout=10
)
def validate_input(data: dict) -> None:
    is_valid = validate(data)
    memory.add(key="validation_result", value=is_valid)

# Medium operations
@Action(
    id="call_external_api",
    name="Call API",
    description="Call external API",
    timeout=60
)
def call_external_api(url: str) -> None:
    response = requests.get(url)
    memory.add(key="api_response", value=response.json())

# Long operations
@Action(
    id="process_large_dataset",
    name="Process Dataset",
    description="Process large dataset",
    timeout=300
)
def process_large_dataset(data: list) -> None:
    result = expensive_processing(data)
    memory.add(key="processed_data", value=result)
Add retries for operations that might fail transiently.
# Network calls - use retries
@Action(
    id="fetch_from_api",
    name="Fetch from API",
    description="Fetch data from external API",
    retries=3,
    timeout=30
)
def fetch_from_api(url: str) -> None:
    response = requests.get(url)
    memory.add(key=f"api_data_{url}", value=response.json())

# Pure computation - no retries needed
@Action(
    id="calculate_result",
    name="Calculate Result",
    description="Calculate result",
    timeout=10
)
def calculate_result(x: int, y: int) -> None:
    result = x + y
    memory.add(key="calculation_result", value=result)

​
Limitations

Current limitations (subject to change):
  • Max execution time: 15 minutes (900 seconds)
  • Max memory: 10GB per action
  • Max payload size: 6MB input/output
  • No persistent file system: Use memory or external storage
  • No background threads: All work must complete before return

​
Next Steps

Agents

Learn how to orchestrate actions with agents

Memory

Master memory operations

Integrations

Explore built-in integrations

Strategies

Add manual approval and wait conditions