> ## Documentation Index
> Fetch the complete documentation index at: https://docs.xpander.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Application Lifecycle Management
> Learn how to manage application initialization and cleanup using @on_boot and @on_shutdown decorators
### Virtual Environment Setup
```bash setup.sh theme={"dark"}
python3 -m venv .venv
source .venv/bin/activate
pip install "xpander-sdk[agno]"
```
### Environment Setup
The `.env` file is created when you download your agent code:
```bash theme={"dark"}
xpander init
```
This downloads your agent code from the platform with the `.env` file pre-configured with your keys.
```python lifecycle_management.py theme={"dark"}
from dotenv import load_dotenv
load_dotenv()
from xpander_sdk import Backend, on_boot, on_shutdown, on_task
from agno.agent import Agent
import asyncio
# Global resources that need lifecycle management
database_connection = None
api_cache = {}
metrics_collector = None
@on_boot
def validate_environment():
"""Validate environment variables and configuration."""
import os
print("๐ Validating environment...")
required_vars = ["XPANDER_API_KEY", "XPANDER_ORGANIZATION_ID"]
missing = [var for var in required_vars if not os.getenv(var)]
if missing:
raise EnvironmentError(f"Missing required variables: {missing}")
print("โ
Environment validation passed")
@on_boot
async def initialize_database():
"""Initialize database connection."""
global database_connection
print("๐๏ธ Connecting to database...")
# Simulate async database connection
await asyncio.sleep(0.1)
database_connection = {"status": "connected", "pool": "mock_pool"}
print("โ
Database connection established")
@on_boot
def setup_cache():
"""Initialize application cache."""
global api_cache
print("๐ง Initializing cache...")
api_cache = {
"user_sessions": {},
"api_responses": {},
"temp_data": {}
}
print("โ
Cache initialized")
@on_boot
async def start_metrics_collection():
"""Start background metrics collection."""
global metrics_collector
print("๐ Starting metrics collection...")
await asyncio.sleep(0.05)
metrics_collector = {"active": True, "start_time": "now"}
print("โ
Metrics collection started")
@on_task
async def handle_data_request(task):
"""Process data requests using initialized resources."""
global database_connection, api_cache
print(f"๐จ Processing task: {task.id}")
# Use database connection
if database_connection and database_connection["status"] == "connected":
print("๐ Querying database...")
# Simulate database query
query_result = f"Data for: {task.input.text[:50]}"
else:
query_result = "Database unavailable"
# Use cache for faster responses
cache_key = hash(task.input.text)
if cache_key in api_cache["api_responses"]:
print("โก Using cached response")
cached_response = api_cache["api_responses"][cache_key]
else:
print("๐ Generating new response")
cached_response = f"Processed: {query_result}"
api_cache["api_responses"][cache_key] = cached_response
# Set task result
task.result = {
"status": "completed",
"data": cached_response,
"source": "database" if database_connection else "fallback",
"cached": cache_key in api_cache["api_responses"]
}
print(f"โ
Task {task.id} completed")
return task
@on_shutdown
async def save_cache_to_storage():
"""Save cache data before shutdown."""
global api_cache
print("๐พ Saving cache data...")
if api_cache:
# Simulate saving cache to persistent storage
await asyncio.sleep(0.1)
cache_size = sum(len(str(v)) for v in api_cache.values())
print(f"โ
Saved {cache_size} bytes of cache data")
@on_shutdown
async def close_database_connection():
"""Close database connections gracefully."""
global database_connection
print("๐๏ธ Closing database connections...")
if database_connection:
# Simulate graceful database shutdown
await asyncio.sleep(0.1)
database_connection = None
print("โ
Database connections closed")
@on_shutdown
def stop_metrics_collection():
"""Stop metrics collection and save final report."""
global metrics_collector
print("๐ Stopping metrics collection...")
if metrics_collector and metrics_collector["active"]:
# Simulate saving metrics report
metrics_collector["active"] = False
print("โ
Metrics report saved")
@on_shutdown
def final_cleanup():
"""Final cleanup tasks."""
global api_cache, metrics_collector
print("๐งน Performing final cleanup...")
# Clear sensitive data from memory
if api_cache:
api_cache.clear()
# Final cleanup
print("โ
All cleanup completed - ready for shutdown")
# Initialize backend and agent
backend = Backend()
agno_agent = Agent(**backend.get_args())
# The agent is now ready with proper lifecycle management
print("๐ Agent ready with lifecycle management!")
print("๐ Try asking: 'Process some sample data'")
agno_agent.print_response(message="Process this sample data for analysis")
```
## How to Run
```bash lifecycle_management.py theme={"dark"}
python lifecycle_management.py
```
## What This Example Demonstrates
### Initialization Phase (`@on_boot`)
1. **Environment Validation** - Checks required environment variables before starting
2. **Database Setup** - Establishes database connections asynchronously
3. **Cache Initialization** - Sets up in-memory caches for performance
4. **Metrics Collection** - Starts background monitoring systems
### Task Processing (`@on_task`)
* Uses resources initialized during boot phase
* Demonstrates database queries and cache usage
* Shows how lifecycle management enables shared resources
### Cleanup Phase (`@on_shutdown`)
1. **Data Persistence** - Saves cache data before shutdown
2. **Connection Cleanup** - Gracefully closes database connections
3. **Metrics Finalization** - Stops monitoring and saves reports
4. **Memory Cleanup** - Clears sensitive data from memory
## Key Features
Handlers execute in registration order, allowing proper dependency management
Seamlessly combines synchronous and asynchronous initialization/cleanup tasks
Boot failures prevent startup, shutdown errors are logged but don't block exit
Proper initialization and cleanup of shared resources like databases and caches
## Best Practices Shown
1. **Environment Validation** - Check requirements early in boot process
2. **Resource Dependencies** - Initialize resources in correct order
3. **Graceful Cleanup** - Properly close connections and save data on shutdown
4. **Error Resilience** - Handle cleanup gracefully even if resources are unavailable
5. **Memory Management** - Clear sensitive data during shutdown
## Production Deployment
For creating and deploying agents to production, see the [Setup and Deployment Guide](/Examples/00-setup-deployment).