Real data. Live inventory.

One blocking call stalls everything.

FastMCP runs on FastAPI's ASGI stack — a single-threaded async event loop. When one coroutine is waiting on I/O, the loop switches to another. A blocking database call breaks this model and serializes all requests.

When you call psycopg2.connect() or a synchronous SQLAlchemy session inside an async def tool, the call does not yield to the event loop. It blocks the thread, freezing every other in-flight request until the database responds.

import psycopg2

@mcp.tool()
async def search_products(query: str) -> list[dict]:
    conn = psycopg2.connect(os.environ["DATABASE_URL"])  # BLOCKS
    cur = conn.cursor()
    cur.execute("SELECT * FROM products WHERE name ILIKE %s", (f"%{query}%",))
    return cur.fetchall()

import asyncpg

@mcp.tool()
async def search_products(query: str, ctx) -> list[dict]:
    async with ctx.lifespan_context["pool"].acquire() as conn:
        rows = await conn.fetch(
            "SELECT * FROM products WHERE name ILIKE $1 AND active = true",
            f"%{query}%"
        )
    return [dict(r) for r in rows]

The minimum viable products table.

This schema supports search_products (full-text search on name and description), initiate_checkout (price retrieval, inventory check), and Stripe integration (stripe_price_id column). Run this once against your database before connecting the server.

CREATE EXTENSION IF NOT EXISTS pg_trgm;

CREATE TABLE IF NOT EXISTS products (
    id              SERIAL          PRIMARY KEY,
    name            TEXT            NOT NULL,
    slug            TEXT            NOT NULL UNIQUE,
    description     TEXT,
    price           NUMERIC(10,2)   NOT NULL,
    currency        TEXT            NOT NULL DEFAULT 'USD',
    inventory_count INTEGER         NOT NULL DEFAULT 0,
    stripe_price_id TEXT,
    active          BOOLEAN         NOT NULL DEFAULT TRUE,
    created_at      TIMESTAMPTZ     NOT NULL DEFAULT NOW()
);

-- Speeds up ILIKE search on name and description
CREATE INDEX IF NOT EXISTS idx_products_name ON products (name);
CREATE INDEX IF NOT EXISTS idx_products_active ON products (active);

-- Full-text search index (recommended for catalogs with 1000+ products)
CREATE INDEX IF NOT EXISTS idx_products_fts ON products
    USING GIN (to_tsvector('english', name || ' ' || COALESCE(description, '')));

-- Trigram index for ILIKE search (pg_trgm extension required)
CREATE INDEX IF NOT EXISTS idx_products_name_trgm
    ON products USING GIN (name gin_trgm_ops);

asyncpg — fastest, simplest.

asyncpg is a pure-asyncio Postgres driver with no synchronous fallback. It integrates cleanly with FastMCP's lifespan pattern and is the fastest Python driver for Postgres. Best choice for a small, well-defined schema.

fastmcp>=3.2.0
asyncpg==0.31.0
uvicorn>=0.29.0

import os
import asyncpg
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastmcp import FastMCP

DATABASE_URL = os.environ["DATABASE_URL"]  # postgresql://user:pass@host:PORT/dbname

@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[dict]:
    """Create the connection pool on startup; close it on shutdown."""
    pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=1,
        max_size=3,       # Keep small for Vercel serverless
        command_timeout=30,
        statement_cache_size=0,  # Required for Supabase pooler (port 6543)
    )
    try:
        yield {"pool": pool}
    finally:
        await pool.close()

mcp = FastMCP("ProductServer", lifespan=lifespan)


@mcp.tool()
async def search_products(query: str, ctx) -> list[dict]:
    """Search products by name. Returns active matching products."""
    pool = ctx.lifespan_context["pool"]
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            """
            SELECT id, name, slug, description, price::text, currency, inventory_count
            FROM products
            WHERE active = TRUE
              AND (name ILIKE $1 OR description ILIKE $1)
            ORDER BY name
            LIMIT 20
            """,
            f"%{query}%",
        )
    return [dict(r) for r in rows]


@mcp.tool()
async def initiate_checkout(product_id: int, quantity: int, ctx) -> dict:
    """Check inventory and return product details for checkout."""
    pool = ctx.lifespan_context["pool"]
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            """
            SELECT id, name, price::text, currency, stripe_price_id, inventory_count
            FROM products
            WHERE id = $1 AND active = TRUE
            """,
            product_id,
        )
    if row is None:
        return {"error": "Product not found"}
    if row["inventory_count"] < quantity:
        return {"error": "Insufficient inventory"}
    unit_price = float(row["price"])
    return {
        "product_id": row["id"],
        "name": row["name"],
        "unit_price": unit_price,
        "currency": row["currency"],
        "quantity": quantity,
        "total": unit_price * quantity,
        "stripe_price_id": row["stripe_price_id"],
    }


if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

Note the price::text cast in the SELECT. asyncpg returns Decimal for NUMERIC columns, which fails JSON serialization. Casting to text and converting to float() in Python is the correct pattern.

Supabase — fastest to start.

