diff --git a/include/neuron/connection/mqtt_client.h b/include/neuron/connection/mqtt_client.h
index c6f06679..85621f5a 100644
--- a/include/neuron/connection/mqtt_client.h
+++ b/include/neuron/connection/mqtt_client.h
@@ -31,6 +31,8 @@ extern "C" {
 #include "utils/zlog.h"
 
 typedef void (*neu_mqtt_client_cb_t)(void *data);
+typedef void (*neu_mqtt_client_subscribe_cb_t)(const uint8_t *payload,
+                                               uint32_t len, void *data);
 
 typedef enum {
     NEU_MQTT_VERSION_V31  = 3,
@@ -73,7 +75,8 @@ int neu_mqtt_client_close(neu_mqtt_client_t *client);
 int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
                             const char *topic, const char *payload, size_t len);
 int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
-                              const char *topic, neu_mqtt_client_cb_t cb);
+                              const char *                   topic,
+                              neu_mqtt_client_subscribe_cb_t cb, void *data);
 
 #ifdef __cplusplus
 }
diff --git a/plugins/nano-mqtt/mqtt_handle.c b/plugins/nano-mqtt/mqtt_handle.c
index 67424fd4..1b42d387 100644
--- a/plugins/nano-mqtt/mqtt_handle.c
+++ b/plugins/nano-mqtt/mqtt_handle.c
@@ -17,8 +17,11 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
+#include "connection/mqtt_client.h"
 #include "errcodes.h"
 #include "version.h"
+#include "json/neu_json_mqtt.h"
+#include "json/neu_json_rw.h"
 
 #include "mqtt_handle.h"
 #include "mqtt_plugin.h"
@@ -176,22 +179,265 @@ static char *generate_upload_json(neu_plugin_t *            plugin,
     return json_str;
 }
 
