# How to build an MCP server that queries a database

- Tool: MCP
- Difficulty: Advanced
- Time required: 30-45 min
- Compatibility: MCP TypeScript SDK 1.x, PostgreSQL 13+, pg/node-postgres
- Last updated: March 2026

## TL;DR

Build an MCP server that gives AI clients safe access to a PostgreSQL database. Use connection pooling with pg, expose tools for running parameterized queries, and register resources for schema inspection. Always use parameterized queries to prevent SQL injection and restrict to read-only operations unless write access is explicitly needed.

## Building an MCP Server with Database Access

Connecting an MCP server to a database lets AI clients query data, inspect schemas, and perform data operations through natural language. This is one of the most powerful MCP patterns — the AI can answer questions about your data without you writing custom queries.

However, database access requires careful security design. AI models can generate unexpected queries, so you need parameterized queries to prevent SQL injection, read-only restrictions to prevent accidental data modification, and connection pooling to prevent resource exhaustion. This tutorial builds a production-ready PostgreSQL MCP server step by step.

## Before you start

- PostgreSQL 13+ running locally or remotely
- MCP TypeScript SDK installed (@modelcontextprotocol/sdk)
- pg (node-postgres) package installed
- A database with tables and data to query
- Basic SQL knowledge

## Step-by-step guide

### 1. Set up connection pooling with pg

Create a PostgreSQL connection pool that manages connections efficiently. Connection pooling is critical for MCP servers because each tool call may execute a query, and creating a new connection per call would quickly exhaust database resources. Configure the pool with reasonable limits and load the connection string from an environment variable.

```
// TypeScript
import { Pool } from "pg";

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 10,                    // max connections in pool
  idleTimeoutMillis: 30000,   // close idle connections after 30s
  connectionTimeoutMillis: 5000, // fail if can't connect in 5s
});

// Test connection on startup
pool.query("SELECT 1").then(() => {
  console.error("Database connected");
}).catch((err) => {
  console.error("Database connection failed:", err.message);
  process.exit(1);
});
```

**Expected result:** A connection pool ready to handle concurrent queries from MCP tool calls.

### 2. Create a read-only query tool with parameterized queries

Register a tool that accepts SQL queries from the AI but restricts them to SELECT statements. Validate that the query starts with SELECT before executing. Always use parameterized queries when the tool accepts user-provided values to prevent SQL injection. Return results as formatted JSON.

```
// TypeScript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";

const server = new McpServer({ name: "postgres-server", version: "1.0.0" });

server.registerTool("query", {
  description: "Run a read-only SQL query against the database. Only SELECT statements are allowed.",
  inputSchema: {
    sql: z.string().describe("SQL SELECT query to execute"),
  },
}, async ({ sql }) => {
  const trimmed = sql.trim();

  // Enforce read-only
  if (!/^SELECT\b/i.test(trimmed)) {
    return {
      content: [{ type: "text", text: "Error: Only SELECT queries are allowed. Use INSERT, UPDATE, or DELETE tools for write operations." }],
      isError: true,
    };
  }

  // Block dangerous patterns
  if (/;\s*(DROP|DELETE|UPDATE|INSERT|ALTER|CREATE|TRUNCATE)/i.test(trimmed)) {
    return {
      content: [{ type: "text", text: "Error: Multi-statement queries with write operations are not allowed." }],
      isError: true,
    };
  }

  try {
    const result = await pool.query(trimmed);
    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          rowCount: result.rowCount,
          rows: result.rows.slice(0, 100), // Limit response size
        }, null, 2),
      }],
    };
  } catch (error) {
    console.error("Query error:", error);
    return {
      content: [{ type: "text", text: `Error: ${(error as Error).message}` }],
      isError: true,
    };
  }
});
```

**Expected result:** The AI can run SELECT queries and receive JSON-formatted results, but cannot execute write operations.

### 3. Expose database schema as a resource

Register an MCP resource that returns your database schema. This gives the AI the context it needs to write correct queries without you listing every table and column in tool descriptions. The schema resource should include table names, column names, types, and relationships.

