Record background work with events

Some things worth recording are not tool calls: a job that changes state minutes after the tool that started it returned, a phase inside a long call, a watcher thread noticing something. Record them as events. They travel the same queue as call records and land in ffb_events.

Record an event

from fastmcp import FastMCP
from fastmcp_feedback.instrumentation import DatabaseSink, instrument

app = FastMCP("Solver")
mw = instrument(app, [DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True)])


@app.tool
async def submit(case: str) -> dict:
    job_id = f"job-{case}"
    mw.record_event("job.submitted", key=job_id, attrs={"case": case, "nproc": 8})
    return {"job_id": job_id}

record_event(kind, *, key=None, attrs=None, user_sub=None, call_id=None) is synchronous and returns True when the event was queued. It never raises and never blocks; when the queue is full the event is dropped and counted, like a call record.

• kind is a dotted name such as job.submitted, at most 128 characters.
• key is what you join on, such as a job id. It is stored as a string of at most 255 characters.
• attrs is any JSON-like dict. It is redacted like call arguments and made JSON-safe: NaN and infinities become "nan" and "inf", datetimes become ISO strings, anything else unknown becomes str(value).

Inside and outside a tool call

An event recorded while a tool is running gets that call’s id as call_id and its session, so it joins to the call’s row. Outside a call, call_id is None. A task the tool started that records after the tool returned counts as outside; pass call_id= yourself if you kept the id.

record_event is safe from any thread. Off the server’s event loop (a watcher thread, a sync tool running in a worker thread) the event is handed to the loop with call_soon_threadsafe. The loop is known once the server has handled a tool call; before that, a call from a thread with no loop returns False.

import threading


def watch(job_id: str) -> None:
    # e.g. tail the solver's log; here, report at once
    mw.record_event("job.finished", key=job_id, attrs={"duration_s": 3512, "converged": True})


@app.tool
async def start(case: str) -> dict:
    job_id = f"job-{case}"
    mw.record_event("job.started", key=job_id)
    threading.Thread(target=watch, args=(job_id,)).start()
    return {"job_id": job_id}

Choose keys that stay unique

A key you join on must be unique for at least your retention period, including across restarts. A per-boot counter (speech-1, speech-2, …) starts over on every restart, and the second boot’s speech-1 then joins to the first boot’s events. Use a UUID, or a token chosen at startup plus a counter:

import itertools
import secrets

BOOT = secrets.token_hex(4)
_counter = itertools.count(1)


def next_speech_id() -> str:
    return f"speech-{BOOT}-{next(_counter)}"

Read a key’s events

events_for returns a key’s events oldest first, with occurred_at as a UTC datetime:

import asyncio

from fastmcp import Client


async def main():
    async with Client(app) as client:
        await client.call_tool("start", {"case": "wing"})
    await asyncio.sleep(0.1)  # let the watcher thread report
    for ev in await mw.events_for("job-wing"):
        print(ev["occurred_at"].isoformat(), ev["kind"], ev["call_id"] is not None, ev["attrs"])
    finished = await mw.events_for("job-wing", kinds=["job.finished"], limit=100)
    print(len(finished))
    await mw.aclose()


asyncio.run(main())

2026-10-01T18:02:11.304512+00:00 job.started True {}
2026-10-01T18:02:11.305893+00:00 job.finished False {'duration_s': 3512, 'converged': True}
1

It reads the DatabaseSink and adds events still in the queue, so an event is visible as soon as record_event returns. With more than limit (1000) matches it returns the newest limit. Without a DatabaseSink it answers from the last 2000 events kept in memory (recent_events= on the middleware).

Join a call to its events

Events recorded inside a call carry its id, so e.call_id = c.id joins those. For events recorded later, have the enricher write the key into the call’s extra, then join on it. On SQLite:

SELECT e.occurred_at, e.kind, e.attrs
FROM ffb_events e
JOIN ffb_tool_calls c ON json_extract(c.extra, '$.job_id') = e.key
WHERE c.id = :call_id
ORDER BY e.occurred_at;

On PostgreSQL the condition is c.extra->>'job_id' = e.key.

Caution

ffb_tool_calls has no call_id column. Its primary key is id, and ffb_events.call_id holds that value. Ids are random UUIDs, so always order by occurred_at (events) or started_at (calls), never by id.

Sampling and retention

Events are always recorded: exclusion and sampling apply to tool calls only. A call of a sampled tool that recorded an event is always kept, so the event’s call_id has a row to join to. Events recorded inside an excluded tool have no call_id. Retention deletes events by occurred_at.

In mode off, record_event does nothing and returns False.