+static char *generate_read_resp_json(neu_plugin_t *         plugin,
+                                     neu_json_mqtt_t *      mqtt,
+                                     neu_resp_read_group_t *data)
+{
+    neu_resp_tag_value_t *tags     = data->tags;
+    uint16_t              len      = data->n_tag;
+    char *                json_str = NULL;
+    neu_json_read_resp_t  json     = { 0 };
+
+    if (0 != tag_values_to_json(tags, len, &json)) {
+        plog_error(plugin, "tag_values_to_json fail");
+        return NULL;
+    }
+
+    neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, mqtt,
+                              neu_json_encode_mqtt_resp, &json_str);
+
+    if (json.tags) {
+        free(json.tags);
+    }
+    return json_str;
+}
+
+static char *generate_write_resp_json(neu_plugin_t *    plugin,
+                                      neu_json_mqtt_t * mqtt,
+                                      neu_resp_error_t *data)
+{
+    (void) plugin;
+
+    neu_json_error_resp_t error    = { .error = data->error };
+    char *                json_str = NULL;
+
+    neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, mqtt,
+                              neu_json_encode_mqtt_resp, &json_str);
+
+    return json_str;
+}
+
+static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
+                                neu_json_read_req_t *req)
+{
+    plog_info(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group,
+              req->node);
+
+    neu_reqresp_head_t header = { 0 };
+    header.ctx                = mqtt;
+    header.type               = NEU_REQ_READ_GROUP;
+    neu_req_read_group_t cmd  = { 0 };
+    strcpy(cmd.driver, req->node);
+    strcpy(cmd.group, req->group);
+    if (0 != neu_plugin_op(plugin, header, &cmd)) {
+        plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail");
+        return -1;
+    }
+
+    return 0;
+}
+
+static int send_write_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
+                          neu_json_write_req_t *req)
+{
+    plog_info(plugin, "write uuid:%s, group:%s, node:%s", mqtt->uuid,
+              req->group, req->node);
+
+    neu_reqresp_head_t  header = { 0 };
+    neu_req_write_tag_t cmd    = { 0 };
+
+    header.ctx  = mqtt;
+    header.type = NEU_REQ_WRITE_TAG;
+
+    strcpy(cmd.driver, req->node);
+    strcpy(cmd.group, req->group);
+    strcpy(cmd.tag, req->tag);
+
+    switch (req->t) {
+    case NEU_JSON_INT:
+        cmd.value.type      = NEU_TYPE_INT64;
+        cmd.value.value.u64 = req->value.val_int;
+        break;
+    case NEU_JSON_STR:
+        cmd.value.type = NEU_TYPE_STRING;
+        strcpy(cmd.value.value.str, req->value.val_str);
+        break;
+    case NEU_JSON_DOUBLE:
+        cmd.value.type      = NEU_TYPE_DOUBLE;
+        cmd.value.value.d64 = req->value.val_double;
+        break;
+    case NEU_JSON_BOOL:
+        cmd.value.type          = NEU_TYPE_BOOL;
+        cmd.value.value.boolean = req->value.val_bool;
+        break;
+    default:
+        plog_error(plugin, "invalid tag value type: %d", req->t);
+        return -1;
+    }
+
+    if (0 != neu_plugin_op(plugin, header, &cmd)) {
+        plog_error(plugin, "neu_plugin_op(NEU_REQ_WRITE_TAG) fail");
+        return -1;
+    }
+
+    return 0;
+}
+
+void handle_write_req(const uint8_t *payload, uint32_t len, void *data)
+{
+    int                   rv     = 0;
+    neu_plugin_t *        plugin = data;
+    neu_json_write_req_t *req    = NULL;
+
+    char *json_str = malloc(len + 1);
+    if (NULL == json_str) {
+        return;
+    }
+
+    memcpy(json_str, payload, len);
+    json_str[len] = '\0';
+
+    neu_json_mqtt_t *mqtt = NULL;
+    rv                    = neu_json_decode_mqtt_req(json_str, &mqtt);
+    if (0 != rv) {
+        plog_error(plugin, "neu_json_decode_mqtt_req failed");
+        free(json_str);
+        return;
+    }
+
+    rv = neu_json_decode_write_req(json_str, &req);
+    if (0 != rv) {
+        plog_error(plugin, "neu_json_decode_write_req fail");
+        neu_json_decode_mqtt_req_free(mqtt);
+        free(json_str);
+        return;
+    }
+
+    rv = send_write_req(plugin, mqtt, req);
+    if (0 != rv) {
+        neu_json_decode_mqtt_req_free(mqtt);
+    }
+
+    neu_json_decode_write_req_free(req);
+    free(json_str);
+}
+
 int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                           neu_resp_error_t *data)
 {
-    (void) plugin;
-    (void) mqtt_json;
-    (void) data;
-    return 0;
+    int   rv       = 0;
+    char *json_str = NULL;
+
+    if (NULL == plugin->client) {
+        rv = NEU_ERR_MQTT_IS_NULL;
+        goto end;
+    }
+
+    if (!neu_mqtt_client_is_connected(plugin->client)) {
+        rv = NEU_ERR_MQTT_FAILURE;
+        goto end;
+    }
+
+    json_str = generate_write_resp_json(plugin, mqtt_json, data);
+    if (NULL == json_str) {
+        plog_error(plugin, "generate write resp json fail, uuid:%s",
+                   mqtt_json->uuid);
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    const char *   topic = plugin->write_resp_topic;
+    neu_mqtt_qos_e qos   = NEU_MQTT_QOS0;
+    rv = neu_mqtt_client_publish(plugin->client, qos, topic, json_str,
+                                 strlen(json_str));
+    if (0 != rv) {
+        plog_error(plugin, "pub [%s, QoS%d] error", topic, qos);
+        rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
+    }
+
+end:
+    free(json_str);
+    neu_json_decode_mqtt_req_free(mqtt_json);
+    return rv;
+}
+
+void handle_read_req(const uint8_t *payload, uint32_t len, void *data)
+{
+    int                  rv     = 0;
+    neu_plugin_t *       plugin = data;
+    neu_json_read_req_t *req    = NULL;
+
+    char *json_str = malloc(len + 1);
+    if (NULL == json_str) {
+        return;
+    }
+
+    memcpy(json_str, payload, len);
+    json_str[len] = '\0';
+
+    neu_json_mqtt_t *mqtt = NULL;
+    rv                    = neu_json_decode_mqtt_req(json_str, &mqtt);
+    if (0 != rv) {
+        plog_error(plugin, "neu_json_decode_mqtt_req failed");
+        free(json_str);
+        return;
+    }
+
+    rv = neu_json_decode_read_req(json_str, &req);
+    if (0 != rv) {
+        plog_error(plugin, "neu_json_decode_read_req fail");
+        neu_json_decode_mqtt_req_free(mqtt);
+        free(json_str);
+        return;
+    }
+
+    rv = send_read_req(plugin, mqtt, req);
+    if (0 != rv) {
+        neu_json_decode_mqtt_req_free(mqtt);
+    }
+
+    neu_json_decode_read_req_free(req);
+    free(json_str);
 }
 
 int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                          neu_resp_read_group_t *data)
 {
-    (void) plugin;
-    (void) mqtt_json;
-    (void) data;
-    return 0;
+    int   rv       = 0;
+    char *json_str = NULL;
+
+    if (NULL == plugin->client) {
+        rv = NEU_ERR_MQTT_IS_NULL;
+        goto end;
+    }
+
+    if (!neu_mqtt_client_is_connected(plugin->client)) {
+        rv = NEU_ERR_MQTT_FAILURE;
+        goto end;
+    }
+
+    json_str = generate_read_resp_json(plugin, mqtt_json, data);
+    if (NULL == json_str) {
+        plog_error(plugin, "generate read resp json fail");
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    const char *   topic = plugin->read_resp_topic;
+    neu_mqtt_qos_e qos   = NEU_MQTT_QOS0;
+    rv = neu_mqtt_client_publish(plugin->client, qos, topic, json_str,
+                                 strlen(json_str));
+    if (0 != rv) {
+        plog_error(plugin, "pub [%s, QoS%d] error", topic, qos);
+        rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
+        goto end;
+    }
+
+end:
+    neu_json_decode_mqtt_req_free(mqtt_json);
+    free(json_str);
+    free(data->tags);
+    return rv;
 }
 
 int handle_trans_data(neu_plugin_t *            plugin,
diff --git a/plugins/nano-mqtt/mqtt_handle.h b/plugins/nano-mqtt/mqtt_handle.h
index fd32b861..8e97cef6 100644
--- a/plugins/nano-mqtt/mqtt_handle.h
+++ b/plugins/nano-mqtt/mqtt_handle.h
@@ -26,9 +26,13 @@ extern "C" {
 
 #include "neuron.h"
 
+void handle_write_req(const uint8_t *payload, uint32_t len, void *data);
+
 int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                           neu_resp_error_t *data);
 
+void handle_read_req(const uint8_t *payload, uint32_t len, void *data);
+
 int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                          neu_resp_read_group_t *data);
 
diff --git a/plugins/nano-mqtt/mqtt_plugin.c b/plugins/nano-mqtt/mqtt_plugin.c
index 1c39b6b1..a49c8294 100644
--- a/plugins/nano-mqtt/mqtt_plugin.c
+++ b/plugins/nano-mqtt/mqtt_plugin.c
@@ -17,9 +17,11 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
-#include "mqtt_plugin.h"
+#include "utils/asprintf.h"
+
 #include "mqtt_config.h"
 #include "mqtt_handle.h"
+#include "mqtt_plugin.h"
 
 const neu_plugin_module_t neu_plugin_module;
 
@@ -69,6 +71,16 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin)
         neu_mqtt_client_free(plugin->client);
         plugin->client = NULL;
     }
