Avanti Fellows PostgreSQL MCP Server

Instructions

Only SELECT queries are allowed. Use this to explore data,
debug issues, or validate assumptions about the data.

Args:
    sql: A SELECT query to execute

Returns:
    JSON array of results, or error message

Implementation Reference

• mcp_postgres/server.py:47-73 (handler)

  The primary handler function for the 'query' MCP tool. It validates the SQL as read-only, executes it using asyncpg, serializes results to JSON, and handles errors.

  @mcp.tool()
  async def query(sql: str) -> str:
      """Execute a read-only SQL query against the database.
  
      Only SELECT queries are allowed. Use this to explore data,
      debug issues, or validate assumptions about the data.
  
      Args:
          sql: A SELECT query to execute
  
      Returns:
          JSON array of results, or error message
      """
      if not is_read_only(sql):
          return json.dumps({"error": "Only SELECT queries are allowed"})
  
      try:
          async with get_connection() as conn:
              rows = await conn.fetch(sql)
              # Convert to list of dicts, handling special types
              results = []
              for row in rows:
                  results.append({k: _serialize_value(v) for k, v in dict(row).items()})
              return json.dumps(results, indent=2, default=str)
      except Exception as e:
          return json.dumps({"error": str(e)})

• mcp_postgres/server.py:36-45 (helper)

  Helper function called by the query handler to ensure only safe, read-only SELECT queries are executed.

  def is_read_only(sql: str) -> bool:
      """Check if SQL is read-only (SELECT only)."""
      normalized = sql.strip().upper()
      # Must start with SELECT or WITH (for CTEs)
      if not (normalized.startswith("SELECT") or normalized.startswith("WITH")):
          return False
      # Block dangerous keywords even in subqueries
      dangerous = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE", "CREATE", "GRANT", "REVOKE"]
      return not any(kw in normalized for kw in dangerous)

• mcp_postgres/server.py:26-34 (helper)

  Async context manager for obtaining PostgreSQL connections, used by the query handler.

  @asynccontextmanager
  async def get_connection():
      """Get a database connection."""
      conn = await asyncpg.connect(**DB_CONFIG)
      try:
          yield conn
      finally:
          await conn.close()

• mcp_postgres/server.py:266-276 (helper)

  Utility function used in query handler to serialize PostgreSQL values (e.g., dates, bytes) to JSON-compatible formats.

  def _serialize_value(value):
      """Serialize special PostgreSQL types to JSON-compatible values."""
      if value is None:
          return None
      if isinstance(value, (dict, list)):
          return value
      if hasattr(value, "isoformat"):  # datetime, date, time
          return value.isoformat()
      if isinstance(value, bytes):
          return value.decode("utf-8", errors="replace")
      return value

• mcp_postgres/server.py:47-47 (registration)

  The @mcp.tool() decorator registers the query function as an MCP tool.

  @mcp.tool()