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.

# 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)

# 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:

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:

  - 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.

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

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.

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.

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:

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:

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):

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.

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:

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

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

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.

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.

---

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.

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.

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

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).

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

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.

times — run at specific times of day

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

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:

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

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.