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

# Build a Background App

> Run on a schedule and submit context to the proactive agent on your Truffle

Background apps run on a schedule and feed **context** into Truffle's proactive agent. Foreground apps are callable tools; background apps are context producers.

You subclass `BackgroundWorkerApp`, implement a handful of methods, and call `app.main()`. The runtime handles the schedule, retries, and the gRPC plumbing.

> The richer and more concrete your submissions are, the better proactive behavior you'll get. "FED-RATE-SEP moved up 12c (was 41c, now 53c)" routes much better than "market alert".

***

## Background patterns in the examples

The bundled examples use background apps in a few distinct ways:

| Example  | Shape  | What the background does                                                                           |
| -------- | ------ | -------------------------------------------------------------------------------------------------- |
| `arxiv`  | Hybrid | Uses text-configured research interests to find relevant papers and submit recommendation context. |
| `notion` | Hybrid | Uses remote MCP OAuth state to scan workspace changes and submit digest context.                   |
| `viator` | Hybrid | Runs without auth and submits travel/location context around Viator search tools.                  |
| `whoop`  | Hybrid | Shares OAuth token helpers with the foreground and submits health/recovery summaries.              |

The common pattern is: keep API/auth clients in shared modules, keep polling/diffing logic in a plain worker, and keep runtime submission logic in the `BackgroundWorkerApp` subclass.

***

## The worker + app split

Reference apps consistently keep the *work* (API calls, parsing, diffing) in a plain `Worker` class, and the *SDK wiring* in a small `BackgroundWorkerApp` subclass. This makes the worker easy to unit test without touching the runtime.

```
my_app/
├── bg_worker.py           # plain class: verify(), run_cycle() → CycleResult
├── my_app_background.py   # BackgroundWorkerApp subclass + app.main()
└── truffile.yaml
```

***

## Minimal example

### The worker (`bg_worker.py`)

The worker knows nothing about the SDK. It returns a `CycleResult` dataclass describing what happened.

```python theme={null}
# bg_worker.py
from __future__ import annotations

import os
from dataclasses import dataclass, field

import httpx


@dataclass(frozen=True)
class CycleResult:
    summary: str | None = None
    uris: tuple[str, ...] = ()
    priority: int = 0
    changed: bool = False
    auth_error: str | None = None
    error: str | None = None


class MyBackgroundWorker:
    def __init__(self) -> None:
        self._http = httpx.Client(timeout=30.0)
        self._last_fingerprint: str | None = None

    def close(self) -> None:
        try:
            self._http.close()
        except Exception:
            pass

    def verify(self) -> tuple[bool, str]:
        """Called by `truffile deploy` (and `--verify`) to check credentials."""
        token = os.getenv("MY_APP_ACCESS_TOKEN", "").strip()
        if not token:
            return False, "MY_APP_ACCESS_TOKEN is not set"
        try:
            response = self._http.get(
                "https://api.example.com/me",
                headers={"Authorization": f"Bearer {token}"},
            )
            if response.status_code in (401, 403):
                return False, "MY_APP_ACCESS_TOKEN is invalid or expired"
            response.raise_for_status()
            user = response.json().get("login", "unknown")
            return True, f"Verified as @{user}"
        except Exception as exc:
            return False, f"Verify request failed: {exc}"

    def run_cycle(self) -> CycleResult:
        token = os.getenv("MY_APP_ACCESS_TOKEN", "").strip()
        if not token:
            return CycleResult(auth_error="MY_APP_ACCESS_TOKEN is not set")
        try:
            response = self._http.get(
                "https://api.example.com/activity",
                headers={"Authorization": f"Bearer {token}"},
            )
            if response.status_code in (401, 403):
                return CycleResult(auth_error=f"API returned {response.status_code}")
            response.raise_for_status()
            items = response.json().get("items", [])
        except Exception as exc:
            return CycleResult(error=str(exc))

        # Skip if nothing has changed since the last cycle.
        fingerprint = repr(items)
        if fingerprint == self._last_fingerprint:
            return CycleResult(changed=False)
        self._last_fingerprint = fingerprint

        if not items:
            return CycleResult(changed=False)

        summary = "\n".join(f"- {item['title']} ({item['url']})" for item in items[:5])
        uris = tuple(item["url"] for item in items[:5])
        return CycleResult(summary=summary, uris=uris, priority=1, changed=True)
```

