Connecting Agents to Databases (PostgreSQL & Vector Search)

An analytics agent goes live on a Monday. By Tuesday afternoon the on-call engineer is looking at FATAL: sorry, too many clients already in the Postgres log, and the web application — which has nothing to do with the agent — is down.

The agent's MCP server opens a fresh connection at the top of every tool call and closes it at the bottom. That looks harmless in a unit test. In production, a single research question triggers around forty tool calls, three analysts are using it, and the retry logic opens a new connection for each attempt. Postgres max_connections is the default 100. Each backend process costs roughly 5–10 MB of server memory before it does any work, and each connection costs 30–50 ms to establish once you count the TCP handshake, the TLS negotiation and authentication.

Price the waste. Forty calls at 40 ms of setup is 1.6 seconds per question spent on connection handshakes alone, against roughly 0.2 ms to check a connection out of a pool — a factor of two hundred. And that is the benign failure. The real one is that the agent's churn starved the connection slots the application needed.

A database MCP server is not "a wrapper around a SQL client". It is a piece of production infrastructure sitting between a non-deterministic caller and your most valuable data, and it has to be built like one.

Where the server sits

   agent host                 database MCP server            PostgreSQL  +-----------+            +---------------------+        +--------------+  |  model    |    MCP     |  pool               |  pool  | read replica |  |  loop     |<---------->|  validate           |<------>| + pgvector   |  |           |  JSON-RPC  |  bound + summarise  |        |              |  +-----------+            +---------------------+        +--------------+        ^                            ^        |                            |   sees summaries,             holds credentials,   never raw rows              enforces every limit

Everything the model is allowed to know about the database passes through this process, which makes it the only sensible place to enforce limits. The agent cannot be trusted to cap its own row counts; it is a language model, and "return only what you need" is a suggestion it will sometimes ignore.

Connection pooling

A pool holds a small set of already-authenticated connections and lends them out. The mechanics matter because the sizing rule is counter-intuitive: a bigger pool is usually slower. Beyond the number of cores and spindles the database can actually work in parallel, extra connections add context switching and lock contention, not throughput.