Supabase exposes a REST API over Postgres via PostgREST. The supabase-py async client uses httpx internally — no blocking, asyncio-compatible. Best for getting started quickly with a full BaaS stack.

fastmcp>=3.2.0
supabase==2.30.0
uvicorn>=0.29.0

-- 1. Run the CREATE TABLE migration from Section 2 in the SQL Editor
-- 2. Enable Row Level Security (RLS) on the products table
-- 3. Create a policy allowing service_role to SELECT, UPDATE on products
-- 4. Copy: Settings → API → Project URL and service_role secret key

import os
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from supabase import acreate_client, AsyncClient
from fastmcp import FastMCP

SUPABASE_URL = os.environ["SUPABASE_URL"]          # https://abcdefgh.supabase.co
SUPABASE_KEY = os.environ["SUPABASE_SERVICE_KEY"]  # service_role key — never expose client-side

@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[dict]:
    client: AsyncClient = await acreate_client(SUPABASE_URL, SUPABASE_KEY)
    try:
        yield {"supabase": client}
    finally:
        pass  # AsyncClient has no explicit close


mcp = FastMCP("ProductServer", lifespan=lifespan)


@mcp.tool()
async def search_products(query: str, ctx) -> list[dict]:
    """Search active products by name or description."""
    client: AsyncClient = ctx.lifespan_context["supabase"]
    response = (
        await client.table("products")
        .select("id, name, slug, description, price, currency, inventory_count")
        .eq("active", True)
        .or_(f"name.ilike.%{query}%,description.ilike.%{query}%")
        .limit(20)
        .execute()
    )
    return response.data or []


@mcp.tool()
async def initiate_checkout(product_id: int, quantity: int, ctx) -> dict:
    """Retrieve product details for checkout and validate inventory."""
    client: AsyncClient = ctx.lifespan_context["supabase"]
    resp = (
        await client.table("products")
        .select("id, name, price, currency, stripe_price_id, inventory_count")
        .eq("id", product_id)
        .eq("active", True)
        .single()
        .execute()
    )
    if not resp.data:
        return {"error": "Product not found"}
    product = resp.data
    if product["inventory_count"] < quantity:
        return {"error": "Insufficient inventory"}
    unit_price = float(product["price"])
    return {
        "product_id": product["id"],
        "name": product["name"],
        "unit_price": unit_price,
        "currency": product["currency"],
        "quantity": quantity,
        "total": unit_price * quantity,
        "stripe_price_id": product["stripe_price_id"],
    }


if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

SQLAlchemy 2.x — when you need ORM.

Use SQLAlchemy 2.x async when you need Alembic migrations, ORM relationships, or a team convention that standardizes on SQLAlchemy. Both this and asyncpg use the same driver on the wire — the performance difference is negligible.

fastmcp>=3.2.0
SQLAlchemy[asyncio]==2.0.49
asyncpg==0.31.0
uvicorn>=0.29.0

import os
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from fastmcp import FastMCP

# Replace "postgresql://" with "postgresql+asyncpg://" for the async driver
DATABASE_URL = os.environ["DATABASE_URL"].replace(
    "postgresql://", "postgresql+asyncpg://", 1
)

@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[dict]:
    engine = create_async_engine(
        DATABASE_URL,
        pool_size=3,
        max_overflow=0,   # No overflow on serverless — keep connections predictable
        pool_timeout=10,
        pool_recycle=300,
        pool_pre_ping=True,  # Validates connection before each checkout
    )
    session_factory = async_sessionmaker(engine, expire_on_commit=False)
    try:
        yield {"engine": engine, "session_factory": session_factory}
    finally:
        await engine.dispose()


mcp = FastMCP("ProductServer", lifespan=lifespan)


@mcp.tool()
async def search_products(query: str, ctx) -> list[dict]:
    """Search products by name or description."""
    session_factory = ctx.lifespan_context["session_factory"]
    async with session_factory() as session:
        result = await session.execute(
            text(
                """
                SELECT id, name, slug, description, price::text, currency, inventory_count
                FROM products
                WHERE active = TRUE
                  AND (name ILIKE :q OR description ILIKE :q)
                ORDER BY name
                LIMIT 20
                """
            ),
            {"q": f"%{query}%"},
        )
        rows = result.mappings().all()
    return [dict(row) for row in rows]


if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

pool_pre_ping=True sends a lightweight SELECT 1 before reusing a connection from the pool. This catches dropped connections after a database restart or firewall timeout without surfacing errors to your tool callers.

Which database, which plan.

Vercel Postgres was sunset in December 2024 — all existing databases were migrated to Neon automatically. For new projects, use the Neon integration from the Vercel Marketplace. Re-verify all pricing before launch.

Practical guidance: Starting out → Supabase free tier (fastest setup, built-in pooler). Deploying on Vercel → Neon via Marketplace (DATABASE_URL injected automatically). Already on Railway → add Postgres as a Railway service.

Vercel dashboard to server.py.

Vercel injects environment variables directly into the function process at runtime. You don't need python-dotenv in production — os.environ["KEY"] works without any additional library.

Setting variables in Vercel