### The app (`my_app_background.py`)

```python theme={null}
# my_app_background.py
from __future__ import annotations

import atexit

from truffile.app_runtime import BackgroundWorkerApp

from bg_worker import CycleResult, MyBackgroundWorker


class MyBackgroundApp(BackgroundWorkerApp[MyBackgroundWorker, CycleResult]):
    def __init__(self) -> None:
        super().__init__("myapp", logger_name="myapp.background")

    def build_worker(self) -> MyBackgroundWorker:
        return MyBackgroundWorker()

    def verify_worker(self, worker: MyBackgroundWorker) -> tuple[bool, str]:
        return worker.verify()

    def run_cycle(self, worker: MyBackgroundWorker) -> CycleResult:
        return worker.run_cycle()

    def handle_cycle_result(self, ctx, result: CycleResult) -> None:
        if result.auth_error:
            self.report_auth_failure(ctx, result.auth_error)
            return

        self.reset_auth_failures()

        if result.error:
            self.logger.error("Cycle failed: %s", result.error)
            return

        if not result.changed or not result.summary:
            return

        self.submit_text(
            ctx,
            content=result.summary,
            uris=result.uris,
            priority=result.priority,
        )


app = MyBackgroundApp()


def _cleanup() -> None:
    worker = getattr(app, "_worker", None)
    if worker is not None:
        close = getattr(worker, "close", None)
        if callable(close):
            try:
                close()
            except Exception:
                pass


atexit.register(_cleanup)


if __name__ == "__main__":
    app.main()
```

That's the whole pattern. Everything below explains each method.

***

## The `BackgroundWorkerApp` base class

Subclass `BackgroundWorkerApp[WorkerT, ResultT]` with the two types of your worker and your cycle result. Call `super().__init__(name, logger_name=...)`.

You implement four methods:

| Method                             | What it does                                                                  |
| ---------------------------------- | ----------------------------------------------------------------------------- |
| `build_worker()`                   | Returns an instance of your worker. Called once and cached.                   |
| `verify_worker(worker)`            | Returns `(ok, message)` — used during deploy and `--verify`.                  |
| `run_cycle(worker)`                | Runs one iteration. Returns your cycle result.                                |
| `handle_cycle_result(ctx, result)` | Inspects the result and decides whether to submit context or report an error. |

And you get these helpers for free:

| Helper                                                       | What it does                                                                                                                                 |
| ------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------- |
| `self.submit_text(ctx, content=..., uris=..., priority=...)` | Queue text context for the proactive agent.                                                                                                  |
| `self.report_auth_failure(ctx, description)`                 | Dedup auth failures; after `AUTH_FAILURE_THRESHOLD` (default 3) consecutive failures, reports to the runtime with `needs_intervention=True`. |
| `self.reset_auth_failures()`                                 | Clear the counter after a successful cycle.                                                                                                  |
| `self.logger`                                                | A pre-configured logger.                                                                                                                     |

***

## `app.main()` and `--verify`

`app.main()` is the entry point:

```python theme={null}
if __name__ == "__main__":
    app.main()
```

It handles two modes automatically:

* **Normal mode** — starts the background loop and blocks. The runtime calls `run_cycle()` on each schedule tick.
* **`--verify` mode** — when `--verify` is in `sys.argv`, `main()` calls `verify_worker()`, prints the result, and exits with `0` (ok) or `1` (failed). `truffile deploy` uses this to check credentials immediately after your `text` step collects them:

  ```yaml theme={null}
  - name: Configure access token
    type: text
    fields:
      - name: my_app_access_token
        label: Access Token
        type: password
        env: MY_APP_ACCESS_TOKEN
    validator:
      type: bash
      run: python ./my_app_background.py --verify
  ```

***

## Handling cycle results

`handle_cycle_result(ctx, result)` is where you decide what to do with each cycle's output. The pattern is roughly:

1. If it's an **auth error**, report it (deduped) and return.
2. Otherwise, reset the auth failure counter.
3. If it's a **runtime error**, log it and return. (The runtime will still retry next cycle.)
4. If nothing changed, return without submitting.
5. Otherwise, call `submit_text(...)` with a well-formed context string.

```python theme={null}
def handle_cycle_result(self, ctx, result: CycleResult) -> None:
    if result.auth_error:
        self.report_auth_failure(ctx, result.auth_error)
        return
    self.reset_auth_failures()
    if result.error:
        self.logger.error("Cycle failed: %s", result.error)
        return
    if not result.changed or not result.summary:
        return
    self.submit_text(
        ctx,
        content=result.summary,
        uris=result.uris,
        priority=result.priority,
    )
```

### Auth failure dedup

Background apps poll often. If a token goes bad, you don't want a noisy failure report every 30 minutes — you want one clear report once the situation is clearly stuck.

`self.report_auth_failure(ctx, description)` handles this for you:

* Logs a warning on the first few failures
* After `AUTH_FAILURE_THRESHOLD` (default: 3) consecutive failures, reports to the runtime with `needs_intervention=True` so the user is prompted
* Subsequent failures stay suppressed until a successful cycle calls `self.reset_auth_failures()`

You can override `AUTH_FAILURE_THRESHOLD` on your subclass if 3 is too few or too many.

***

## Calling your own foreground tools from the background

If your app already has a foreground with tools like `gmail_search`, `github_list_issues`, or `kalshi_get_positions`, your background doesn't need to reimplement that logic — it can call its own foreground tools directly and reuse all the auth, parsing, and error handling that lives there.

The runtime gives your background a lease on the foreground container via **`ctx.connect_foreground()`**. It returns a `ForegroundConnection` that speaks MCP to your foreground.

### Basic shape

```python theme={null}
from __future__ import annotations

import asyncio

from truffile.app_runtime.background import (
    BackgroundRunContext,
    run_background,
)


async def _run_cycle(ctx: BackgroundRunContext) -> None:
    fg = await ctx.connect_foreground()
    try:
        # call any tool your foreground app exposes, by name
        result = await fg.call_tool("my_tool", query="hello", limit=5)
        ...
    finally:
        await fg.close()


def my_background(ctx: BackgroundRunContext) -> None:
    asyncio.run(_run_cycle(ctx))


if __name__ == "__main__":
    run_background(my_background)
```

Key points:

* `ctx.connect_foreground()` is **async**, so your cycle has to be too. The standard pattern is a sync entry point (`def my_background(ctx)`) that calls `asyncio.run(_run_cycle(ctx))` into an async worker.
* `fg.call_tool(name, **kwargs)` takes the tool name and any keyword arguments. The kwargs become the tool's input — exactly the same as if the agent had called it.
* Always call `await fg.close()` in a `finally` block so the lease is released when the cycle ends.

<Note>
  Because this pattern needs an async function with direct access to `ctx`, it's easiest to write with the lower-level `run_background(fn)` entry point instead of `BackgroundWorkerApp`.
</Note>

### Real example: reuse foreground auth

The most common reason to reach into the foreground is to reuse its auth check. Instead of rewriting token loading and validation in `bg_worker.py`, just call your foreground's `check_auth` tool:

```python theme={null}
async def _run_cycle(ctx: BackgroundRunContext) -> None:
    fg = await ctx.connect_foreground()
    try:
        # 1. reuse the foreground's auth check
        auth_result = await fg.call_tool("google_check_auth")
        auth_text = _extract_text(auth_result)
        if "error" in auth_text.lower() or "not authenticated" in auth_text.lower():
            _handle_auth_failure(ctx, auth_text)
            return

        # 2. now make domain calls through foreground tools
        search_result = await fg.call_tool(
            "gmail_search",
            query="is:inbox is:unread -from:me",
            max_results=30,
        )
        messages = _parse_search_results(_extract_text(search_result))

        # 3. drill into each thread via another foreground tool
        unanswered = []
        for msg in messages[:10]:
            thread = await fg.call_tool("gmail_get_thread", thread_id=msg["thread_id"])
            if not _user_has_replied(_extract_text(thread)):
                unanswered.append(msg)

        # 4. submit a context message the same way any other bg cycle would
        if unanswered:
            ctx.bg.submit_context(
                content=_build_digest(unanswered),
                uris=[f"https://mail.google.com/mail/u/0/#inbox/{m['thread_id']}" for m in unanswered],
                priority=1,
            )
    finally:
        await fg.close()
```