+
+    free(plugin->read_req_topic);
+    plugin->read_req_topic = NULL;
+    free(plugin->read_resp_topic);
+    plugin->read_resp_topic = NULL;
+    free(plugin->write_req_topic);
+    plugin->write_req_topic = NULL;
+    free(plugin->write_resp_topic);
+    plugin->write_resp_topic = NULL;
+
     plog_info(plugin, "uninitialize plugin `%s` success",
               neu_plugin_module.module_name);
     return NEU_ERR_SUCCESS;
@@ -185,6 +197,52 @@ error:
     return rv;
 }
 
+static int create_topic(neu_plugin_t *plugin)
+{
+    if (plugin->read_req_topic) {
+        // topics already created
+        return 0;
+    }
+
+    neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
+                 plugin->common.name);
+    if (NULL == plugin->read_req_topic) {
+        return -1;
+    }
+
+    neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
+                 plugin->common.name);
+    if (NULL == plugin->read_resp_topic) {
+        free(plugin->read_req_topic);
+        plugin->read_req_topic = NULL;
+        return -1;
+    }
+
+    neu_asprintf(&plugin->write_req_topic, "/neuron/%s/write/req",
+                 plugin->common.name);
+    if (NULL == plugin->read_req_topic) {
+        free(plugin->read_req_topic);
+        plugin->read_req_topic = NULL;
+        free(plugin->read_resp_topic);
+        plugin->read_resp_topic = NULL;
+        return -1;
+    }
+
+    neu_asprintf(&plugin->write_resp_topic, "/neuron/%s/write/resp",
+                 plugin->common.name);
+    if (NULL == plugin->read_req_topic) {
+        free(plugin->read_req_topic);
+        plugin->read_req_topic = NULL;
+        free(plugin->read_resp_topic);
+        plugin->read_resp_topic = NULL;
+        free(plugin->write_req_topic);
+        plugin->write_req_topic = NULL;
+        return -1;
+    }
+
+    return 0;
+}
+
 static int mqtt_plugin_start(neu_plugin_t *plugin)
 {
     int         rv          = 0;
@@ -202,11 +260,34 @@ static int mqtt_plugin_start(neu_plugin_t *plugin)
         goto end;
     }
 