1. Open your project in the Vercel dashboard
2. Navigate to Settings → Environment Variables
3. Click Add New. Enter the key and value.
4. Check all three scopes: Production, Preview, Development
5. Click Save. Then redeploy — Vercel does not hot-reload env vars on running instances.

vercel env pull .env.local

import os

# Required vars — fail fast on startup if missing (raises KeyError, not silent None)
DATABASE_URL: str = os.environ["DATABASE_URL"]
SUPABASE_URL: str = os.environ["SUPABASE_URL"]
SUPABASE_SERVICE_KEY: str = os.environ["SUPABASE_SERVICE_KEY"]

# Optional var with fallback
DB_POOL_SIZE: int = int(os.environ.get("DB_POOL_SIZE", "3"))

Local development with python-dotenv

import os

# Only load .env locally — Vercel injects vars at runtime
if os.environ.get("VERCEL") is None:
    from dotenv import load_dotenv
    load_dotenv(".env.local")

DATABASE_URL: str = os.environ["DATABASE_URL"]

Add .env.local to .gitignore immediately. Never commit database credentials to version control.

50 function instances. One connection limit.

Vercel functions are ephemeral. Under moderate load, Vercel may spin up 10, 20, or 50 instances simultaneously — each opening its own pool. With max_size=10, that's 500 connections. Postgres's default max_connections is 100.

Solution A: Transaction-mode pooler (PgBouncer)

PgBouncer in transaction mode sits between your app and Postgres. It accepts thousands of connections and multiplexes them onto a small pool of real Postgres connections — released back at the end of each transaction, not the session.

Supabase: Built-in pooler on port 6543. Connect via the transaction-mode pooler URL instead of the direct connection (port 5432).

Neon: Built-in serverless pooler. Use the -pooler.neon.tech connection string from the Neon dashboard.

# Use the POOLER URL from Supabase: Settings → Database → Connection Pooling
# Note: port 6543, not 5432
DATABASE_POOLER_URL = os.environ["DATABASE_POOLER_URL"]
# Example: postgresql://postgres:PASSWORD@db.PROJECT_REF.supabase.co:6543/postgres

pool = await asyncpg.create_pool(
    DATABASE_POOLER_URL,
    min_size=1,
    max_size=3,          # Small per-instance pool; PgBouncer handles aggregation
    statement_cache_size=0,  # REQUIRED for PgBouncer transaction mode
)

Solution B: Minimal pool size (no pooler)

pool = await asyncpg.create_pool(
    dsn=os.environ["DATABASE_URL"],
    min_size=1,
    max_size=1,  # One connection per Vercel function instance
    command_timeout=15,  # Neon cold-start can take ~500ms; give it time
)

With max_size=1, 50 Vercel instances = 50 Postgres connections — manageable for most free tiers. This is the right choice when you can't use a pooler.

Every mistake, exact fix.

From static list to live database.

1. Create the products table

   Run the migration SQL from Section 2 against your database. For Supabase: paste into the SQL Editor and execute. For Neon/Railway: use psql DATABASE_URL -f migration.sql or the provider's query console. Verify that the pg_trgm extension was created and the trigram index on products.name exists before proceeding.

2. Set environment variables in Vercel

   In the Vercel dashboard, go to Settings → Environment Variables and add DATABASE_URL (or SUPABASE_URL and SUPABASE_SERVICE_KEY). Check all three scopes: Production, Preview, Development. Run vercel env pull .env.local to sync them locally. Add .env.local to .gitignore.

3. Install async packages and pin versions

   Add asyncpg==0.31.0 to requirements.txt. For Supabase: supabase==2.30.0. For SQLAlchemy: SQLAlchemy[asyncio]==2.0.49. Run pip install -r requirements.txt locally to confirm no dependency conflicts.

4. Rewrite server.py with lifespan and async tools

   Replace the static PRODUCTS list with a lifespan function that opens an asyncpg pool at startup and closes it on shutdown. Rewrite search_products and initiate_checkout as async def tools that acquire connections from ctx.lifespan_context["pool"]. Use the complete, runnable examples from Sections 3, 4, or 5 verbatim — substitute your actual column names.

5. Configure pooling and deploy

   For Supabase: switch to the pooler connection string (port 6543) and add statement_cache_size=0. Set max_size=3. Test locally with real data from your database, then deploy with vercel --prod. Run MCP Inspector against the production URL to confirm live queries return your seeded data.

Questions from the real connection.

import asyncio, asyncpg, os

SEED_DATA = [
    ("Blue Widget Pro", "blue-widget-pro", "Heavy-duty widget", 29.99, "USD", 100),
    ("Red Widget Lite", "red-widget-lite", "Lightweight widget", 9.99, "USD", 250),
]

async def seed():
    conn = await asyncpg.connect(os.environ["DATABASE_URL"])
    await conn.executemany(
        """INSERT INTO products (name, slug, description, price, currency, inventory_count)
           VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (slug) DO NOTHING""",
        SEED_DATA,
    )
    await conn.close()
    print(f"Seeded {len(SEED_DATA)} products.")

asyncio.run(seed())