### What `call_tool` returns

`call_tool(name, **kwargs)` returns an MCP `CallToolResult` object. The result content is a list of content blocks — for text tools, that's usually one or more `TextContent` blocks. A small helper to extract the text:

```python theme={null}
def _extract_text(result) -> str:
    if hasattr(result, "content"):
        parts = []
        for block in result.content:
            if hasattr(block, "text"):
                parts.append(block.text)
        return "\n".join(parts)
    return str(result)
```

If your foreground tools return JSON with `ok(...)` / `err(...)`, parse the extracted text with `json.loads`.

### Listing available tools

If you need to discover what tools the foreground exposes (useful during development or for dynamic dispatch):

```python theme={null}
tools = await fg.list_tools()
for tool in tools:
    print(tool.name, "-", tool.description)
```

### Always close the connection

Wrap your foreground calls in a `try` / `finally` and call `await fg.close()` when you're done — even on exceptions. Skipping this leaves the foreground container held open longer than it needs to be.

```python theme={null}
fg = await ctx.connect_foreground()
try:
    ...
finally:
    await fg.close()
```

### Using this with `BackgroundWorkerApp`

The `run_background(fn)` entry point is the cleanest fit for foreground calls because it's async-native. If you're already using `BackgroundWorkerApp` for a simpler cycle and want to mix in a foreground call, bridge from sync to async inside `handle_cycle_result`:

```python theme={null}
import asyncio

class MyBackgroundApp(BackgroundWorkerApp[MyWorker, CycleResult]):
    ...

    def handle_cycle_result(self, ctx, result: CycleResult) -> None:
        if result.needs_foreground_followup:
            asyncio.run(self._followup(ctx, result))

    async def _followup(self, ctx, result: CycleResult) -> None:
        fg = await ctx.connect_foreground()
        try:
            await fg.call_tool("my_refresh_tool")
        finally:
            await fg.close()
```

The tradeoff is one extra event loop per cycle. For apps that *mostly* live in foreground calls, prefer the `run_background(fn)` pattern end-to-end.

***

## Writing good context submissions

The quality of your background context *directly* affects how proactive behavior feels. Rich, concrete items win every time.

**Weak:**

> Market alert

**Strong:**

