Building safe MCP servers for a PostgreSQL database

Building safe MCP servers for your PostgreSQL database

Wednesday, August 12, 2026

Building safe MCP servers for your PostgreSQL database

Model Context Protocol (MCP) is an open protocol that describes how agents can connect to external tools and data sources, and is now widely supported by the most popular coding agents (like GitHub Copilot, Claude Code, and Codex) and agent frameworks (like LangChain and Pydantic AI). If you want to give agents a standard way to access the data in a database, you can build your own MCP server and expose tools for the agent to query or even modify data. But you need to design your MCP server carefully, to ensure that agents can do everything that users want - but nothing that you don't want them to do!

In this blog post, we'll walk through the range of ways to build MCP servers on top of a PostgreSQL database, since PostgreSQL is the most popular open source database and is production-ready with hosted offerings like Azure Database for PostgreSQL. You can apply these same principles to any database, however.

There's a spectrum of ways to build MCP servers on top of a database. We'll start with the most flexible option, exploratory servers that allow the agent to generate full SQL queries, conclude with the strictest option, fully typed tools for templated queries, and explore options in the middle too.

Free-form SQL

Let's take a look at a simple MCP server that gives the agent as much information and control as possible. For all of our examples, we use the Python language and the FastMCP package, but SDKs are available in multiple languages. All code is available in the GitHub repository.

We start off by giving the server a name, which the agent will see and consider when deciding which MCP server to invoke for a given user query:

mcp = FastMCP("Bees database MCP server")

For this example, my database stores observations of bees, so I name it accordingly.

We then define an execute_sql tool that accepts any SQL string, executes it against the database, and returns the rows.

@mcp.tool() async def execute_sql(sql: str) -> str: """Execute a SQL query against the database and return results.""" engine = await _get_engine() async with engine.connect() as conn: result = await conn.execute(text(sql)) if result.returns_rows: columns = list(result.keys()) rows = result.fetchall() return {"columns": columns, "rows": [[str(v) for v in row] for row in rows]} await conn.commit() return f"Statement executed. Rows affected: {result.rowcount}"

How will the agent know what SQL can be passed into that tool, however? We need to give it a way to discover the schema, so we also define a get_db_schema tool that dumps out the entire schema with table names, columns, and data types.

@mcp.tool() async def get_db_schema() -> str: """Return the database schema for all public tables.""" engine = await _get_engine() return await get_db_schema_text(engine)

We can test this MCP server out with a coding agent like GitHub Copilot. When we ask the agent "Which bees are active in El Cerrito in April?", the agent realizes that the Bees MCP server has relevant tools for the task, first calls get_db_schema, then calls execute_sql with a SELECT query. The database returns the results and the agent formats them into a Markdown table.

This MCP server works - we got the answer we wanted - but as you may have already noticed, there are multiple problems and risks to this approach.

Problem: Schema bloat

Let's tackle the problem with the get_db_schema tool first - it dumps everything! My observations database has only 5 tables and 60 columns, but a production database may have hundreds of tables and thousands of columns. Dumping the entire schema can confuse the LLM with irrelevant information, and unnecessarily fill up its context window.

What can we do instead? Progressive schema discovery. We provide two tools: list_tables that only returns table names, and describe_table that returns the columns only for the given table.

@mcp.tool() async def list_tables() -> str: """List all tables in the public schema. Call this first to discover available tables.""" async with engine.connect() as conn: result = await conn.execute(text( "SELECT table_name FROM information_schema.tables " "WHERE table_schema = 'public' AND table_type = 'BASE TABLE'")) return {"tables": [row[0] for row in result.fetchall()]}

@mcp.tool() async def describe_table(table_name: str) -> str: """Describe the columns of a specific table. Call list_tables() first to see available tables.""" async with engine.connect() as conn: result = await conn.execute(text( "SELECT column_name, data_type, is_nullable FROM information_schema.columns " "WHERE table_schema = 'public' AND table_name = :table_name "), {"table_name": table_name}) rows = result.fetchall() columns = [{"name": col, "type": dt, "nullable": n == "YES"} for col, dt, n in rows] return {"table": table_name, "columns": columns}

When we expose these tools to GitHub Copilot, the...

Building safe MCP servers for a PostgreSQL database

Related Articles

Incident with Github.com

(no title)

Scientists reverse brain aging, with a nasal spray

NHS apologises and admits Palantir have access to identifiable patient data

Cursor Origin