Building MCP servers for your database: Flexibility, safety, and tradeoffs

The Model Context Protocol (MCP) is an open standard that outlines how various agents can connect with external tools and data sources. It’s now well-supported by popular coding agents such as GitHub Copilot, Claude Code, and Codex, as well as agent frameworks like LangChain and Pydantic AI.

If you’re looking to provide agents with a consistent way to access data from a database, you can create your own MCP server. This will allow the agent to both query and modify data, but it’s essential to design your MCP server thoughtfully. This way, you can ensure that agents can perform all the functions users may want—while restricting access to actions you’d prefer they avoid.

In this article, we’ll explore various methods to establish MCP servers using a PostgreSQL database. PostgreSQL stands out as the leading open-source database and is fully operational with hosted solutions like Azure Database for PostgreSQL. These techniques can be applied to any database, however.

We’ll begin with the most flexible option: exploratory servers that allow the agent to create full SQL queries. We’ll conclude with the most restrictive option, which offers fully typed tools for templated queries, while also examining intermediate options.

Let’s dive into a straightforward MCP server that offers the agent comprehensive control and information. In our examples, we will use Python along with the FastMCP package, although SDKs are available in various programming languages. All of the code can be found in the GitHub repository.

First, we need to assign a name to the server, which the agent will reference when determining which MCP server to use for a given user inquiry:

mcp = FastMCP("Bees Database MCP Server")

For this demonstration, my database tracks observations of bees, hence the fitting name.

Next, we define an execute_sql tool that can accept any SQL string, executes it on the database, and returns the results:

@mcp.tool()
async def execute_sql(sql: str) -> str:
  """Runs a SQL query against the database and returns 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}"

But how does the agent know what SQL can be passed into this tool? We need to allow the agent to discover the schema, so we will also create a get_db_schema tool that reveals the entire schema, including table names, columns, and data types:

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

We can put this MCP server to the test using a coding agent like GitHub Copilot. If we ask the agent, “Which bees are active in El Cerrito during April?”, it first recognizes that the Bees MCP server has relevant tools to resolve the task. It calls get_db_schema first, followed by execute_sql with a SELECT query. The database will provide the answers, and the agent formats these into a Markdown table:

This MCP server functions well—we received the response we were looking for—but several issues have arisen with this method.

Let’s address the concern with the get_db_schema tool first—it returns everything! While my observations database has only 5 tables and 60 columns, a production database could contain hundreds of tables and thousands of columns. Revealing the entire schema can overwhelm the LLM with irrelevant data and clutter its context window.

So, what’s the alternative? We could implement progressive schema discovery. We create two additional tools: one called list_tables that simply returns table names and another called describe_table that gives the columns for a specific table:

@mcp.tool()
async def list_tables() -> str:
  """Lists 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:
  """Describes 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 provide these tools to GitHub Copilot, the agent first calls list_tables, then makes multiple calls to describe_table, one for each relevant table:

Although this design requires three tool calls for schema discovery, rather than just one as before, it prevents unnecessary context clutter for larger databases. You can weigh the benefits of this trade-off against the size of your schema.

Next, let’s address the potential risks. The execute_sql function can run any valid SQL, including operations that update or delete data. If a user requests “How many bee observations are marked as ‘needs_id’? I might need to delete,” it could lead to the accidental deletion of thousands of rows with one command. If this level of access is acceptable, that’s fine; however, in many scenarios, you’ll probably want to prohibit modifications altogether or at least ask for confirmation first.

We can start by creating a read-only version of the SQL execution tool. The execute_readonly_sql function below includes several safety measures, including a check to ensure that the SQL includes only SELECT statements, a time limit of 30 seconds to avoid costly queries, and a cap of 100 rows:

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True), timeout=30.0)
async def execute_readonly_sql(sql: str) -> dict:
  """Executes a read-only SQL query against the database.
  Only SELECT statements are allowed. Non-SELECT statements are rejected.
  Results are capped at 100 rows."""
  try:
    validated_sql = validate_readonly_sql(sql)
  except ValueError as e:
    raise ToolError(str(e))

  async with engine.connect() as conn:
    result = await conn.execute(text(validated_sql))
    columns = list(result.keys())
    rows = result.fetchmany(MAX_LIMIT)  # Cap rows beyond the limit
    return {"columns": columns, "rows": [[str(v) for v in row] for v in rows]}

This tool is marked with readOnlyHint=True, which is part of the allowed annotations from the MCP specification. This read-only hint informs the MCP client that this tool does not modify data, potentially influencing how the client handles tool rendering or approvals. However, keep in mind that this is merely a hint—not a strict rule. It’s up to the server developer to enforce true read-only functionality within the tool’s logic.

The aim of validate_readonly_sql is to ensure that the provided SQL string is strictly a SELECT statement. In Python, I achieved this verification using the pglast library, which helps parse the Abstract Syntax Tree (AST) of the SQL string, confirming it has only one statement that is specifically a SELECT statement:

def validate_readonly_sql(sql: str) -> str:
  try:
    stmts = pglast.parse_sql(sql)
  except pglast.parser.ParseError as e:
    raise ValueError(f"SQL parse error: {e}")

  if len(stmts) != 1:
    raise ValueError("Only one statement is allowed")
  if (stmt_type := type(stmts[0].stmt).__name__) != "SelectStmt":
    raise ValueError(f"Only SELECT statements are allowed, got {stmt_type}")
  return sql

This will prevent most destructive SQL commands like:

We’re not out of the woods yet! Some potentially dangerous SQL commands can still bypass these checks. While we could extend the AST-based parsing to block those, PostgreSQL provides an even more robust solution: read-only enforcement at the database level.

When we connect to the database, we can run the following command to ensure that only read-only transactions are allowed:

SET default_transaction_read_only = ON

This setting will block these CTEs that start with WITH and attempt to hide mutations inside:

Additionally, we can create a dedicated PostgreSQL role specifically for the MCP server, granting it permission solely to execute SELECT queries on a specified schema:

CREATE ROLE mcp_readonly;
GRANT CONNECT ON DATABASE bees TO mcp_readonly;
GRANT USAGE ON SCHEMA public TO mcp_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_readonly;

This role can also block SELECT statements that execute potentially harmful built-in SQL functions, such as:

• SELECT pg_terminate_backend(pid)
• SELECT pg_read_file('/etc/passwd')
• SELECT pg_reload_conf()

You could opt to enforce read-only access solely through the least-privilege role, but this risks exposing your server if the role is not set correctly. Employing multiple layers of protection is recommended for added security.

Is it conceivable that a malicious user or an errant agent might still slip through a harmful query? For those seeking complete assurance, the best approach is to refrain from exposing SQL entirely.

Instead, we can define tools tailored to common user requirements, allowing them to input values that get safely integrated into templated SQL queries—or through ORM calls.

For instance, the search_species tool below accepts a search term and a limit, executing a templated SQL query on a predefined table:

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
async def search_species(q: str, limit: int = 10) -> list[SpeciesResults]:
  """Search for bee species by their scientific or common names.
  Utilize this to resolve a name to a taxon_id before invoking other tools."""
  sql = text("""
    SELECT taxon_id, scientific_name, common_name, family, genus FROM species
    WHERE to_tsvector('simple',
        coalesce(scientific_name, '') || ' ' || coalesce(common_name, ''))
        @@ plainto_tsquery('simple', :q)
    ORDER BY scientific_name ASC LIMIT :limit""")
  async with engine.connect() as conn:
    result = await conn.execute(sql, {"q": q, "limit": min(limit, 50)})
    return [SpeciesResult(...) for row in result.fetchall()]

We need to create additional tools for every SQL query that may be required to address user inquiries, like a search_observations tool that takes parameters like latitude, longitude, date, and species.

<pWhen we provide GitHub Copilot with these tools and pose the question, "Are there any carpenter bees located around Berkeley?", the agent will first call search_species with the request for “carpenter bee,” retrieving the names and scientific metadata for the corresponding bees. Next, it would execute search_observations with the geographic coordinates for Berkeley:

The main benefit of this approach is that the agent never directly composes the SQL statements, which prevents it from inadvertently executing any destructive, expensive, or slow queries.

However, there’s a significant downside: the agent can only respond to the subset of user questions you anticipate. If you decide to adopt this approach, consider ways to monitor unanswered user queries, possibly by introducing a give_feedback tool on the server to support feature requests.

If you’re developing an MCP server primarily for administration purposes (as opposed to data analysis and exploration), you might indeed want to permit deletion—but with caution. In a database admin interface, a delete action is typically highlighted with a bright red button and often prompts for confirmation before proceeding:

We can create a similar user interface for our MCP server, thanks to form-based elicitation, a recent update to the MCP specification. In compatible MCP clients, this feature will trigger a form with your desired question and options. From there, you can adjust what the tool does based on user selection.

For instance, the delete_observation tool uses elicitation to confirm that the user genuinely wishes to delete the record found in the database:

@mcp.tool(annotations=ToolAnnotations(destructiveHint=True))
async def delete_observation(ctx: Context, observation_id: int) -> str:
  """Delete a bee observation."""
  row = ... # look up the record
  result = await ctx.elicit(
    f"Permanently delete observation #{row.observation_id}?\n"
    f".  {row.scientific_name} on {row.observed_data}\n",
    response_type=["yes, delete it", "no, keep it"])
  if result.action == "cancel" or result.data == "no, keep it":
    return "Deletion cancelled."
  await session.execute(
    text("DELETE FROM observations WHERE observation_id = :oid"),
    {"oid": observation_id}
  )
  await session.commit()
  return f"Deleted observation #{observation_id}"

When we ask GitHub Copilot to delete an observation, it invokes the delete_observation tool, prompting a confirmation dialog. The user must explicitly click to confirm the deletion:

Elicitation can also be beneficial beyond destructive operations. You can use it to resolve ambiguities in user queries (“Did you mean…?”) or suggest alternative queries when a request might be too resource-intensive (for instance, narrowing a broad 200 km search radius to 50 km).

We’ve examined a variety of methods for exposing your database as an MCP server:

Free-form SQL works well for internal prototypes where you need flexibility. Read-only SQL is well-suited for data analysis tasks that require independent investigation. Templated queries are the safest choice for production and user-interactive scenarios. Regardless of your method, always enforce database-level permissions to minimise risk.

Creating MCP servers for your database is an excellent way to enable users to engage with data through natural language, but it’s crucial to design your tools with safety in mind.

For more information, feel free to check out the complete source code on GitHub, which showcases four MCP servers demonstrating the techniques we explored, and can be executed either locally or on Azure.