diff --git a/pymysql/converters.py b/pymysql/converters.py
index 4c00bc45..fc596ded 100644
--- a/pymysql/converters.py
+++ b/pymysql/converters.py
@@ -333,6 +333,12 @@ def convert_set(s):
     return set(s.split(","))
 
 
+def convert_json(b):
+    # JSON is returned as binary data.
+    # Decode with utf-8 regardless connection encoding.
+    return b.decode('utf-8')
+
+
 def through(x):
     return x
 
@@ -410,6 +416,7 @@ def convert_characters(connection, field, data):
     FIELD_TYPE.VARCHAR: through,
     FIELD_TYPE.DECIMAL: Decimal,
     FIELD_TYPE.NEWDECIMAL: Decimal,
+    FIELD_TYPE.JSON: convert_json,
 }
 
 
diff --git a/pymysql/tests/test_basic.py b/pymysql/tests/test_basic.py
index a52fcb2e..52d9c7ff 100644
--- a/pymysql/tests/test_basic.py
+++ b/pymysql/tests/test_basic.py
@@ -1,15 +1,16 @@
-import pymysql.cursors
-
-from pymysql.tests import base
-from pymysql import util
-from pymysql.err import ProgrammingError
-
-import time
+# coding: utf-8
 import datetime
+import json
+import time
 import warnings
 
 from unittest2 import SkipTest
 
+from pymysql import util
+import pymysql.cursors
+from pymysql.tests import base
+from pymysql.err import ProgrammingError
+
 
 __all__ = ["TestConversion", "TestCursor", "TestBulkInserts"]
 
@@ -238,6 +239,26 @@ def test_single_tuple(self):
         self.assertEqual([(1,)], list(c.fetchall()))
         c.close()
 
+    def test_json(self):
+        args = self.databases[0].copy()
+        args["charset"] = "utf8mb4"
+        conn = pymysql.connect(**args)
+        if not self.mysql_server_is(conn, (5, 7, 0)):
+            raise SkipTest("JSON type is not supported on MySQL <= 5.6")
+
+        self.safe_create_table(conn, "test_json", """\
+create table test_json (
+    id int not null,
+    json JSON not null,
+    primary key (id)
+);""")
+        cur = conn.cursor()
+        json_str = u'{"hello": "こんにちは"}'
+        cur.execute("INSERT INTO test_json (id, `json`) values (42, %s)", (json_str,))
+        cur.execute("SELECT `json` from `test_json` WHERE `id`=42")
+        res = cur.fetchone()[0]
+        self.assertEqual(json.loads(res), json.loads(json_str))
+
 
 class TestBulkInserts(base.PyMySQLTestCase):