+    if (0 != create_topic(plugin)) {
+        plog_error(plugin, "create topics fail");
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, 0, plugin->read_req_topic,
+                                  handle_read_req, plugin)) {
+        plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
+        rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+        goto end;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, 0, plugin->write_req_topic,
+                                  handle_write_req, plugin)) {
+        plog_error(plugin, "subscribe [%s] fail", plugin->write_req_topic);
+        rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+        goto end;
+    }
+
 end:
     if (0 == rv) {
         plog_info(plugin, "start plugin `%s` success", plugin_name);
     } else {
         plog_error(plugin, "start plugin `%s` failed", plugin_name);
+        neu_mqtt_client_close(plugin->client);
     }
     return rv;
 }
diff --git a/plugins/nano-mqtt/mqtt_plugin.h b/plugins/nano-mqtt/mqtt_plugin.h
index d783b60c..0718a8a8 100644
--- a/plugins/nano-mqtt/mqtt_plugin.h
+++ b/plugins/nano-mqtt/mqtt_plugin.h
@@ -33,6 +33,10 @@ struct neu_plugin {
     neu_plugin_common_t common;
     mqtt_config_t       config;
     neu_mqtt_client_t * client;
+    char *              read_req_topic;
+    char *              read_resp_topic;
+    char *              write_req_topic;
+    char *              write_resp_topic;
 };
 
 #ifdef __cplusplus
diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c
index cde8d083..a503635d 100644
--- a/src/connection/mqtt_client.c
+++ b/src/connection/mqtt_client.c
@@ -30,6 +30,8 @@
 
 #include "connection/mqtt_client.h"
 #include "utils/asprintf.h"
+#include "utils/utextend.h"
+#include "utils/uthash.h"
 #include "utils/zlog.h"
 
 #define log(level, ...)                               \
@@ -48,6 +50,14 @@
         }                                                                    \
     } while (0)
 
+typedef struct {
+    char *                         topic;
+    neu_mqtt_client_subscribe_cb_t cb;
+    void *                         data;
+    nng_aio *                      aio;
+    UT_hash_handle                 hh;
+} subscription_t;
+
 struct neu_mqtt_client_s {
     nng_socket              sock;
     char *                  url;
@@ -61,6 +71,7 @@ struct neu_mqtt_client_s {
     neu_mqtt_client_cb_t    disconnect_cb;
     void *                  disconnect_cb_data;
     nng_mqtt_sqlite_option *sqlite_cfg;
+    subscription_t *        subscriptions;
     zlog_category_t *       log;
 };
 
@@ -106,6 +117,44 @@ static void disconnect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     }
 }
 
+static void recv_cb(void *arg)
+{
+    int                rv           = 0;
+    subscription_t *   subscription = arg;
+    nng_aio *          aio          = subscription->aio;
+    neu_mqtt_client_t *client       = nng_aio_get_input(aio, 0);
+
+    if (0 != (rv = nng_aio_result(aio))) {
+        log(error, "mqtt client recv error: %s", nng_strerror(rv));
+        if (NNG_EAGAIN == rv) {
+            nng_recv_aio(client->sock, aio);
+        }
+        return;
+    }
+
+    nng_msg *msg = nng_aio_get_msg(aio);
+
+    uint32_t    payload_len;
+    uint8_t *   payload = nng_mqtt_msg_get_publish_payload(msg, &payload_len);
+    uint32_t    topic_len;
+    const char *topic = nng_mqtt_msg_get_publish_topic(msg, &topic_len);
+    uint8_t     qos   = nng_mqtt_msg_get_publish_qos(msg);
+
+    log(info, "recv [%.*s, QoS%d] %" PRIu32 " bytes", (unsigned) topic_len,
+        topic, (int) qos, payload_len);
+
+    // find matching subscription
+    HASH_FIND(hh, client->subscriptions, topic, topic_len, subscription);
+    if (NULL != subscription) {
+        subscription->cb(payload, payload_len, subscription->data);
+    } else {
+        log(warn, "[%.*s] no subscription found", (unsigned) topic_len, topic);
+    }
+
+    nng_msg_free(msg);
+    nng_recv_aio(client->sock, aio);
+}
+
 static nng_msg *alloc_conn_msg(neu_mqtt_client_t *client,
                                neu_mqtt_version_e version)
 {
@@ -159,6 +208,78 @@ alloc_sqlite_config(neu_mqtt_client_t *client)
     return cfg;
 }
 
