> ## Documentation Index
> Fetch the complete documentation index at: https://docs.capsule.new/llms.txt
> Use this file to discover all available pages before exploring further.

# Decorators

> Lifecycle, message, task, schedule, endpoint, and ASGI decorators.

Capsule decorators fall into two families:

* **functional app decorators** called on `app`
* **class-based decorators** imported from `cpsl`

## Worked flow

In a real app, decorators usually work together rather than in isolation:

```python theme={null}
@app.boot()
async def on_boot():
    print("[support-ops] runtime ready")


@app.message()
async def handle(session: cpsl.Session, msg: cpsl.Message):
    if "sync" in msg.text.lower():
        handle = await sync_threads.submit(session=session)
        await session.show_task(handle, message="Thread sync started")
        return

    await session.reply("Ask for a sync or ask for analysis.")


@app.task(retries=2, lock="mailbox:sync")
async def sync_threads(session: cpsl.Session | None = None):
    if session:
        await session.notify("Syncing Gmail threads...")
    ...


@app.schedule("0 * * * *")
async def hourly_sync():
    await sync_threads.submit()
```

That one slice already shows the usual Capsule pattern:

* lifecycle hooks prepare the runtime
* message handlers decide what the user wants
* tasks own side effects and longer work
* schedules keep the app moving without user input

## Functional app decorators

These require `app = cpsl.App(..., image=cpsl.Image(), ...)`.

| Decorator             | Purpose                               |
| --------------------- | ------------------------------------- |
| `@app.boot()`         | Run once when the runtime starts      |
| `@app.shutdown()`     | Run when the runtime is shutting down |
| `@app.enter()`        | Run when a new session is created     |
| `@app.exit()`         | Run when a session closes             |
| `@app.message()`      | Handle inbound messages               |
| `@app.task(...)`      | Register a background task            |
| `@app.schedule(cron)` | Register a cron job                   |
| `@app.endpoint(...)`  | Register an HTTP endpoint             |
| `@app.asgi(path=...)` | Mount an ASGI app                     |

## Class-based decorators

When you use `@app.cls(...)`, the handler methods use the global decorator versions:

* `@cpsl.boot()`
* `@cpsl.shutdown()`
* `@cpsl.enter()`
* `@cpsl.exit()`
* `@cpsl.message()`
* `@cpsl.task(...)`
* `@cpsl.schedule(cron)`
* `@cpsl.endpoint(...)`
* `@cpsl.asgi(path=...)`

## Lifecycle hooks

### `boot()`

Runs once when the runtime starts.

```python theme={null}
@app.boot()
async def on_boot():
    print("ready")
```

### `shutdown()`

Runs during runtime shutdown. Good for cleanup and flush logic.

```python theme={null}
@app.shutdown()
async def on_shutdown():
    print("cleaning up")
```

### `enter()`

Runs when a new session is created. Receives `session`.

```python theme={null}
@app.enter()
async def on_enter(session: cpsl.Session):
    session.data["started"] = True
```

### `exit()`

Runs when a session closes.

```python theme={null}
@app.exit()
async def on_exit(session: cpsl.Session):
    print(f"closing {session.id}")
```

## `message()`

Registers the main inbound handler.

```python theme={null}
from openai import AsyncOpenAI

client = AsyncOpenAI()

@app.message()
async def handle(session: cpsl.Session, msg: cpsl.Message):
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a concise Capsule assistant."},
            *session.chat_messages(msg),
        ],
    )
    await session.reply(response.choices[0].message.content or "No response.")
```

For most apps, this is the main entry point. In a real app, remember to include `image=cpsl.Image(python_packages=["openai"])` and `secrets=["OPENAI_API_KEY"]` on `App(...)` if you want to call the OpenAI SDK directly.

## `task(...)`

Registers a background task and turns the function into a `TaskDescriptor`.

### Arguments

| Argument       | Meaning                                    |
| -------------- | ------------------------------------------ |
| `retries`      | Max retry attempts                         |
| `timeout`      | Task timeout in seconds                    |
| `lock`         | Lock template like `"user:{user_id}"`      |
| `retry_for`    | Exception types that should retry          |
| `callback_url` | URL to POST completion/failure payloads to |
| `process`      | Run in a separate OS process               |

### Example

```python theme={null}
@app.task(retries=2, timeout=60)
async def sync_threads(session: cpsl.Session | None = None):
    if session:
        await session.notify("Syncing Gmail threads...")
    await threads.insert_one(
        {
            "subject": "Renewal review",
            "status": "needs-review",
            "priority": "high",
            "thread_id": "thread_123",
        }
    )
```

Submitting it looks like this:

```python theme={null}
handle = await sync_threads.submit(session=session)
await session.show_task(handle, message="Syncing mailbox...")
```

After decoration, the task object supports:

* `.submit(...)`
* `.schedule(...)`
* `.find(...)`
* `.count(...)`
* `.cancel(...)`

## `schedule(cron)`

Registers a cron handler.

```python theme={null}
@app.schedule("*/5 * * * *")
async def every_five_minutes():
    ...
```

The cron string is standard 5-field cron syntax in UTC.

A useful pattern is to keep the schedule thin and hand work off to a task:

```python theme={null}
@app.schedule("0 * * * *")
async def hourly_sync():
    await sync_threads.submit()
```

## `endpoint(method="GET", path="/", authorized=True)`

Registers a plain HTTP handler.

```python theme={null}
@app.endpoint(method="POST", path="/callback", authorized=False)
async def callback(ctx: cpsl.RequestContext):
    ...
```

Use `authorized=False` for public callback handlers.

In a real handler you would usually inspect `ctx.request` or just return structured data:

```python theme={null}
@app.endpoint(method="GET", path="/health", authorized=False)
async def health(ctx: cpsl.RequestContext):
    return {"ok": True}
```

## `asgi(path="/app")`

Mounts a full ASGI application such as FastAPI or Starlette.

```python theme={null}
from fastapi import FastAPI

api = FastAPI()


@app.asgi(path="/api")
def api_app():
    return api
```

## See also

* [App API Reference](/reference/app-api)
* [Tasks And Scheduling](/concepts/tasks-and-scheduling)
* [Session And Request Context Reference](/reference/session-and-request-context)