Skip to main content

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.

@on_boot registers a function to run before the SSE listener subscribes to the platform: useful for warming caches, opening database pools, validating environment, or pre-fetching data. @on_shutdown registers a function to run during graceful shutdown. 
from xpander_sdk import on_boot, on_shutdown

@on_boot
async def warm_caches():
    print("Boot: warming caches…")
    await prefetch_models()

@on_shutdown
async def cleanup():
    print("Shutting down…")
    await close_connections()
Both decorators accept either sync or async functions, and you can register multiple handlers: the runtime invokes them in registration order.

​
Decorator forms

@on_boot
def fn(): ...

@on_boot(configuration=config)
async def fn(): ...

@on_shutdown
def fn(): ...

@on_shutdown(configuration=config)
async def fn(): ...
ParameterTypeDefaultDescription
configurationConfigurationNoneReserved for future use. Currently only the Events module reads config; boot/shutdown handlers receive no arguments.

​
Lifecycle ordering

Process start
  ↓
@on_boot handlers (in registration order, await one at a time)
  ↓
Events.start(): subscribes to SSE, begins task dispatch
  ↓
… task dispatch ongoing …
  ↓
SIGINT / SIGTERM received
  ↓
Events.stop(): cancels tracked tasks, closes thread pool
  ↓
@on_shutdown handlers (in registration order, errors logged but don't block)
  ↓
Process exit
If a @on_boot handler raises, the worker fails to start: the exception propagates and the process exits. If a @on_shutdown handler raises, the error is logged but the runtime continues with the next shutdown handler.

​
Examples

​
Open a DB pool at boot, close on shutdown

import asyncpg
from xpander_sdk import on_boot, on_shutdown

pool = None

@on_boot
async def open_pool():
    global pool
    pool = await asyncpg.create_pool(dsn="postgresql://…")

@on_shutdown
async def close_pool():
    if pool:
        await pool.close()

​
Validate env at boot

import os
from xpander_sdk import on_boot

@on_boot
def check_env():
    for var in ("OPENAI_API_KEY", "REDIS_URL"):
        if not os.environ.get(var):
            raise RuntimeError(f"Missing required env var: {var}")
Raising in @on_boot aborts startup before the worker takes any tasks: useful for fail-fast environment validation.

​
Multiple handlers

@on_boot
def first():
    print("1")

@on_boot
def second():
    print("2")
Both run, in declaration order: 1 then 2. The runtime awaits each one before moving on.

​
Sync and async mix

@on_boot
def synchronous():
    print("sync boot")

@on_boot
async def asynchronous():
    print("async boot")
    await asyncio.sleep(0)
Sync handlers are called directly; async ones are awaited. Either works.

​
Notes

  • Boot/shutdown handlers are class-level on Events. They register globally per process: there’s no scoping to a specific Events instance. This is fine for a one-process-one-agent model.
  • Shutdown handlers run after the SSE listener is stopped but before final cleanup. By then the process is no longer accepting new tasks; in-flight tasks are cancelled.