Store records in Postgres

DatabaseSink takes any async SQLAlchemy URL. For PostgreSQL that means the asyncpg driver.

Install the driver

uv add "fastmcp-feedback[instrumentation]" asyncpg

The instrumentation extra brings sqlalchemy[asyncio] (with greenlet) and aiosqlite; asyncpg is the Postgres driver. Add [instrumentation,embeddings] as well if you plan to store embeddings.

Point the sink at the database

import os

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

app = FastMCP("My Server")

# e.g. postgresql+asyncpg://app:secret@db.example.com:5432/app
sink = DatabaseSink(os.environ["DATABASE_URL"], create_tables=True)
mw = instrument(app, [sink])

With create_tables=True the sink creates ffb_tool_calls, ffb_events and ffb_feedback_call_links on its first write, if they are missing. It never alters an existing table. ffb_embeddings is created only when an EmbeddingSink first writes, so a server that does not embed never needs pgvector.

Two processes starting against the same empty database at once is handled: a failed CREATE is retried once, and the retry skips what the other process made.

Check that rows arrive

Records reach the database from a background task, so flush before you look:

import asyncio

from fastmcp import Client


@app.tool
def ping() -> str:
    return "pong"


async def main():
    async with Client(app) as client:
        await client.call_tool("ping", {})
    await mw.flush()  # wait until queued records have been written

    from sqlalchemy import text
    async with sink.engine.connect() as conn:
        count = (await conn.execute(text("SELECT count(*) FROM ffb_tool_calls"))).scalar()
    print("rows:", count)
    await mw.aclose()


asyncio.run(main())

On shutdown, await mw.aclose() flushes the queue and closes the sinks. A sink built from a URL disposes its engine then; an engine you passed in is left alone.

Postgres-specific behavior

JSON columns (identity, args, result, extra, attrs) are JSONB. Event attributes are made JSON-safe first, so NaN and infinities are stored as the strings "nan" and "inf" rather than failing the batch.
Timestamps are TIMESTAMP WITH TIME ZONE, in UTC.
Query JSON fields with ->>, for example extra->>'job_id'.

The full layout is in the database schema reference.