Database & Data Warehouse MCP Servers (Postgres, Snowflake)

Database access is one of the highest-value, and highest-risk, categories of enterprise MCP tool. The value is obvious: agents that can query real operational data are far more useful than ones limited to static documents. The risk is that a database tool wired up carelessly gives a model, or anything that can manipulate the model’s inputs, a path to unrestricted data access or destructive writes. This lesson also covers a detail teams frequently get wrong under real load: sizing the connection pool for the concurrency pattern agent traffic actually produces, which differs meaningfully from typical web traffic.

Never Build SQL by String Interpolation

# NEVER DO THIS: model-influenced input concatenated directly into SQL
async def bad_query_orders(filter_clause: str) -> list[dict]:
    query = f"SELECT * FROM orders WHERE {filter_clause}"  # SQL injection
    return await db.fetch(query)

# Correct: a small set of parameterized, purpose-built queries
class OrdersQueryInput(BaseModel):
    customer_id: str
    status: str | None = Field(None, description="One of: pending, shipped, delivered, cancelled")
    limit: int = Field(20, ge=1, le=100)

@mcp.tool(name="query_orders")
async def query_orders(input: OrdersQueryInput) -> list[dict]:
    """Query orders for a customer, optionally filtered by status. Read-only."""
    query = """
        SELECT order_id, status, total_amount, created_at
        FROM orders
        WHERE customer_id = $1
          AND ($2::text IS NULL OR status = $2)
        ORDER BY created_at DESC
        LIMIT $3
    """
    rows = await db_pool.fetch(query, input.customer_id, input.status, input.limit)
    return [dict(r) for r in rows]

Every value flows through the driver’s parameter binding ($1, $2, $3), never through string formatting, so there is no path for injected SQL regardless of what the model passes as arguments.

Defense in Depth: Read-Only Roles and Row Limits

-- Dedicated role for the MCP query tool, independent of the application's own DB user
CREATE ROLE mcp_agent_readonly WITH LOGIN PASSWORD '...';
GRANT CONNECT ON DATABASE analytics TO mcp_agent_readonly;
GRANT USAGE ON SCHEMA public TO mcp_agent_readonly;
GRANT SELECT ON orders, customers, order_items TO mcp_agent_readonly;
-- No INSERT/UPDATE/DELETE/DDL granted, at the database layer, not just in application code

# Row and cost guardrails enforced in the pool connection itself
pool = await asyncpg.create_pool(
    dsn=os.environ["MCP_READONLY_DATABASE_URL"],  # connects as mcp_agent_readonly
    min_size=2,
    max_size=10,
    command_timeout=10,  # Kill runaway queries after 10s regardless of LIMIT clauses
)

Sizing the Connection Pool for Agent Traffic Concurrency

A connection pool sized purely from average requests-per-minute, the way it might be for a traditional web backend, tends to under-provision for agent traffic, and it is worth understanding why explicitly rather than discovering it as an intermittent production issue. A single agent turn often triggers several tool calls concurrently within that one turn (Lesson 13’s get_customer_profile tool, for instance, awaits three underlying calls with asyncio.gather), and several independent agent sessions can each be mid-turn at the same moment. The result is that peak concurrent connection demand can spike well above what average throughput alone would suggest, even when total query volume over a longer window looks entirely reasonable.

flowchart TD
    Turn1["Agent session A, one turn"] --> Q1["3 concurrent queries\n(asyncio.gather)"]
    Turn2["Agent session B, one turn"] --> Q2["2 concurrent queries"]
    Turn3["Agent session C, one turn"] --> Q3["4 concurrent queries"]
    Q1 --> Pool["Connection Pool"]
    Q2 --> Pool
    Q3 --> Pool
    Pool -->|"peak concurrent demand: 9\nfar above what avg req/min suggests"| DB[("Database")]

    style Turn1 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Turn2 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Turn3 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Pool fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style DB fill:#f0fdf9,stroke:#0D9488,color:#0F172A

The practical fix is to size max_size from an estimate of peak concurrent connections per in-flight agent turn multiplied by the expected number of simultaneously active sessions, with headroom, rather than from average throughput, and to monitor actual pool exhaustion events (a metric most connection pool libraries expose, e.g. pool.get_size() versus pool.get_max_size() sampled periodically) so the estimate can be corrected against real production behavior instead of staying a one-time guess.

# Expose pool saturation as a metric, not just a silent wait-for-connection delay
async def report_pool_saturation():
    while True:
        in_use = pool.get_size() - pool.get_idle_size()
        saturation_pct = in_use / pool.get_max_size()
        metrics.gauge("mcp.db_pool.saturation", saturation_pct)
        if saturation_pct > 0.9:
            logging.warning(f"DB pool at {saturation_pct:.0%} capacity, consider raising max_size")
        await asyncio.sleep(15)

Natural-Language-to-SQL for Data Warehouses

For analytical tools (Snowflake, BigQuery) where the query shape genuinely needs to vary, a common pattern lets the model draft SQL against a documented, restricted schema, but the tool validates the generated SQL before execution rather than trusting it outright.

flowchart TD
    Model["Model drafts SQL\n(against documented schema/views only)"] --> Validate{"Static validation:\nSELECT-only?\nallowed tables only?\nno DDL/DML keywords?"}
    Validate -->|fails| Reject["Rejected, structured error\nreturned to model"]
    Validate -->|passes| Explain["EXPLAIN query first\n(estimate cost/rows)"]
    Explain --> CostCheck{"Estimated cost\nwithin budget?"}
    CostCheck -->|no| RejectCost["Rejected: query too expensive,\nask model to narrow scope"]
    CostCheck -->|yes| Run["Execute against read-only role,\nrow-limited"]

    style Model fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Validate fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style Explain fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style CostCheck fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style Run fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style Reject fill:#fef2f2,stroke:#dc2626,color:#0F172A
    style RejectCost fill:#fef2f2,stroke:#dc2626,color:#0F172A

FORBIDDEN_KEYWORDS = {"insert", "update", "delete", "drop", "alter", "truncate", "grant", "create"}

@mcp.tool(name="query_warehouse")
async def query_warehouse(sql: str) -> dict:
    """Execute a read-only analytical query against approved warehouse views."""
    lowered = sql.lower()
    if any(kw in lowered for kw in FORBIDDEN_KEYWORDS):
        return {"isError": True, "message": "Only SELECT queries against approved views are permitted"}

    plan = await warehouse_client.explain(sql)
    if plan.estimated_bytes_scanned > MAX_BYTES_SCANNED:
        return {"isError": True, "message": "Query would scan too much data, narrow the date range or filters"}

    result = await warehouse_client.execute(sql, row_limit=1000)
    return {"rows": result.rows, "row_count": len(result.rows)}

Keyword blocklisting is a coarse first filter, not a complete defense on its own, a sufficiently motivated adversarial input could in principle construct SQL that evades a naive substring check (nested comments, unusual whitespace, encoded keywords). The EXPLAIN-then-execute step matters as a second, independent layer precisely because it evaluates the query’s actual planned behavior rather than trusting a surface-level text scan, and executing exclusively against the read-only database role from earlier in this lesson is the third and most important layer, since it makes destructive operations structurally impossible regardless of whether the first two checks are bypassed.

The next lesson extends this integration pattern to third-party SaaS systems: Salesforce, Slack, Jira, and GitHub, each with its own authentication quirks and rate limits.