1import asyncpg2from contextlib import asynccontextmanager34POOL: asyncpg.Pool | None = None56async def init_pool() -> None:7    global POOL8    POOL = await asyncpg.create_pool(9        dsn="postgresql://agent_ro@db.internal:5432/analytics",10        min_size=2,             # keep a couple warm so the first call is fast11        max_size=10,            # hard ceiling on this server's footprint12        max_inactive_connection_lifetime=300,   # recycle idle connections13        command_timeout=30,     # client-side ceiling on any single statement14        server_settings={15            "application_name": "mcp-analytics",   # shows up in pg_stat_activity16            "statement_timeout": "10000",          # 10 s, enforced by the server17            "idle_in_transaction_session_timeout": "15000",18        },19    )2021@asynccontextmanager22async def acquire():23    async with POOL.acquire() as conn:24        yield conn

Sizing with numbers: if the median query takes 20 ms, one connection sustains 50 queries per second. A pool of 10 sustains 500 per second — vastly more than any interactive agent generates. If you find yourself wanting max_size=50, the queries are slow, not the pool small.

Pool size is a limit on how much damage a runaway agent can do to your database, not a performance dial. Choose it as a budget.

Executing SQL safely

The injection problem, concretely

Suppose a tool accepts a customer email and the handler builds a query by formatting a string:

sql = f"select id, total_cents from orders where email = '{email}'"

Now the model — persuaded by a poisoned document it read earlier, or simply confused — passes:

email = "x' union select id, 0 from api_keys --"

The database receives one statement that reads a table nobody meant to expose. Parameterisation removes this completely:

rows = await conn.fetch("select id, total_cents from orders where email = $1", email)

The driver sends the query text and the value over separate protocol fields. The value is never parsed as SQL, so the entire string — quotes, union, comment marker and all — is treated as one improbable email address that matches zero rows.

What parameterisation does not protect against

Three gaps survive, and they are where careful teams still get hurt.

1. Identifiers cannot be parameters. Table names, column names and sort directions are part of the query's structure. order by $1 does not do what you want; the database treats it as a constant, silently sorting by nothing.
2. Cost is not injection but is still an attack. select * from events a, events b is perfectly valid, parameter-free SQL that produces a cross join of a billion rows.
3. Reading everything is a leak even when it is authorised. A syntactically innocent select * from users exfiltrates a table into a model's context.

1import re2import sqlglot3from sqlglot import exp45IDENT = re.compile(r"^[a-z_][a-z0-9_]{0,62}$")6ALLOWED_TABLES = {"orders", "customers", "order_items", "products"}78def validate_select(sql: str) -> str:9    """Parse the statement and refuse anything that is not a bounded read."""10    try:11        parsed = sqlglot.parse(sql, read="postgres")12    except Exception as e:13        raise QueryRejected(f"could not parse SQL: {e}")1415    if len(parsed) != 1:16        raise QueryRejected("exactly one statement is allowed")17    stmt = parsed[0]18    if not isinstance(stmt, exp.Select):19        raise QueryRejected(f"only SELECT is allowed, got {type(stmt).__name__}")2021    for table in stmt.find_all(exp.Table):22        if table.name not in ALLOWED_TABLES:23            raise QueryRejected(f"table not allowed: {table.name}")2425    if stmt.find(exp.Star) and not stmt.args.get("limit"):26        raise QueryRejected("SELECT * requires an explicit LIMIT")27    return sql

Parsing beats regexes here, and the reason is worth internalising. A blocklist of the words drop, delete and update is defeated by a column literally named update_count, by comments (DR/**/OP), and by case tricks. A parser understands the statement's structure, so "is this a SELECT?" is answered by the shape of the tree rather than by string matching.

Add two more gates below the parser. Run the server's connection as a role with SELECT and nothing else, so that a parser bypass still cannot write. And gate expensive plans with EXPLAIN before executing:

1import json23MAX_PLAN_COST = 500_000.045async def guard_cost(conn, sql: str, args: list) -> None:6    # asyncpg returns a json column as a str unless you register a codec.7    plan = json.loads(await conn.fetchval(f"explain (format json) {sql}", *args))8    cost = plan[0]["Plan"]["Total Cost"]9    if cost > MAX_PLAN_COST:10        raise QueryRejected(11            f"estimated cost {cost:,.0f} exceeds the limit of {MAX_PLAN_COST:,.0f}. "12            f"Add a WHERE clause on an indexed column, or narrow the date range.")

Note the wording of that rejection. It tells the model what to change. "Query too expensive" produces an identical retry; naming the fix produces a better query on the next turn.

Bounding what comes back

Even a legitimate query can destroy a conversation. A result of 4,300 rows across 12 columns serialises to roughly 90,000 tokens — nearly half a 200k context window, for data the model cannot reason over anyway. Cap in three places:

1MAX_ROWS, MAX_CHARS = 200, 40_00023async def run_read(conn, sql: str, args: list) -> str:4    await guard_cost(conn, sql, args)5    rows = await conn.fetch(f"select * from ({sql}) sub limit {MAX_ROWS + 1}", *args)6    truncated = len(rows) > MAX_ROWS7    rows = rows[:MAX_ROWS]89    header = " | ".join(rows[0].keys()) if rows else "(no columns)"10    body = "\n".join(" | ".join(str(v)[:120] for v in r.values()) for r in rows)11    out = f"{header}\n{body}"12    if len(out) > MAX_CHARS:13        out = out[:MAX_CHARS] + "\n... output truncated at 40,000 characters ..."14    if truncated:15        out += (f"\n\nShowing the first {MAX_ROWS} rows; more matched. "16                f"Add a WHERE clause or use an aggregate to narrow the result.")17    return out

Fetching MAX_ROWS + 1 is the small trick that lets you say "there were more" without counting them. Delimited text rather than JSON is worth about a 40% token saving on wide results, because JSON repeats every key on every row.

Schema introspection

An agent writing SQL against a schema it has not seen will invent column names. Exposing the schema is the single highest-value thing a database server does, and it belongs as a resource — it is a read with no side effects, and the host may want to load it into context proactively.

1@mcp.resource("db://analytics/schema/{table}")2async def table_schema(table: str) -> str:3    if not IDENT.match(table) or table not in ALLOWED_TABLES:4        raise ValueError(f"unknown table: {table}")5    async with acquire() as conn:6        cols = await conn.fetch("""7            select c.column_name, c.data_type, c.is_nullable, c.column_default,8                   col_description(('public.' || c.table_name)::regclass,9                                   c.ordinal_position) as comment10              from information_schema.columns c11             where c.table_schema = 'public' and c.table_name = $112             order by c.ordinal_position""", table)13        fks = await conn.fetch("""14            select kcu.column_name, ccu.table_name as ref_table,15                   ccu.column_name as ref_column16              from information_schema.table_constraints tc17              join information_schema.key_column_usage kcu18                on kcu.constraint_name = tc.constraint_name19              join information_schema.constraint_column_usage ccu20                on ccu.constraint_name = tc.constraint_name21             where tc.table_name = $1 and tc.constraint_type = 'FOREIGN KEY'""", table)22        n = await conn.fetchval(23            "select reltuples::bigint from pg_class where relname = $1", table)2425    lines = [f"TABLE public.{table}  (~{n:,} rows)"]26    for c in cols:27        null = "" if c["is_nullable"] == "YES" else " NOT NULL"28        note = f"  -- {c['comment']}" if c["comment"] else ""29        lines.append(f"  {c['column_name']}: {c['data_type']}{null}{note}")30    for f in fks:31        lines.append(f"  FK {f['column_name']} -> {f['ref_table']}.{f['ref_column']}")32    return "\n".join(lines)

Include the column comments and the foreign keys. A model that can see FK customer_id -> customers.id writes the join correctly on the first attempt; one that cannot will guess at customers.customer_id and fail. The approximate row count from pg_class.reltuples costs nothing and tells the model whether a full scan is reasonable.

Semantic search with pgvector

Exact matching fails on the queries agents actually receive. "Why do customers cancel?" does not contain the word "churn", and where reason ilike '%cancel%' misses every ticket that said "not renewing". Vector search compares meaning by embedding text as a point in high-dimensional space and finding nearby points.

1create extension if not exists vector;23create table doc_chunks (4  id          bigserial primary key,5  doc_id      text        not null,6  chunk_index int         not null,7  content     text        not null,8  metadata    jsonb       not null default '{}',9  embedding   vector(1536) not null,10  created_at  timestamptz not null default now()11);1213-- Approximate-nearest-neighbour index. Cosine distance must match the14-- operator used at query time, or the index will simply be ignored.15create index doc_chunks_hnsw16    on doc_chunks using hnsw (embedding vector_cosine_ops)17    with (m = 16, ef_construction = 64);1819create index doc_chunks_meta on doc_chunks using gin (metadata);

Why the index is not optional: a 1536-dimension float32 vector occupies 1536 × 4 = 6,144 bytes. Two million chunks is 12.3 GB that an exact scan must read and compare on every query — seconds, not milliseconds. HNSW builds a navigable graph and answers in single-digit milliseconds at roughly 95% recall. You trade a small chance of missing a result for a thousandfold speed-up, and for retrieval feeding a model that trade is nearly always correct.

1async def semantic_search(query: str, k: int = 5, doc_type: str | None = None) -> str:2    vec = await embed(query)                    # 1536 floats from your embedding model3    literal = "[" + ",".join(f"{x:.6f}" for x in vec) + "]"4    async with acquire() as conn, conn.transaction():   # SET LOCAL needs a transaction5        await conn.execute("set local hnsw.ef_search = 64")   # recall/latency dial6        rows = await conn.fetch("""7            select doc_id, chunk_index, content, metadata,8                   1 - (embedding <=> $1::vector) as similarity9              from doc_chunks10             where ($3::text is null or metadata->>'type' = $3)11             order by embedding <=> $1::vector12             limit $2""", literal, k, doc_type)13    return "\n\n".join(14        f"[{r['doc_id']}#{r['chunk_index']}  sim={r['similarity']:.3f}]\n"15        f"{r['content'][:1200]}" for r in rows)

Three operators exist and mixing them up silently ruins results: <=> is cosine distance, <-> is Euclidean, <#> is negative inner product. Cosine distance runs from 0 (identical direction) to 2 (opposite), so similarity is 1 - distance. The order by must use the same operator class the index was built with, or Postgres quietly falls back to a sequential scan and your fast search becomes a slow one with no error to tell you.

Hybrid search, and why you want it

Vector search is weak exactly where keyword search is strong: exact identifiers. A query for ord_8812 has no useful semantics. Run both and fuse the rankings with Reciprocal Rank Fusion, which scores each document as the sum of 1 / (60 + rank) across the lists it appears in.

Concretely, with two candidate chunks:

Chunk A wins the vector list outright, but B is respectable in both and takes the fused top spot. That is the behaviour you want: agreement across two independent signals beats a single strong opinion. The constant 60 damps the influence of rank-1 positions, which is what stops one list from dominating.

Resources or tools: which is which

The rule of thumb: if the host could sensibly load it before the user says anything, it is a resource. If it needs an argument the model invents, it is a tool.

A model can compose any query it can imagine, so every limit that protects the database has to live in the server. Nothing above it is bounded.

Transactions and writes

The tempting design is begin_transaction, execute, commit as three separate tools. Do not build it. A model that opens a transaction and then decides to ask a clarifying question leaves a connection idle in transaction, holding locks, blocking VACUUM, and eventually blocking DDL for the whole table. There is no timeout on the model's attention span.

Expose atomic units instead — one tool, one transaction, committed or rolled back before the call returns:

1@mcp.tool()2async def record_refund(order_id: str, amount_cents: int, reason: str,3                        idempotency_key: str) -> str:4    """Record a refund against an order. Atomic. Safe to retry with the same key."""5    if not (0 < amount_cents <= 1_000_00):6        return f"amount_cents must be between 1 and 100000, got {amount_cents}"7    async with acquire() as conn:8        async with conn.transaction():9            prior = await conn.fetchval(10                "select refund_id from refunds where idempotency_key = $1",11                idempotency_key)12            if prior:13                return f"Already recorded as {prior} (idempotent replay)."14            order = await conn.fetchrow(15                "select total_cents, refunded_cents from orders "16                "where id = $1 for update", order_id)17            if order is None:18                return f"No such order: {order_id}"19            remaining = order["total_cents"] - order["refunded_cents"]20            if amount_cents > remaining:21                return (f"Refund of {amount_cents} exceeds the remaining "22                        f"refundable balance of {remaining} cents on {order_id}.")23            refund_id = await conn.fetchval(24                "insert into refunds (order_id, amount_cents, reason, idempotency_key) "25                "values ($1,$2,$3,$4) returning refund_id",26                order_id, amount_cents, reason, idempotency_key)27            await conn.execute(28                "update orders set refunded_cents = refunded_cents + $1 where id = $2",29                amount_cents, order_id)30    return f"Recorded refund {refund_id} of {amount_cents} cents against {order_id}."

The idempotency key is what makes retries safe. Agents retry — after a timeout, after a truncated response, after the model forgets it already called the tool. Without a key, a network blip during a refund produces two refunds. With one, the second call returns the first result and changes nothing. The for update lock closes the concurrent-double-refund race that the key alone would not.

Notice too that every failure returns a sentence, not an exception. "Refund of 5000 exceeds the remaining refundable balance of 1200 cents on ord_8812" lets the agent explain the situation to the user. A stack trace does not.

The moment a transaction spans two tool calls, its lifetime is decided by a language model — and a language model has no timeout on its attention.

What this means when you build one

The design pressure on a database MCP server all points the same way: assume the caller is well-intentioned, occasionally confused, and never bound by your assumptions.

That produces a short list of decisions to make before writing a handler. Pick the database role first — a role with only SELECT on four tables makes an entire class of bugs unreachable, and it takes ten minutes to create. Put a number on every unbounded thing: rows, characters, plan cost, statement time, pool size, transaction lifetime. Each of those is a place where an agent will otherwise ask for infinity. Return prose, not exceptions, because every error message is really a prompt telling the model what to do next. And keep transactions inside a single tool call, because the moment a transaction spans two calls, its lifetime is controlled by a language model.

The team from the opening changed four things: a pool with max_size=8, a read-only role, a 200-row cap, and a 10-second statement_timeout. Peak connection count from the agent went from 60-plus to 8, the median tool call dropped from 340 ms to 22 ms because the handshake disappeared, and the application stopped falling over. None of it made the agent less capable. It made the agent survivable.