```
// TypeScript
server.registerResource("database-schema", "db://schema", {
  description: "Complete database schema with tables, columns, and types",
  mimeType: "text/plain",
}, async (uri) => {
  const result = await pool.query(`
    SELECT
      t.table_name,
      c.column_name,
      c.data_type,
      c.is_nullable,
      c.column_default
    FROM information_schema.tables t
    JOIN information_schema.columns c
      ON t.table_name = c.table_name AND t.table_schema = c.table_schema
    WHERE t.table_schema = 'public'
    ORDER BY t.table_name, c.ordinal_position
  `);

  // Format as readable text
  const tables = new Map<string, string[]>();
  for (const row of result.rows) {
    const cols = tables.get(row.table_name) || [];
    cols.push(`  ${row.column_name} ${row.data_type}${row.is_nullable === 'NO' ? ' NOT NULL' : ''}${row.column_default ? ` DEFAULT ${row.column_default}` : ''}`);
    tables.set(row.table_name, cols);
  }

  const schema = Array.from(tables.entries())
    .map(([table, cols]) => `TABLE ${table}:\n${cols.join('\n')}`)
    .join('\n\n');

  return {
    contents: [{ uri: uri.href, text: schema }],
  };
});
```

**Expected result:** The AI reads the schema resource to understand your database structure before writing queries.

### 4. Add a table listing tool for quick discovery

A lightweight tool that lists all tables with row counts helps the AI quickly understand the database without reading the full schema. This is faster for initial exploration and costs less context than the full schema resource.

```
// TypeScript
server.registerTool("list-tables", {
  description: "List all tables in the database with row counts",
  inputSchema: {},
}, async () => {
  try {
    const result = await pool.query(`
      SELECT
        schemaname,
        relname AS table_name,
        n_live_tup AS approximate_row_count
      FROM pg_stat_user_tables
      ORDER BY n_live_tup DESC
    `);
    return {
      content: [{ type: "text", text: JSON.stringify(result.rows, null, 2) }],
    };
  } catch (error) {
    return {
      content: [{ type: "text", text: `Error: ${(error as Error).message}` }],
      isError: true,
    };
  }
});
```

**Expected result:** The AI gets a quick overview of all tables and their sizes before diving into specific queries.

### 5. Add a describe-table tool for column details

Give the AI a tool to inspect a specific table's columns, types, constraints, and indexes. This is more efficient than loading the entire schema when the AI only needs details about one table. For teams building database MCP servers for production analytics, RapidDev can help design the query layer with proper indexing and access control.

```
// TypeScript
server.registerTool("describe-table", {
  description: "Get detailed column information for a specific table",
  inputSchema: {
    tableName: z.string().regex(/^[a-zA-Z_][a-zA-Z0-9_]*$/).describe("Table name (alphanumeric and underscores only)"),
  },
}, async ({ tableName }) => {
  try {
    const columns = await pool.query(
      `SELECT column_name, data_type, is_nullable, column_default
       FROM information_schema.columns
       WHERE table_schema = 'public' AND table_name = $1
       ORDER BY ordinal_position`,
      [tableName]
    );

    if (columns.rowCount === 0) {
      return {
        content: [{ type: "text", text: `Error: Table '${tableName}' not found` }],
        isError: true,
      };
    }

    const indexes = await pool.query(
      `SELECT indexname, indexdef FROM pg_indexes
       WHERE schemaname = 'public' AND tablename = $1`,
      [tableName]
    );

    return {
      content: [{
        type: "text",
        text: JSON.stringify({ columns: columns.rows, indexes: indexes.rows }, null, 2),
      }],
    };
  } catch (error) {
    return {
      content: [{ type: "text", text: `Error: ${(error as Error).message}` }],
      isError: true,
    };
  }
});
```

**Expected result:** The AI gets column details and indexes for a specific table before constructing targeted queries.

## Complete code example

File: `src/index.ts`

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { Pool } from "pg";

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 10,
  idleTimeoutMillis: 30000,
});

const server = new McpServer({ name: "postgres-mcp", version: "1.0.0" });

// Resource: full schema
server.registerResource("schema", "db://schema", {
  description: "Database schema",
  mimeType: "text/plain",
}, async (uri) => {
  const r = await pool.query(
    `SELECT table_name, column_name, data_type, is_nullable
     FROM information_schema.columns WHERE table_schema='public'
     ORDER BY table_name, ordinal_position`
  );
  const tables = new Map<string, string[]>();
  for (const row of r.rows) {
    const c = tables.get(row.table_name) || [];
    c.push(`  ${row.column_name} ${row.data_type}${row.is_nullable==='NO'?' NOT NULL':''}`);
    tables.set(row.table_name, c);
  }
  const text = [...tables].map(([t,c])=>`TABLE ${t}:\n${c.join('\n')}`).join('\n\n');
  return { contents: [{ uri: uri.href, text }] };
});

// Tool: list tables
server.registerTool("list-tables", {
  description: "List all database tables with approximate row counts",
  inputSchema: {},
}, async () => {
  const r = await pool.query(
    `SELECT relname AS table_name, n_live_tup AS row_count
     FROM pg_stat_user_tables ORDER BY n_live_tup DESC`
  );
  return { content: [{ type: "text", text: JSON.stringify(r.rows, null, 2) }] };
});

