> ## Documentation Index
> Fetch the complete documentation index at: https://docs.ziet.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Actions
> Deep dive into Ziet Actions - serverless functions with built-in retries, timeouts, and observability
## 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
```python theme={null}
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:
```python theme={null}
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
Unique identifier for the action. Defaults to `module.function_name`.
```python theme={null}
@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})
```
Human-readable description of what the action does. Used in logs and dashboard.
```python theme={null}
@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})
```
Maximum execution time in seconds. Action is terminated if exceeded.
```python theme={null}
@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)
```
Number of retry attempts on failure. Uses exponential backoff.
```python theme={null}
@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:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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 โ](/integrations/overview)
## Using Memory
Share data between actions using memory:
```python theme={null}
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 โ](/core/memory)
## Testing Actions
Test your actions locally before deploying:
```bash theme={null}
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**
```python theme={null}
@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**
```python theme={null}
@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.
```python theme={null}
@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.
```python theme={null}
# 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.
```python theme={null}
# 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
Learn how to orchestrate actions with agents
Master memory operations
Explore built-in integrations
Add manual approval and wait conditions