+static subscription_t *subscription_new(neu_mqtt_client_t *            client,
+                                        const char *                   topic,
+                                        neu_mqtt_client_subscribe_cb_t cb,
+                                        void *                         data)
+{
+    int             rv           = 0;
+    subscription_t *subscription = NULL;
+
+    subscription = calloc(1, sizeof(*subscription));
+    if (NULL == subscription) {
+        return NULL;
+    }
+
+    subscription->topic = strdup(topic);
+    if (NULL == subscription->topic) {
+        free(subscription);
+        return NULL;
+    }
+
+    if ((rv = nng_aio_alloc(&subscription->aio, recv_cb, subscription)) != 0) {
+        log(error, "nng_aio_alloc fail: %s", nng_strerror(rv));
+        free(subscription->topic);
+        free(subscription);
+        return NULL;
+    }
+
+    if ((rv = nng_aio_set_input(subscription->aio, 0, client)) != 0) {
+        log(error, "nng_aio_set_input fail: %s", nng_strerror(rv));
+        nng_aio_free(subscription->aio);
+        free(subscription->topic);
+        free(subscription);
+        return NULL;
+    }
+
+    subscription->cb   = cb;
+    subscription->data = data;
+
+    return subscription;
+}
+
+static inline void subscription_free(subscription_t *subscription)
+{
+    if (subscription) {
+        nng_aio_stop(subscription->aio);
+        nng_aio_free(subscription->aio);
+        free(subscription->topic);
+        free(subscription);
+    }
+}
+
+static inline void subscriptions_free(subscription_t *subscriptions)
+{
+    subscription_t *sub = NULL, *tmp = NULL;
+    HASH_ITER(hh, subscriptions, sub, tmp)
+    {
+        HASH_DEL(subscriptions, sub);
+        subscription_free(sub);
+    }
+}
+
+static void add_subscription(neu_mqtt_client_t *client, subscription_t *sub)
+{
+    subscription_t *old = NULL;
+
+    HASH_FIND_STR(client->subscriptions, sub->topic, old);
+    if (old) {
+        HASH_DEL(client->subscriptions, old);
+        subscription_free(old);
+    }
+    HASH_ADD_STR(client->subscriptions, topic, sub);
+}
+
 neu_mqtt_client_t *neu_mqtt_client_new(const char *host, uint16_t port,
                                        neu_mqtt_version_e version)
 {
@@ -196,6 +317,7 @@ void neu_mqtt_client_free(neu_mqtt_client_t *client)
         if (client->sqlite_cfg) {
             nng_mqtt_free_sqlite_opt(client->sqlite_cfg);
         }
+        subscriptions_free(client->subscriptions);
         nng_msg_free(client->conn_msg);
         free(client->url);
         free(client);
@@ -412,16 +534,23 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
 }
 
 int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
-                              const char *topic, neu_mqtt_client_cb_t cb)
+                              const char *                   topic,
+                              neu_mqtt_client_subscribe_cb_t cb, void *data)
 {
-    int      rv      = 0;
-    nng_msg *sub_msg = NULL;
+    int             rv           = 0;
+    nng_msg *       sub_msg      = NULL;
+    subscription_t *subscription = NULL;
 
     if ((rv = nng_mqtt_msg_alloc(&sub_msg, 0)) != 0) {
         log(error, "nng_mqtt_msg_alloc fail: %s", nng_strerror(rv));
         return -1;
     }
 
+    subscription = subscription_new(client, topic, cb, data);
+    if (NULL == subscription) {
+        goto error;
+    }
+
     nng_mqtt_topic_qos topic_qos[] = {
         { .qos   = qos,
           .topic = { .buf = (uint8_t *) topic, .length = strlen(topic) } },
@@ -435,10 +564,18 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
     rv = nng_sendmsg(client->sock, sub_msg, NNG_FLAG_NONBLOCK);
     if (0 != rv) {
         log(error, "nng_sendmsg fail: %s", nng_strerror(rv));
-        return -1;
+        goto error;
     }
 
-    (void) cb; // TODO: handle callback on subscribed topic
+    add_subscription(client, subscription);
+    log(info, "sub [%s, QoS%d]", topic, qos);
+
+    nng_recv_aio(client->sock, subscription->aio);
 
     return 0;
+
+error:
+    nng_msg_free(sub_msg);
+    subscription_free(subscription);
+    return -1;
 }