// Tool: describe table
server.registerTool("describe-table", {
  description: "Get columns and indexes for a table",
  inputSchema: {
    table: z.string().regex(/^[a-zA-Z_]\w*$/).describe("Table name"),
  },
}, async ({ table }) => {
  const cols = await pool.query(
    `SELECT column_name,data_type,is_nullable,column_default
     FROM information_schema.columns
     WHERE table_schema='public' AND table_name=$1
     ORDER BY ordinal_position`, [table]
  );
  if (!cols.rowCount) return { content:[{type:"text",text:`Table '${table}' not found`}], isError:true };
  return { content: [{ type: "text", text: JSON.stringify(cols.rows, null, 2) }] };
});

// Tool: read-only query
server.registerTool("query", {
  description: "Run a read-only SELECT query",
  inputSchema: { sql: z.string().describe("SQL SELECT query") },
}, async ({ sql }) => {
  if (!/^SELECT\b/i.test(sql.trim())) {
    return { content:[{type:"text",text:"Error: Only SELECT allowed"}], isError:true };
  }
  try {
    const r = await pool.query(sql);
    return { content: [{ type: "text", text: JSON.stringify({
      rowCount: r.rowCount, rows: r.rows.slice(0, 100)
    }, null, 2) }] };
  } catch (e) {
    return { content:[{type:"text",text:`Error: ${(e as Error).message}`}], isError:true };
  }
});

const transport = new StdioServerTransport();
await server.connect(transport);
console.error("PostgreSQL MCP server running");
```

## Common mistakes

- **Not using connection pooling** — undefined Fix: Create a Pool instance and reuse it across all tool calls. Creating a new connection per query exhausts database resources quickly.
- **Allowing arbitrary SQL without validation** — undefined Fix: Validate that queries start with SELECT. Block multi-statement queries containing write operations. Consider using a read-only database user.
- **Not limiting result size** — undefined Fix: Slice results to a reasonable maximum (e.g., 100 rows). Large result sets overflow the AI's context window and waste tokens.
- **Hardcoding the database connection string** — undefined Fix: Load DATABASE_URL from environment variables. Never commit connection strings to source control.
- **Using string concatenation for query parameters** — undefined Fix: Always use parameterized queries ($1, $2) with the values array. String concatenation enables SQL injection.

## Best practices

- Use a dedicated read-only database user for the MCP server connection
- Always use parameterized queries ($1, $2) — never concatenate values into SQL strings
- Expose the database schema as a resource so the AI can write correct queries
- Limit query results to 100 rows to prevent context window overflow
- Set query timeouts (statement_timeout) to prevent runaway queries
- Log all queries to stderr for audit and debugging purposes
- Use connection pooling with reasonable limits (10-20 connections)
- Validate table names with regex when used in query identifiers

## Frequently asked questions

### Should I use a read-only database user for the MCP server?

Yes. Create a PostgreSQL user with SELECT-only grants and use that for the MCP connection. This provides defense in depth — even if SQL validation is bypassed, the database itself blocks writes.

### How do I prevent the AI from running expensive queries?

Set statement_timeout on the database connection (e.g., SET statement_timeout = '5000' for 5 seconds). Also limit result rows and block queries without WHERE clauses on large tables.

### Can I expose multiple databases from one MCP server?

Yes. Create multiple Pool instances and use tool names or parameters to route queries to the correct database. For example: query-analytics, query-users for different databases.

### How do I handle database migrations with an MCP server?

Run migrations through your normal workflow (Prisma, Drizzle, raw SQL). After schema changes, the AI will see the updated schema next time it reads the schema resource. No MCP server restart needed.

### Is it safe to let the AI run arbitrary SELECT queries?

With proper safeguards (read-only user, statement timeout, result limits), SELECT queries are safe. The main risk is performance — a complex JOIN on large tables can be slow. Use statement_timeout to prevent this.

### Can I use this pattern with MySQL or SQLite?

Yes. Replace the pg Pool with mysql2 or better-sqlite3. The MCP tool pattern is the same — validate SQL, execute with parameterized queries, return JSON results. For multi-database MCP architectures, the RapidDev team can help design the right abstraction layer.

---

Source: https://www.rapidevelopers.com/mcp-tutorial/how-to-build-mcp-server-with-database
© RapidDev — https://www.rapidevelopers.com/mcp-tutorial/how-to-build-mcp-server-with-database