Actions

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

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})

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})

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)

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:

Error is logged with full stack trace
Action retries (if retries > 0)
Memory stores error details
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

Single Responsibility

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)

Descriptive Names

Use clear, action-oriented names.✅ Good

fetch_user_profile
calculate_shipping_cost
send_confirmation_email

❌ Bad

get_data
do_thing
process

Document Behavior

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)

Set Appropriate Timeouts

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)

Use Retries for Network Calls

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

Getting Started

Agentic

Application

Integrations

Deployment

What are Actions?

Basic Action

Action Parameters

Parameter Details

Retry Behavior

Timeout Protection

Error Handling

Automatic Error Handling

Manual Error Handling

Using Integrations

Using Memory

Testing Actions

Best Practices

Limitations

Next Steps

Agents

Memory

Integrations

Strategies

​What are Actions?

​Basic Action

​Action Parameters

​Parameter Details

​Retry Behavior

​Timeout Protection

​Error Handling

​Automatic Error Handling

​Manual Error Handling

​Using Integrations

​Using Memory

​Testing Actions

​Best Practices

​Limitations

​Next Steps

Agents

Memory

Integrations

Strategies

What are Actions?

Basic Action

Action Parameters

Parameter Details

Retry Behavior

Timeout Protection

Error Handling

Automatic Error Handling

Manual Error Handling

Using Integrations

Using Memory

Testing Actions

Best Practices

Limitations

Next Steps