> Price alert: FED-RATE-SEP moved up 12c (was 41c, now 53c)
> Related: [https://kalshi.com/markets/FED-RATE-SEP](https://kalshi.com/markets/FED-RATE-SEP)

**Weak:**

> New PRs in your repos

**Strong:**

> @octocat opened PR #342 in truffle-ai/pyfw-codex: "Fix mDNS timeout on retry"
> 2 files changed, +18/-4. Requested reviews: @alice, @bob
> [https://github.com/truffle-ai/pyfw-codex/pull/342](https://github.com/truffle-ai/pyfw-codex/pull/342)

A few guidelines:

* **Include the entities.** Names, tickers, IDs, usernames, repo paths, order numbers.
* **Include numbers with context.** `$12k` alone is useless; `$12k (was $8.4k yesterday)` is actionable.
* **Include links.** Pass them in `uris=(...)` so the agent can open them if it decides to act.
* **Pick priority honestly.** Default priority is fine for most things. Bump it for genuine urgency (alerts, settlements, revoked tokens).
* **Don't spam.** Use a fingerprint or diff to skip cycles where nothing meaningfully changed — return a `CycleResult` with `changed=False` and don't submit.

<Note>
  Background context can trigger action in *any* app, not just yours. If an Instagram message about an Amazon order lands in the proactive agent's context, Truffle might use your Amazon app to add the item to cart. Write submissions with that in mind.
</Note>

***

## Schedule configuration

The `default_schedule` block in `truffile.yaml` under `metadata.background` controls when your cycle runs. Three policy types are supported: **`interval`**, **`always`**, and **`times`**.

### `interval` — run every N minutes/hours

The most common mode. Your cycle runs, yields, sleeps for the configured duration, then runs again.

```yaml theme={null}
metadata:
  background:
    process:
      cmd: ["python", "my_app_background.py"]
      working_directory: /
      environment:
        PYTHONUNBUFFERED: "1"
    default_schedule:
      type: interval
      interval:
        duration: 30m
        schedule:
          daily_window: "00:00-23:59"
```

* **`duration`** — interval between runs. Accepts `ms`, `s`, `m`, `h`, `d` suffixes (e.g. `5m`, `30m`, `2h`, `1d`).
* **`schedule.daily_window`** — optional time window that gates execution. `"09:00-18:00"` only runs during business hours; `"00:00-23:59"` (the effective default) runs any time.
* **`schedule.allowed_days`** — optional list like `[mon, tue, wed, thu, fri]` to restrict to specific weekdays.
* **`schedule.forbidden_days`** — optional inverse: list of days to skip (e.g. `[sat, sun]`). Mutually exclusive with `allowed_days`.

<Tip>
  During development, set `duration: 1m` so you can see results quickly. Bump it back up before shipping.
</Tip>

### `always` — keep running as long as there's work

For apps that need to react to live data the moment it arrives. The runtime keeps your process alive continuously instead of running-then-sleeping. Use it when a fixed interval would either miss updates (too slow) or burn cycles (too fast).

```yaml theme={null}
metadata:
  background:
    process:
      cmd: ["python", "my_app_background.py"]
      working_directory: /
    default_schedule:
      type: always
```

No duration, no schedule block — just `type: always`.

Good fits:

* Watching a websocket or server-sent event stream (market data, chat messages, push notifications)
* Apps that need to submit context to the proactive agent **the moment** something happens — e.g. a Kalshi watcher that fires a high-priority submission the instant a position moves, not 30 minutes later
* Any integration where the upstream service pushes data instead of you polling for it

<Warning>
  Under `always`, your cycle function is expected to block (on a websocket read, a queue, whatever) rather than return quickly. If you return immediately, the runtime will just call you again and you'll burn CPU doing nothing. Write your loop to stay inside `run_cycle` for as long as the connection is alive, and return only when you want the runtime to restart you.
</Warning>

### `times` — run at specific times of day

For apps that should run at fixed clock times (e.g. a daily 9am digest):

```yaml theme={null}
metadata:
  background:
    process:
      cmd: ["python", "my_app_background.py"]
    default_schedule:
      type: times
      times:
        run_times:
          - "09:00"
          - "17:00"
        allowed_days: [mon, tue, wed, thu, fri]
```

* **`run_times`** — list of `HH:MM` (or `HH:MM:SS`) times of day, each triggers one cycle.
* **`allowed_days`** / **`forbidden_days`** — optional day-of-week gating, same semantics as `interval`.

***

## Cleanup on shutdown

Close your worker's network clients on exit:

```python theme={null}
import atexit

def _cleanup() -> None:
    worker = getattr(app, "_worker", None)
    if worker is not None:
        close = getattr(worker, "close", None)
        if callable(close):
            try:
                close()
            except Exception:
                pass

atexit.register(_cleanup)
```

This isn't optional. Leaving HTTP connections and other resources open at process exit is the most common source of flaky redeploys.

***

## Deploy and verify

```bash theme={null}
truffile validate ./my-app
truffile deploy --dry-run ./my-app
truffile deploy ./my-app
```

Once deployed:

* Your `run_cycle()` will be invoked on the configured interval
* `handle_cycle_result()` decides what to submit
* `submit_text(...)` calls land in the proactive agent's context
* Auth failures past the threshold surface as "needs intervention" in the app UI

***

## When to build a background app

Use a background app when you want:

* **Periodic polling** of an external system (new messages, new PRs, new fills, new feed items)
* **Monitoring** with thresholds (price alerts, rate changes, error budgets)
* **Digest generation** (daily/weekly/hourly summaries)
* **Context emission** that feeds proactive actions in other apps

If you *also* need callable tools, add a `foreground` section to the same `truffile.yaml` — it becomes a hybrid app. Use the same shared client/auth module from both entry points, and keep background-specific diffing or polling logic in a plain worker class so it stays easy to test.