switch to JSON-RPC as message format

Instead of the weird `struct msg` I had, I switched to the JSON-RPC
format. It's basically the same, but has a well-defined semantics in
case of errors.

18 files changed, 1232 insertions, 499 deletions

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 4c2c0e1..5ce967f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -32,7 +32,7 @@ jobs:
       - name: Install dependencies
         run: |
           sudo apt-get update
-          sudo DEBIAN_FRONTEND=noninteractive apt-get install -yq --no-install-recommends libgit2-dev libsodium-dev libsqlite3-dev python3-pytest valgrind
+          sudo DEBIAN_FRONTEND=noninteractive apt-get install -yq --no-install-recommends libgit2-dev libjson-c-dev libsodium-dev libsqlite3-dev python3-pytest valgrind
       - name: Build
         run: make install
       - name: Upload binaries
@@ -55,7 +55,7 @@ jobs:
       - name: Install dependencies
         run: |
           sudo apt-get update
-          sudo DEBIAN_FRONTEND=noninteractive apt-get install -yq --no-install-recommends gcovr libgit2-dev libsodium-dev libsqlite3-dev python3-pytest valgrind
+          sudo DEBIAN_FRONTEND=noninteractive apt-get install -yq --no-install-recommends gcovr libgit2-dev libjson-c-dev libsodium-dev libsqlite3-dev python3-pytest valgrind
       - name: Generate report
         run: make coverage
       - name: Upload report
diff --git a/Dockerfile b/Dockerfile
index 11ce0b0..6f7de1d 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -4,7 +4,7 @@ ARG install_dir="/app/install"
 
 FROM base AS builder
 
-RUN build_deps='bash bsd-compat-headers build-base clang cmake coreutils git libgit2-dev libsodium-dev py3-pytest sqlite-dev valgrind' && \
+RUN build_deps='bash bsd-compat-headers build-base clang cmake coreutils git json-c-dev libgit2-dev libsodium-dev py3-pytest sqlite-dev valgrind' && \
     apk add -q --no-cache $build_deps
 
 ARG COMPILER=clang
@@ -31,7 +31,7 @@ FROM base
 
 LABEL maintainer="Egor Tensin <Egor.Tensin@gmail.com>"
 
-RUN runtime_deps='libgit2 libsodium sqlite tini' && \
+RUN runtime_deps='json-c libgit2 libsodium sqlite tini' && \
     apk add -q --no-cache $runtime_deps
 
 ARG install_dir
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index d994943..a6a42df 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -60,8 +60,9 @@ add_my_executable(server server_main.c server.c
     const.c
     event_loop.c
     file.c
+    json.c
+    json_rpc.c
     log.c
-    msg.c
     net.c
     process.c
     protocol.c
@@ -74,16 +75,22 @@ add_my_executable(server server_main.c server.c
     string.c
     tcp_server.c
     worker_queue.c)
-target_link_libraries(server PRIVATE pthread sodium sqlite3)
+target_link_libraries(server PRIVATE json-c pthread sodium sqlite3)
 target_include_directories(server PRIVATE "${CMAKE_CURRENT_BINARY_DIR}")
 
 add_my_executable(client client_main.c client.c
+    base64.c
     cmd_line.c
     const.c
     file.c
+    json.c
+    json_rpc.c
     log.c
-    msg.c
-    net.c)
+    net.c
+    process.c
+    protocol.c
+    run_queue.c)
+target_link_libraries(client PRIVATE json-c sodium)
 
 add_my_executable(worker worker_main.c worker.c
     base64.c
@@ -94,12 +101,13 @@ add_my_executable(worker worker_main.c worker.c
     event_loop.c
     file.c
     git.c
+    json.c
+    json_rpc.c
     log.c
-    msg.c
     net.c
     process.c
     protocol.c
     run_queue.c
     signal.c
     string.c)
-target_link_libraries(worker PRIVATE git2 pthread sodium)
+target_link_libraries(worker PRIVATE git2 json-c pthread sodium)
diff --git a/src/client.c b/src/client.c
index 5259d52..8c6c597 100644
--- a/src/client.c
+++ b/src/client.c
@@ -8,10 +8,15 @@
 #include "client.h"
 #include "cmd_line.h"
 #include "compiler.h"
+#include "const.h"
+#include "json_rpc.h"
 #include "log.h"
-#include "msg.h"
+#include "net.h"
+#include "protocol.h"
+#include "run_queue.h"
 
 #include <stdlib.h>
+#include <string.h>
 
 struct client {
 	int dummy;
@@ -36,30 +41,70 @@ void client_destroy(struct client *client)
 	free(client);
 }
 
+static int make_request(struct jsonrpc_request **request, int argc, const char **argv)
+{
+	if (argc < 1) {
+		exit_with_usage_err("no action specified");
+		return -1;
+	}
+
+	if (!strcmp(argv[0], CMD_RUN)) {
+		if (argc != 3)
+			return -1;
+
+		struct run *run = NULL;
+		int ret = run_create(&run, 0, argv[1], argv[2]);
+		if (ret < 0)
+			return ret;
+
+		ret = run_request_create(request, run);
+		run_destroy(run);
+		return ret;
+	}
+
+	return -1;
+}
+
 int client_main(UNUSED const struct client *client, const struct settings *settings, int argc,
                 const char **argv)
 {
-	struct msg *response = NULL;
 	int ret = 0;
 
-	if (argc < 1) {
-		exit_with_usage_err("no message to send to the server");
-		return -1;
+	struct jsonrpc_request *request = NULL;
+	ret = make_request(&request, argc, argv);
+	if (ret < 0) {
+		exit_with_usage_err("invalid request");
+		return ret;
 	}
 
-	ret = msg_connect_and_talk_argv(settings->host, settings->port, argv, &response);
-	if (ret < 0 || !response || !msg_is_success(response)) {
-		log_err("Failed to connect to server or it couldn't process the request\n");
-		if (response)
-			msg_dump(response);
-		if (!ret)
-			ret = -1;
+	ret = net_connect(settings->host, settings->port);
+	if (ret < 0)
+		goto free_request;
+	int fd = ret;
+
+	ret = jsonrpc_request_send(request, fd);
+	if (ret < 0)
+		goto close;
+
+	struct jsonrpc_response *response = NULL;
+	ret = jsonrpc_response_recv(&response, fd);
+	if (ret < 0)
+		goto close;
+
+	if (jsonrpc_response_is_error(response)) {
+		log_err("server failed to process the request\n");
+		ret = -1;
 		goto free_response;
 	}
 
 free_response:
-	if (response)
-		msg_free(response);
+	jsonrpc_response_destroy(response);
+
+close:
+	net_close(fd);
+
+free_request:
+	jsonrpc_request_destroy(request);
 
 	return ret;
 }
diff --git a/src/client_main.c b/src/client_main.c
index 40d7995..15ad449 100644
--- a/src/client_main.c
+++ b/src/client_main.c
@@ -24,7 +24,10 @@ static struct settings default_settings(void)
 
 const char *get_usage_string(void)
 {
-	return "[-h|--help] [-V|--version] [-v|--verbose] [-H|--host HOST] [-p|--port PORT] ACTION [ARG...]";
+	return "[-h|--help] [-V|--version] [-v|--verbose] [-H|--host HOST] [-p|--port PORT] ACTION [ARG...]\n\
+\n\
+available actions:\n\
+\t" CMD_RUN " URL REV - schedule a CI run of repository at URL, revision REV";
 }
 
 static int parse_settings(struct settings *settings, int argc, char *argv[])
diff --git a/src/command.c b/src/command.c
index 47280fe..9946c13 100644
--- a/src/command.c
+++ b/src/command.c
@@ -8,8 +8,8 @@
 #include "command.h"
 #include "compiler.h"
 #include "event_loop.h"
+#include "json_rpc.h"
 #include "log.h"
-#include "msg.h"
 
 #include <poll.h>
 #include <stdlib.h>
@@ -107,9 +107,10 @@ void cmd_dispatcher_destroy(struct cmd_dispatcher *dispatcher)
 }
 
 static int cmd_dispatcher_handle_internal(const struct cmd_dispatcher *dispatcher,
-                                          const struct msg *command, struct msg **result, void *arg)
+                                          const struct jsonrpc_request *request,
+                                          struct jsonrpc_response **result, void *arg)
 {
-	const char *actual_cmd = msg_get_first_string(command);
+	const char *actual_cmd = jsonrpc_request_get_method(request);
 
 	for (size_t i = 0; i < dispatcher->numof_cmds; ++i) {
 		struct cmd_desc *cmd = &dispatcher->cmds[i];
@@ -117,16 +118,15 @@ static int cmd_dispatcher_handle_internal(const struct cmd_dispatcher *dispatche
 		if (strcmp(cmd->name, actual_cmd))
 			continue;
 
-		return cmd->handler(command, result, arg);
+		return cmd->handler(request, result, arg);
 	}
 
-	log_err("Received an unknown command\n");
-	msg_dump(command);
+	log_err("Received an unknown command: %s\n", actual_cmd);
 	return -1;
 }
 
-int cmd_dispatcher_handle(const struct cmd_dispatcher *dispatcher, const struct msg *command,
-                          struct msg **result)
+int cmd_dispatcher_handle(const struct cmd_dispatcher *dispatcher,
+                          const struct jsonrpc_request *command, struct jsonrpc_response **result)
 {
 	return cmd_dispatcher_handle_internal(dispatcher, command, result, dispatcher->ctx);
 }
@@ -147,32 +147,59 @@ static struct cmd_conn_ctx *make_conn_ctx(int fd, void *arg)
 
 static int cmd_dispatcher_handle_conn_internal(int conn_fd, struct cmd_dispatcher *dispatcher)
 {
-	struct msg *request = NULL, *response = NULL;
 	int ret = 0;
 
 	struct cmd_conn_ctx *new_ctx = make_conn_ctx(conn_fd, dispatcher->ctx);
 	if (!new_ctx)
 		return -1;
 
-	ret = msg_recv(conn_fd, &request);
+	struct jsonrpc_request *request = NULL;
+	ret = jsonrpc_request_recv(&request, conn_fd);
 	if (ret < 0)
 		goto free_ctx;
 
-	ret = cmd_dispatcher_handle_internal(dispatcher, request, &response, new_ctx);
-	if (ret < 0)
-		goto free_response;
+	const int requires_response = !jsonrpc_request_is_notification(request);
+
+	struct jsonrpc_response *default_response = NULL;
+	if (requires_response) {
+		ret = jsonrpc_response_create(&default_response, request, NULL);
+		if (ret < 0)
+			goto free_request;
+	}
 
-	if (response) {
-		ret = msg_send(conn_fd, response);
+	struct jsonrpc_response *default_error = NULL;
+	if (requires_response) {
+		ret = jsonrpc_error_create(&default_error, request, -1, "An error occured");
 		if (ret < 0)
-			goto free_response;
+			goto free_default_response;
+	}
+
+	struct jsonrpc_response *response = NULL;
+	ret = cmd_dispatcher_handle_internal(dispatcher, request, &response, new_ctx);
+
+	if (requires_response) {
+		struct jsonrpc_response *actual_response = response;
+		if (!actual_response) {
+			actual_response = ret < 0 ? default_error : default_response;
+		}
+		if (ret < 0 && !jsonrpc_response_is_error(actual_response)) {
+			actual_response = default_error;
+		}
+		ret = jsonrpc_response_send(actual_response, conn_fd) < 0 ? -1 : ret;
 	}
 
-free_response:
 	if (response)
-		msg_free(response);
+		jsonrpc_response_destroy(response);
+
+	if (default_error)
+		jsonrpc_response_destroy(default_error);
+
+free_default_response:
+	if (default_response)
+		jsonrpc_response_destroy(default_response);
 
-	msg_free(request);
+free_request:
+	jsonrpc_request_destroy(request);
 
 free_ctx:
 	free(new_ctx);
diff --git a/src/command.h b/src/command.h
index 8daa912..0edba98 100644
--- a/src/command.h
+++ b/src/command.h
@@ -9,11 +9,12 @@
 #define __COMMAND_H__
 
 #include "event_loop.h"
-#include "msg.h"
+#include "json_rpc.h"
 
 #include <stddef.h>
 
-typedef int (*cmd_handler)(const struct msg *request, struct msg **response, void *ctx);
+typedef int (*cmd_handler)(const struct jsonrpc_request *request,
+                           struct jsonrpc_response **response, void *ctx);
 
 struct cmd_desc {
 	char *name;
@@ -26,8 +27,8 @@ int cmd_dispatcher_create(struct cmd_dispatcher **, struct cmd_desc *, size_t nu
                           void *ctx);
 void cmd_dispatcher_destroy(struct cmd_dispatcher *);
 
-int cmd_dispatcher_handle(const struct cmd_dispatcher *, const struct msg *command,
-                          struct msg **response);
+int cmd_dispatcher_handle(const struct cmd_dispatcher *, const struct jsonrpc_request *command,
+                          struct jsonrpc_response **response);
 
 struct cmd_conn_ctx {
 	int fd;
diff --git a/src/json.c b/src/json.c
new file mode 100644
index 0000000..a01aae6
--- /dev/null
+++ b/src/json.c
@@ -0,0 +1,256 @@
+/*
+ * Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+ * This file is part of the "cimple" project.
+ * For details, see https://github.com/egor-tensin/cimple.
+ * Distributed under the MIT License.
+ */
+
+#include "json.h"
+#include "log.h"
+#include "net.h"
+
+#include <json-c/json_object.h>
+#include <json-c/json_tokener.h>
+
+#include <errno.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+
+char *json_to_string(struct json_object *obj)
+{
+	const char *result = json_object_to_json_string(obj);
+	if (!result) {
+		json_errno("json_object_to_json_string");
+		return NULL;
+	}
+
+	char *_result = strdup(result);
+	if (!_result) {
+		log_errno("strdup");
+		return NULL;
+	}
+
+	return _result;
+}
+
+struct json_object *json_from_string(const char *src)
+{
+	enum json_tokener_error error;
+
+	struct json_object *result = json_tokener_parse_verbose(src, &error);
+	if (!result) {
+		json_errno("json_tokener_parse_verbose");
+		log_err("JSON: parsing failed: %s\n", json_tokener_error_desc(error));
+		return NULL;
+	}
+
+	return result;
+}
+
+int json_clone(const struct json_object *obj, const char *key, struct json_object **_value)
+{
+	int ret = 0;
+
+	struct json_object *old_value = NULL;
+	ret = json_get(obj, key, &old_value);
+	if (ret < 0)
+		return ret;
+
+	struct json_object *new_value = NULL;
+	ret = json_object_deep_copy(old_value, &new_value, NULL);
+	if (ret < 0)
+		return ret;
+
+	*_value = new_value;
+	return ret;
+}
+
+int json_send(struct json_object *obj, int fd)
+{
+	int ret = 0;
+
+	const char *str = json_to_string(obj);
+	if (!str)
+		return -1;
+
+	struct buf *buf = NULL;
+	ret = buf_pack_strings(&buf, 1, &str);
+	free((char *)str);
+	if (ret < 0)
+		return ret;
+
+	ret = net_send_buf(fd, buf);
+	buf_destroy(buf);
+	if (ret < 0)
+		return ret;
+
+	return ret;
+}
+
+struct json_object *json_recv(int fd)
+{
+	struct json_object *result = NULL;
+	int ret = 0;
+
+	struct buf *buf = NULL;
+	ret = net_recv_buf(fd, &buf);
+	if (ret < 0)
+		return NULL;
+
+	result = json_from_string((const char *)buf_get_data(buf));
+	if (!result)
+		goto destroy_buf;
+
+destroy_buf:
+	buf_destroy(buf);
+
+	return result;
+}
+
+int json_has(const struct json_object *obj, const char *key)
+{
+	return json_object_object_get_ex(obj, key, NULL);
+}
+
+int json_get(const struct json_object *obj, const char *key, struct json_object **value)
+{
+	if (!json_has(obj, key)) {
+		log_err("JSON: key is missing: %s\n", key);
+		return -1;
+	}
+
+	return json_object_object_get_ex(obj, key, value);
+}
+
+int json_get_string(const struct json_object *obj, const char *key, const char **_value)
+{
+	struct json_object *value = NULL;
+
+	int ret = json_get(obj, key, &value);
+	if (ret < 0)
+		return ret;
+
+	if (!json_object_is_type(value, json_type_string)) {
+		log_err("JSON: key is not a string: %s\n", key);
+		return -1;
+	}
+
+	*_value = json_object_get_string(value);
+	return 0;
+}
+
+int json_get_int(const struct json_object *obj, const char *key, int64_t *_value)
+{
+	struct json_object *value = NULL;
+
+	int ret = json_get(obj, key, &value);
+	if (ret < 0)
+		return ret;
+
+	if (!json_object_is_type(value, json_type_int)) {
+		log_err("JSON: key is not an integer: %s\n", key);
+		return -1;
+	}
+
+	errno = 0;
+	int64_t tmp = json_object_get_int64(value);
+	if (errno) {
+		log_err("JSON: failed to parse integer from key: %s\n", key);
+		return -1;
+	}
+
+	*_value = tmp;
+	return 0;
+}
+
+static int json_set_internal(struct json_object *obj, const char *key, struct json_object *value,
+                             unsigned flags)
+{
+	int ret = 0;
+
+	ret = json_object_object_add_ex(obj, key, value, flags);
+	if (ret < 0) {
+		json_errno("json_object_object_add_ex");
+		return ret;
+	}
+
+	return 0;
+}
+
+static int json_set_string_internal(struct json_object *obj, const char *key, const char *_value,
+                                    unsigned flags)
+{
+	struct json_object *value = json_object_new_string(_value);
+	if (!value) {
+		json_errno("json_object_new_string");
+		return -1;
+	}
+
+	int ret = json_set_internal(obj, key, value, flags);
+	if (ret < 0)
+		goto free_value;
+
+	return ret;
+
+free_value:
+	json_object_put(value);
+
+	return ret;
+}
+
+static int json_set_int_internal(struct json_object *obj, const char *key, int64_t _value,
+                                 unsigned flags)
+{
+	struct json_object *value = json_object_new_int64(_value);
+	if (!value) {
+		json_errno("json_object_new_int");
+		return -1;
+	}
+
+	int ret = json_set_internal(obj, key, value, flags);
+	if (ret < 0)
+		goto free_value;
+
+	return ret;
+
+free_value:
+	json_object_put(value);
+
+	return ret;
+}
+
+int json_set(struct json_object *obj, const char *key, struct json_object *value)
+{
+	return json_set_internal(obj, key, value, 0);
+}
+
+int json_set_string(struct json_object *obj, const char *key, const char *value)
+{
+	return json_set_string_internal(obj, key, value, 0);
+}
+
+int json_set_int(struct json_object *obj, const char *key, int64_t value)
+{
+	return json_set_int_internal(obj, key, value, 0);
+}
+
+#ifndef JSON_C_OBJECT_ADD_CONSTANT_KEY
+#define JSON_C_OBJECT_ADD_CONSTANT_KEY JSON_C_OBJECT_KEY_IS_CONSTANT
+#endif
+static const unsigned json_const_key_flags = JSON_C_OBJECT_ADD_CONSTANT_KEY;
+
+int json_set_const_key(struct json_object *obj, const char *key, struct json_object *value)
+{
+	return json_set_internal(obj, key, value, json_const_key_flags);
+}
+
+int json_set_string_const_key(struct json_object *obj, const char *key, const char *value)
+{
+	return json_set_string_internal(obj, key, value, json_const_key_flags);
+}
+
+int json_set_int_const_key(struct json_object *obj, const char *key, int64_t value)
+{
+	return json_set_int_internal(obj, key, value, json_const_key_flags);
+}
diff --git a/src/json.h b/src/json.h
new file mode 100644
index 0000000..77bfde4
--- /dev/null
+++ b/src/json.h
@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+ * This file is part of the "cimple" project.
+ * For details, see https://github.com/egor-tensin/cimple.
+ * Distributed under the MIT License.
+ */
+
+#ifndef __JSON_H__
+#define __JSON_H__
+
+#include "log.h"
+
+#include <json-c/json_object.h>
+
+#include <stdint.h>
+
+#define json_errno(fn)                                                                             \
+	do {                                                                                       \
+		log_err("JSON: %s failed\n", fn);                                                  \
+	} while (0)
+
+char *json_to_string(struct json_object *);
+struct json_object *json_from_string(const char *);
+
+int json_clone(const struct json_object *, const char *key, struct json_object **value);
+
+int json_send(struct json_object *, int fd);
+struct json_object *json_recv(int fd);
+
+int json_has(const struct json_object *, const char *key);
+
+int json_get(const struct json_object *, const char *key, struct json_object **value);
+int json_get_string(const struct json_object *, const char *key, const char **value);
+int json_get_int(const struct json_object *, const char *key, int64_t *value);
+
+int json_set(struct json_object *, const char *key, struct json_object *value);
+int json_set_string(struct json_object *, const char *key, const char *value);
+int json_set_int(struct json_object *, const char *key, int64_t value);
+
+int json_set_const_key(struct json_object *, const char *, struct json_object *value);
+int json_set_string_const_key(struct json_object *, const char *, const char *value);
+int json_set_int_const_key(struct json_object *, const char *, int64_t value);
+
+#endif
diff --git a/src/json_rpc.c b/src/json_rpc.c
new file mode 100644
index 0000000..538cff0
--- /dev/null
+++ b/src/json_rpc.c
@@ -0,0 +1,587 @@
+/*
+ * Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+ * This file is part of the "cimple" project.
+ * For details, see https://github.com/egor-tensin/cimple.
+ * Distributed under the MIT License.
+ */
+
+#include "json_rpc.h"
+#include "json.h"
+#include "log.h"
+
+#include <json-c/json_object.h>
+
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+
+struct jsonrpc_request {
+	struct json_object *impl;
+};
+
+struct jsonrpc_response {
+	struct json_object *impl;
+};
+
+static const char *const jsonrpc_key_version = "jsonrpc";
+static const char *const jsonrpc_key_id = "id";
+static const char *const jsonrpc_key_method = "method";
+static const char *const jsonrpc_key_params = "params";
+
+static const char *const jsonrpc_value_version = "2.0";
+
+static int jsonrpc_check_version(struct json_object *obj)
+{
+	const char *key = jsonrpc_key_version;
+	const char *version = NULL;
+
+	int ret = json_get_string(obj, key, &version);
+	if (ret < 0)
+		return ret;
+
+	if (strcmp(version, jsonrpc_value_version)) {
+		log_err("JSON-RPC: invalid '%s' value: %s\n", key, version);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int jsonrpc_set_version(struct json_object *obj)
+{
+	return json_set_string_const_key(obj, jsonrpc_key_version, jsonrpc_value_version);
+}
+
+static int jsonrpc_check_id_type(struct json_object *id)
+{
+	if (!json_object_is_type(id, json_type_string) && !json_object_is_type(id, json_type_int)) {
+		log_err("JSON-RPC: key '%s' must be either an integer or a string\n",
+		        jsonrpc_key_id);
+		return -1;
+	}
+	return 0;
+}
+
+static int jsonrpc_check_id(struct json_object *obj, int required)
+{
+	const char *key = jsonrpc_key_id;
+
+	if (!json_has(obj, key)) {
+		if (!required)
+			return 0;
+		log_err("JSON-RPC: key is missing: %s\n", key);
+		return -1;
+	}
+
+	struct json_object *id = NULL;
+
+	int ret = json_get(obj, key, &id);
+	if (ret < 0)
+		return ret;
+	return jsonrpc_check_id_type(id);
+}
+
+static int jsonrpc_set_id(struct json_object *obj, int id)
+{
+	return json_set_int_const_key(obj, jsonrpc_key_id, id);
+}
+
+static int jsonrpc_check_method(struct json_object *obj)
+{
+	const char *key = jsonrpc_key_method;
+	const char *method = NULL;
+	return json_get_string(obj, key, &method);
+}
+
+static int jsonrpc_set_method(struct json_object *obj, const char *method)
+{
+	return json_set_string_const_key(obj, jsonrpc_key_method, method);
+}
+
+static int jsonrpc_check_params_type(struct json_object *params)
+{
+	if (!json_object_is_type(params, json_type_object) &&
+	    !json_object_is_type(params, json_type_array)) {
+		log_err("JSON-RPC: key '%s' must be either an object or an array\n",
+		        jsonrpc_key_params);
+		return -1;
+	}
+	return 0;
+}
+
+static int jsonrpc_check_params(struct json_object *obj)
+{
+	const char *key = jsonrpc_key_params;
+
+	if (!json_has(obj, key))
+		return 0;
+
+	struct json_object *params = NULL;
+
+	int ret = json_get(obj, key, &params);
+	if (ret < 0)
+		return ret;
+	return jsonrpc_check_params_type(params);
+}
+
+static int jsonrpc_set_params(struct json_object *obj, struct json_object *params)
+{
+	const char *key = jsonrpc_key_params;
+
+	int ret = jsonrpc_check_params_type(params);
+	if (ret < 0)
+		return ret;
+	return json_set_const_key(obj, key, params);
+}
+
+static const char *const jsonrpc_key_result = "result";
+static const char *const jsonrpc_key_error = "error";
+
+static const char *const jsonrpc_key_code = "code";
+static const char *const jsonrpc_key_message = "message";
+
+static int jsonrpc_check_error(struct json_object *obj)
+{
+	const char *key = jsonrpc_key_error;
+	struct json_object *error = NULL;
+
+	int ret = json_get(obj, key, &error);
+	if (ret < 0)
+		return ret;
+
+	int64_t code = -1;
+
+	ret = json_get_int(error, jsonrpc_key_code, &code);
+	if (ret < 0) {
+		log_err("JSON-RPC: key is missing or not an integer: %s\n", jsonrpc_key_code);
+		return -1;
+	}
+
+	const char *message = NULL;
+
+	ret = json_get_string(error, jsonrpc_key_message, &message);
+	if (ret < 0) {
+		log_err("JSON-RPC: key is missing or not a string: %s\n", jsonrpc_key_message);
+		return -1;
+	}
+
+	return ret;
+}
+
+static int jsonrpc_check_result_or_error(struct json_object *obj)
+{
+	const char *key_result = jsonrpc_key_result;
+	const char *key_error = jsonrpc_key_error;
+
+	if (!json_has(obj, key_result) && !json_has(obj, key_error)) {
+		log_err("JSON-RPC: either '%s' or '%s' must be present\n", key_result, key_error);
+		return -1;
+	}
+
+	if (json_has(obj, key_result))
+		return 0;
+
+	return jsonrpc_check_error(obj);
+}
+
+static int jsonrpc_request_create_internal(struct jsonrpc_request **_request, int *id,
+                                           const char *method, struct json_object *params)
+{
+	int ret = 0;
+
+	struct jsonrpc_request *request = malloc(sizeof(struct jsonrpc_request));
+	if (!request) {
+		log_errno("malloc");
+		ret = -1;
+		goto exit;
+	}
+
+	request->impl = json_object_new_object();
+	if (!request->impl) {
+		json_errno("json_object_new_object");
+		ret = -1;
+		goto free;
+	}
+
+	ret = jsonrpc_set_version(request->impl);
+	if (ret < 0)
+		goto free_impl;
+
+	if (id) {
+		ret = jsonrpc_set_id(request->impl, *id);
+		if (ret < 0)
+			goto free_impl;
+	}
+
+	ret = jsonrpc_set_method(request->impl, method);
+	if (ret < 0)
+		goto free_impl;
+
+	if (params) {
+		ret = jsonrpc_set_params(request->impl, params);
+		if (ret < 0)
+			goto free_impl;
+	}
+
+	*_request = request;
+	goto exit;
+
+free_impl:
+	json_object_put(request->impl);
+free:
+	free(request);
+exit:
+	return ret;
+}
+
+int jsonrpc_request_create(struct jsonrpc_request **_request, int id, const char *method,
+                           struct json_object *params)
+{
+	return jsonrpc_request_create_internal(_request, &id, method, params);
+}
+
+void jsonrpc_request_destroy(struct jsonrpc_request *request)
+{
+	json_object_put(request->impl);
+	free(request);
+}
+
+int jsonrpc_notification_create(struct jsonrpc_request **_request, const char *method,
+                                struct json_object *params)
+{
+	return jsonrpc_request_create_internal(_request, NULL, method, params);
+}
+
+int jsonrpc_request_is_notification(const struct jsonrpc_request *request)
+{
+	return !json_has(request->impl, jsonrpc_key_id);
+}
+
+const char *jsonrpc_request_to_string(struct jsonrpc_request *request)
+{
+	return json_to_string(request->impl);
+}
+
+static int jsonrpc_request_from_json(struct jsonrpc_request **_request, struct json_object *impl)
+{
+	int ret = 0;
+
+	ret = jsonrpc_check_version(impl);
+	if (ret < 0)
+		return ret;
+	ret = jsonrpc_check_id(impl, 0);
+	if (ret < 0)
+		return ret;
+	ret = jsonrpc_check_method(impl);
+	if (ret < 0)
+		return ret;
+	ret = jsonrpc_check_params(impl);
+	if (ret < 0)
+		return ret;
+
+	struct jsonrpc_request *request = malloc(sizeof(struct jsonrpc_request));
+	if (!request) {
+		log_errno("malloc");
+		return -1;
+	}
+	request->impl = impl;
+
+	*_request = request;
+	return ret;
+}
+
+int jsonrpc_request_parse(struct jsonrpc_request **_request, const char *src)
+{
+	struct json_object *impl = json_from_string(src);
+	if (!impl) {
+		log_err("JSON-RPC: failed to parse request\n");
+		return -1;
+	}
+
+	int ret = jsonrpc_request_from_json(_request, impl);
+	if (ret < 0)
+		goto free_impl;
+
+	return ret;
+
+free_impl:
+	json_object_put(impl);
+
+	return ret;
+}
+
+int jsonrpc_request_send(const struct jsonrpc_request *request, int fd)
+{
+	return json_send(request->impl, fd);
+}
+
+int jsonrpc_request_recv(struct jsonrpc_request **request, int fd)
+{
+	struct json_object *impl = json_recv(fd);
+	if (!impl) {
+		log_err("JSON-RPC: failed to receive request\n");
+		return -1;
+	}
+
+	int ret = jsonrpc_request_from_json(request, impl);
+	if (ret < 0)
+		goto free_impl;
+
+	return ret;
+
+free_impl:
+	json_object_put(impl);
+
+	return ret;
+}
+
+const char *jsonrpc_request_get_method(const struct jsonrpc_request *request)
+{
+	const char *method = NULL;
+	int ret = json_get_string(request->impl, jsonrpc_key_method, &method);
+	if (ret < 0) {
+		/* Should never happen. */
+		return NULL;
+	}
+	return method;
+}
+
+static struct json_object *jsonrpc_request_create_params(struct jsonrpc_request *request)
+{
+	const char *const key = jsonrpc_key_params;
+
+	if (!json_has(request->impl, key)) {
+		struct json_object *params = json_object_new_object();
+		if (!params) {
+			json_errno("json_object_new_object");
+			return NULL;
+		}
+		int ret = json_set(request->impl, key, params);
+		if (ret < 0) {
+			json_object_put(params);
+			return NULL;
+		}
+		return params;
+	}
+
+	struct json_object *params = NULL;
+	int ret = json_get(request->impl, key, &params);
+	if (ret < 0)
+		return NULL;
+	return params;
+}
+
+int jsonrpc_request_get_param_string(const struct jsonrpc_request *request, const char *name,
+                                     const char **value)
+{
+	struct json_object *params = NULL;
+	int ret = json_get(request->impl, jsonrpc_key_params, &params);
+	if (ret < 0)
+		return ret;
+	return json_get_string(params, name, value);
+}
+
+int jsonrpc_request_set_param_string(struct jsonrpc_request *request, const char *name,
+                                     const char *value)
+{
+	struct json_object *params = jsonrpc_request_create_params(request);
+	if (!params)
+		return -1;
+	return json_set_string(params, name, value);
+}
+
+int jsonrpc_request_get_param_int(const struct jsonrpc_request *request, const char *name,
+                                  int64_t *value)
+{
+	struct json_object *params = NULL;
+	int ret = json_get(request->impl, jsonrpc_key_params, &params);
+	if (ret < 0)
+		return ret;
+	return json_get_int(params, name, value);
+}
+
+int jsonrpc_request_set_param_int(struct jsonrpc_request *request, const char *name, int64_t value)
+{
+	struct json_object *params = jsonrpc_request_create_params(request);
+	if (!params)
+		return -1;
+	return json_set_int(params, name, value);
+}
+
+int jsonrpc_response_create_internal(struct jsonrpc_response **_response,
+                                     const struct jsonrpc_request *request,
+                                     struct json_object *result, struct json_object *error)
+{
+	int ret = 0;
+
+	struct jsonrpc_response *response = malloc(sizeof(struct jsonrpc_response));
+	if (!response) {
+		log_errno("malloc");
+		ret = -1;
+		goto exit;
+	}
+
+	response->impl = json_object_new_object();
+	if (!response->impl) {
+		json_errno("json_object_new_object");
+		ret = -1;
+		goto free;
+	}
+
+	ret = jsonrpc_set_version(response->impl);
+	if (ret < 0)
+		goto free_impl;
+
+	struct json_object *id = NULL;
+	ret = json_clone(request->impl, jsonrpc_key_id, &id);
+	if (ret < 0)
+		goto free_impl;
+
+	ret = json_set(response->impl, jsonrpc_key_id, id);
+	if (ret < 0) {
+		json_object_put(id);
+		goto free_impl;
+	}
+
+	if (error) {
+		ret = json_set_const_key(response->impl, jsonrpc_key_error, error);
+		if (ret < 0)
+			goto free_impl;
+	} else {
+		ret = json_set_const_key(response->impl, jsonrpc_key_result, result);
+		if (ret < 0)
+			goto free_impl;
+	}
+
+	*_response = response;
+	goto exit;
+
+free_impl:
+	json_object_put(response->impl);
+free:
+	free(response);
+exit:
+	return ret;
+}
+
+int jsonrpc_response_create(struct jsonrpc_response **response,
+                            const struct jsonrpc_request *request, struct json_object *result)
+{
+	return jsonrpc_response_create_internal(response, request, result, NULL);
+}
+
+void jsonrpc_response_destroy(struct jsonrpc_response *response)
+{
+	json_object_put(response->impl);
+	free(response);
+}
+
+int jsonrpc_error_create(struct jsonrpc_response **response, struct jsonrpc_request *request,
+                         int code, const char *message)
+{
+	struct json_object *error = json_object_new_object();
+	if (!error) {
+		json_errno("json_object_new_object");
+		return -1;
+	}
+
+	int ret = 0;
+
+	ret = json_set_int_const_key(error, jsonrpc_key_code, code);
+	if (ret < 0)
+		goto free;
+	ret = json_set_string_const_key(error, jsonrpc_key_message, message);
+	if (ret < 0)
+		goto free;
+
+	ret = jsonrpc_response_create_internal(response, request, NULL, error);
+	if (ret < 0)
+		goto free;
+
+	return ret;
+
+free:
+	json_object_put(error);
+
+	return ret;
+}
+
+int jsonrpc_response_is_error(const struct jsonrpc_response *response)
+{
+	return json_has(response->impl, jsonrpc_key_error);
+}
+
+const char *jsonrpc_response_to_string(struct jsonrpc_response *response)
+{
+	return json_to_string(response->impl);
+}
+
+static int jsonrpc_response_from_json(struct jsonrpc_response **_response, struct json_object *impl)
+{
+	int ret = 0;
+
+	ret = jsonrpc_check_version(impl);
+	if (ret < 0)
+		return ret;
+	ret = jsonrpc_check_id(impl, 1);
+	if (ret < 0)
+		return ret;
+	ret = jsonrpc_check_result_or_error(impl);
+	if (ret < 0)
+		return ret;
+
+	struct jsonrpc_response *response = malloc(sizeof(struct jsonrpc_response));
+	if (!response) {
+		log_errno("malloc");
+		return -1;
+	}
+	response->impl = impl;
+
+	*_response = response;
+	return ret;
+}
+
+int jsonrpc_response_parse(struct jsonrpc_response **_response, const char *src)
+{
+	struct json_object *impl = json_from_string(src);
+	if (!impl) {
+		log_err("JSON-RPC: failed to parse response\n");
+		return -1;
+	}
+
+	int ret = jsonrpc_response_from_json(_response, impl);
+	if (ret < 0)
+		goto free_impl;
+
+	return ret;
+
+free_impl:
+	json_object_put(impl);
+
+	return ret;
+}
+
+int jsonrpc_response_send(const struct jsonrpc_response *response, int fd)
+{
+	return json_send(response->impl, fd);
+}
+
+int jsonrpc_response_recv(struct jsonrpc_response **response, int fd)
+{
+	struct json_object *impl = json_recv(fd);
+	if (!impl) {
+		log_err("JSON-RPC: failed to receive response\n");
+		return -1;
+	}
+
+	int ret = jsonrpc_response_from_json(response, impl);
+	if (ret < 0)
+		goto free_impl;
+
+	return ret;
+
+free_impl:
+	json_object_put(impl);
+
+	return ret;
+}
diff --git a/src/json_rpc.h b/src/json_rpc.h
new file mode 100644
index 0000000..6e2be04
--- /dev/null
+++ b/src/json_rpc.h
@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+ * This file is part of the "cimple" project.
+ * For details, see https://github.com/egor-tensin/cimple.
+ * Distributed under the MIT License.
+ */
+
+#ifndef __JSON_RPC_H__
+#define __JSON_RPC_H__
+
+/* This attempts to adhere to the format described in https://www.jsonrpc.org/specification. */
+
+#include <json-c/json_object.h>
+
+#include <stdint.h>
+
+struct jsonrpc_request;
+
+int jsonrpc_request_create(struct jsonrpc_request **, int id, const char *method,
+                           struct json_object *params);
+void jsonrpc_request_destroy(struct jsonrpc_request *);
+
+int jsonrpc_notification_create(struct jsonrpc_request **, const char *method,
+                                struct json_object *params);
+int jsonrpc_request_is_notification(const struct jsonrpc_request *);
+
+const char *jsonrpc_request_to_string(struct jsonrpc_request *);
+int jsonrpc_request_parse(struct jsonrpc_request **, const char *src);
+
+int jsonrpc_request_send(const struct jsonrpc_request *, int fd);
+int jsonrpc_request_recv(struct jsonrpc_request **, int fd);
+
+const char *jsonrpc_request_get_method(const struct jsonrpc_request *);
+
+int jsonrpc_request_get_param_string(const struct jsonrpc_request *, const char *name,
+                                     const char **);
+int jsonrpc_request_set_param_string(struct jsonrpc_request *, const char *name, const char *);
+int jsonrpc_request_get_param_int(const struct jsonrpc_request *, const char *name, int64_t *);
+int jsonrpc_request_set_param_int(struct jsonrpc_request *, const char *name, int64_t);
+
+struct jsonrpc_response;
+
+int jsonrpc_response_create(struct jsonrpc_response **, const struct jsonrpc_request *,
+                            struct json_object *result);
+void jsonrpc_response_destroy(struct jsonrpc_response *);
+
+int jsonrpc_error_create(struct jsonrpc_response **, struct jsonrpc_request *, int code,
+                         const char *message);
+int jsonrpc_response_is_error(const struct jsonrpc_response *);
+
+const char *jsonrpc_response_to_string(struct jsonrpc_response *);
+int jsonrpc_response_parse(struct jsonrpc_response **, const char *src);
+
+int jsonrpc_response_send(const struct jsonrpc_response *, int fd);
+int jsonrpc_response_recv(struct jsonrpc_response **, int fd);
+
+#endif
diff --git a/src/msg.c b/src/msg.c
deleted file mode 100644
index 0de8a13..0000000
--- a/src/msg.c
+++ /dev/null
@@ -1,308 +0,0 @@
-/*
- * Copyright (c) 2022 Egor Tensin <Egor.Tensin@gmail.com>
- * This file is part of the "cimple" project.
- * For details, see https://github.com/egor-tensin/cimple.
- * Distributed under the MIT License.
- */
-
-#include "msg.h"
-#include "log.h"
-#include "net.h"
-
-#include <stdlib.h>
-#include <string.h>
-
-struct msg {
-	size_t argc;
-	const char **argv;
-};
-
-size_t msg_get_length(const struct msg *msg)
-{
-	return msg->argc;
-}
-
-const char **msg_get_strings(const struct msg *msg)
-{
-	return msg->argv;
-}
-
-const char *msg_get_first_string(const struct msg *msg)
-{
-	return msg->argv[0];
-}
-
-int msg_success(struct msg **msg)
-{
-	static const char *argv[] = {"success", NULL};
-	return msg_from_argv(msg, argv);
-}
-
-int msg_error(struct msg **msg)
-{
-	static const char *argv[] = {"error", NULL};
-	return msg_from_argv(msg, argv);
-}
-
-int msg_is_success(const struct msg *msg)
-{
-	return msg->argc == 1 && !strcmp(msg->argv[0], "success");
-}
-
-int msg_is_error(const struct msg *msg)
-{
-	return msg->argc == 1 && !strcmp(msg->argv[0], "error");
-}
-
-static int msg_copy_argv(struct msg *msg, const char **argv)
-{
-	size_t copied = 0;
-
-	msg->argv = calloc(msg->argc + 1, sizeof(const char *));
-	if (!msg->argv) {
-		log_errno("calloc");
-		return -1;
-	}
-
-	for (copied = 0; copied < msg->argc; ++copied) {
-		msg->argv[copied] = strdup(argv[copied]);
-		if (!msg->argv[copied]) {
-			log_errno("strdup");
-			goto free_copied;
-		}
-	}
-
-	return 0;
-
-free_copied:
-	for (size_t i = 0; i < copied; ++i) {
-		free((char *)msg->argv[i]);
-	}
-
-	free(msg->argv);
-
-	return -1;
-}
-
-int msg_copy(struct msg **_dest, const struct msg *src)
-{
-	int ret = 0;
-
-	struct msg *dest = malloc(sizeof(struct msg));
-	if (!dest) {
-		log_errno("malloc");
-		return -1;
-	}
-
-	dest->argc = src->argc;
-
-	ret = msg_copy_argv(dest, (const char **)src->argv);
-	if (ret < 0)
-		goto free;
-
-	*_dest = dest;
-	return 0;
-
-free:
-	free(dest);
-
-	return -1;
-}
-
-void msg_free(struct msg *msg)
-{
-	for (size_t i = 0; i < msg->argc; ++i)
-		free((char *)msg->argv[i]);
-	free(msg->argv);
-	free(msg);
-}
-
-int msg_from_argv(struct msg **_msg, const char **argv)
-{
-	int ret = 0;
-
-	struct msg *msg = malloc(sizeof(struct msg));
-	if (!msg) {
-		log_errno("malloc");
-		return -1;
-	}
-
-	msg->argc = 0;
-	for (const char **s = argv; *s; ++s)
-		++msg->argc;
-
-	if (!msg->argc) {
-		log_err("A message must contain at least one string\n");
-		goto free;
-	}
-
-	ret = msg_copy_argv(msg, argv);
-	if (ret < 0)
-		goto free;
-
-	*_msg = msg;
-	return 0;
-
-free:
-	free(msg);
-
-	return -1;
-}
-
-int msg_send(int fd, const struct msg *msg)
-{
-	struct buf *buf = NULL;
-	int ret = 0;
-
-	ret = buf_pack_strings(&buf, msg->argc, msg->argv);
-	if (ret < 0)
-		return ret;
-
-	ret = net_send_buf(fd, buf);
-	if (ret < 0)
-		goto destroy_buf;
-
-destroy_buf:
-	buf_destroy(buf);
-
-	return ret;
-}
-
-int msg_send_argv(int fd, const char **argv)
-{
-	struct msg *msg = NULL;
-	int ret = 0;
-
-	ret = msg_from_argv(&msg, argv);
-	if (ret < 0)
-		return ret;
-
-	ret = msg_send(fd, msg);
-	msg_free(msg);
-	if (ret < 0)
-		return ret;
-
-	return ret;
-}
-
-int msg_recv(int fd, struct msg **_msg)
-{
-	struct buf *buf = NULL;
-	int ret = 0;
-
-	ret = net_recv_buf(fd, &buf);
-	if (ret < 0)
-		return ret;
-
-	struct msg *msg = malloc(sizeof(struct msg));
-	if (!msg) {
-		log_errno("malloc");
-		ret = -1;
-		goto destroy_buf;
-	}
-
-	ret = buf_unpack_strings(buf, &msg->argc, &msg->argv);
-	if (ret < 0)
-		goto free_msg;
-
-	if (!msg->argc) {
-		log_err("A message must contain at least one string\n");
-		goto free_msg;
-	}
-
-	*_msg = msg;
-	goto destroy_buf;
-
-free_msg:
-	free(msg);
-
-destroy_buf:
-	buf_destroy(buf);
-
-	return ret;
-}
-
-int msg_talk(int fd, const struct msg *request, struct msg **response)
-{
-	int ret = 0;
-
-	if (!request && !response) {
-		log_err("For communication, there must be at least a request and/or response\n");
-		return -1;
-	}
-
-	if (request) {
-		ret = msg_send(fd, request);
-		if (ret < 0)
-			return ret;
-	}
-
-	if (response) {
-		ret = msg_recv(fd, response);
-		if (ret < 0)
-			return ret;
-	}
-
-	return ret;
-}
-
-int msg_talk_argv(int fd, const char **argv, struct msg **response)
-{
-	struct msg *request = NULL;
-	int ret = 0;
-
-	ret = msg_from_argv(&request, argv);
-	if (ret < 0)
-		return ret;
-
-	ret = msg_talk(fd, request, response);
-	msg_free(request);
-	if (ret < 0)
-		return ret;
-
-	return ret;
-}
-
-int msg_connect_and_talk(const char *host, const char *port, const struct msg *request,
-                         struct msg **response)
-{
-	int fd = -1, ret = 0;
-
-	fd = net_connect(host, port);
-	if (fd < 0)
-		return fd;
-
-	ret = msg_talk(fd, request, response);
-	if (ret < 0)
-		goto close;
-
-close:
-	net_close(fd);
-
-	return ret;
-}
-
-int msg_connect_and_talk_argv(const char *host, const char *port, const char **argv,
-                              struct msg **response)
-{
-	struct msg *request = NULL;
-	int ret = 0;
-
-	ret = msg_from_argv(&request, argv);
-	if (ret < 0)
-		return ret;
-
-	ret = msg_connect_and_talk(host, port, request, response);
-	msg_free(request);
-	if (ret < 0)
-		return ret;
-
-	return ret;
-}
-
-void msg_dump(const struct msg *msg)
-{
-	log("Message[%zu]:\n", msg->argc);
-	for (size_t i = 0; i < msg->argc; ++i)
-		log("\t%s\n", msg->argv[i]);
-}
diff --git a/src/msg.h b/src/msg.h
deleted file mode 100644
index 44ebf44..0000000
--- a/src/msg.h
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Copyright (c) 2022 Egor Tensin <Egor.Tensin@gmail.com>
- * This file is part of the "cimple" project.
- * For details, see https://github.com/egor-tensin/cimple.
- * Distributed under the MIT License.
- */
-
-#ifndef __MSG_H__
-#define __MSG_H__
-
-#include <stddef.h>
-
-struct msg;
-
-int msg_from_argv(struct msg **, const char **argv);
-void msg_free(struct msg *);
-
-int msg_copy(struct msg **, const struct msg *);
-
-size_t msg_get_length(const struct msg *);
-const char **msg_get_strings(const struct msg *);
-const char *msg_get_first_string(const struct msg *);
-
-int msg_success(struct msg **);
-int msg_error(struct msg **);
-
-int msg_is_success(const struct msg *);
-int msg_is_error(const struct msg *);
-
-int msg_recv(int fd, struct msg **);
-
-int msg_send(int fd, const struct msg *);
-int msg_send_argv(int fd, const char **argv);
-
-int msg_talk(int fd, const struct msg *, struct msg **response);
-int msg_talk_argv(int fd, const char **argv, struct msg **response);
-
-int msg_connect_and_talk(const char *host, const char *port, const struct msg *, struct msg **);
-int msg_connect_and_talk_argv(const char *host, const char *port, const char **argv,
-                              struct msg **response);
-
-void msg_dump(const struct msg *);
-
-#endif
diff --git a/src/protocol.c b/src/protocol.c
index 5d1903a..8ee57e1 100644
--- a/src/protocol.c
+++ b/src/protocol.c
@@ -7,132 +7,181 @@
 
 #include "protocol.h"
 #include "base64.h"
+#include "compiler.h"
 #include "const.h"
-#include "log.h"
-#include "msg.h"
+#include "json_rpc.h"
 #include "process.h"
 #include "run_queue.h"
-#include "string.h"
 
 #include <stddef.h>
-#include <stdio.h>
+#include <stdint.h>
 #include <stdlib.h>
 
-static int check_msg_length(const struct msg *msg, size_t expected)
+static const char *const run_key_id = "id";
+static const char *const run_key_url = "url";
+static const char *const run_key_rev = "rev";
+
+int run_request_create(struct jsonrpc_request **request, const struct run *run)
 {
-	size_t actual = msg_get_length(msg);
+	int ret = 0;
 
-	if (actual != expected) {
-		log_err("Invalid number of arguments for a message: %zu\n", actual);
-		msg_dump(msg);
-		return -1;
-	}
+	ret = jsonrpc_request_create(request, 1, CMD_RUN, NULL);
+	if (ret < 0)
+		return ret;
+	ret = jsonrpc_request_set_param_string(*request, run_key_url, run_get_url(run));
+	if (ret < 0)
+		goto free_request;
+	ret = jsonrpc_request_set_param_string(*request, run_key_rev, run_get_rev(run));
+	if (ret < 0)
+		goto free_request;
 
-	return 0;
+	return ret;
+
+free_request:
+	jsonrpc_request_destroy(*request);
+
+	return ret;
 }
 
-int msg_run_parse(const struct msg *msg, struct run **run)
+int run_request_parse(const struct jsonrpc_request *request, struct run **run)
 {
-	int ret = check_msg_length(msg, 3);
+	int ret = 0;
+
+	const char *url = NULL;
+	ret = jsonrpc_request_get_param_string(request, run_key_url, &url);
+	if (ret < 0)
+		return ret;
+	const char *rev = NULL;
+	ret = jsonrpc_request_get_param_string(request, run_key_rev, &rev);
 	if (ret < 0)
 		return ret;
 
-	const char **argv = msg_get_strings(msg);
-	/* We don't know the ID yet. */
-	return run_create(run, 0, argv[1], argv[2]);
+	return run_create(run, 0, url, rev);
 }
 
-int msg_new_worker_create(struct msg **msg)
+int new_worker_request_create(struct jsonrpc_request **request)
 {
-	static const char *argv[] = {CMD_NEW_WORKER, NULL};
-	return msg_from_argv(msg, argv);
+	return jsonrpc_notification_create(request, CMD_NEW_WORKER, NULL);
 }
 
-int msg_start_create(struct msg **msg, const struct run *run)
+int new_worker_request_parse(UNUSED const struct jsonrpc_request *request)
 {
-	char id[16];
-	snprintf(id, sizeof(id), "%d", run_get_id(run));
-
-	const char *argv[] = {CMD_START, id, run_get_url(run), run_get_rev(run), NULL};
-
-	return msg_from_argv(msg, argv);
+	return 0;
 }
 
-int msg_start_parse(const struct msg *msg, struct run **run)
+int start_request_create(struct jsonrpc_request **request, const struct run *run)
 {
 	int ret = 0;
 
-	ret = check_msg_length(msg, 4);
+	ret = jsonrpc_notification_create(request, CMD_START, NULL);
 	if (ret < 0)
 		return ret;
+	ret = jsonrpc_request_set_param_int(*request, run_key_id, run_get_id(run));
+	if (ret < 0)
+		goto free_request;
+	ret = jsonrpc_request_set_param_string(*request, run_key_url, run_get_url(run));
+	if (ret < 0)
+		goto free_request;
+	ret = jsonrpc_request_set_param_string(*request, run_key_rev, run_get_rev(run));
+	if (ret < 0)
+		goto free_request;
+
+	return ret;
 
-	const char **argv = msg_get_strings(msg);
+free_request:
+	jsonrpc_request_destroy(*request);
+
+	return ret;
+}
 
-	int id = 0;
+int start_request_parse(const struct jsonrpc_request *request, struct run **run)
+{
+	int ret = 0;
 
-	ret = string_to_int(argv[1], &id);
+	int64_t id = 0;
+	ret = jsonrpc_request_get_param_int(request, run_key_id, &id);
+	if (ret < 0)
+		return ret;
+	const char *url = NULL;
+	ret = jsonrpc_request_get_param_string(request, run_key_url, &url);
+	if (ret < 0)
+		return ret;
+	const char *rev = NULL;
+	ret = jsonrpc_request_get_param_string(request, run_key_rev, &rev);
 	if (ret < 0)
 		return ret;
 
-	return run_create(run, id, argv[2], argv[3]);
+	return run_create(run, (int)id, url, rev);
 }
 
-int msg_finished_create(struct msg **msg, int run_id, const struct proc_output *output)
+static const char *const finished_key_run_id = "run_id";
+static const char *const finished_key_ec = "exit_code";
+static const char *const finished_key_data = "output";
+
+int finished_request_create(struct jsonrpc_request **request, int run_id,
+                            const struct proc_output *output)
 {
 	int ret = 0;
 
-	char id[16];
-	char ec[16];
-
-	snprintf(id, sizeof(id), "%d", run_id);
-	snprintf(ec, sizeof(ec), "%d", output->ec);
+	ret = jsonrpc_notification_create(request, CMD_FINISHED, NULL);
+	if (ret < 0)
+		return ret;
+	ret = jsonrpc_request_set_param_int(*request, finished_key_run_id, run_id);
+	if (ret < 0)
+		goto free_request;
+	ret = jsonrpc_request_set_param_int(*request, finished_key_ec, output->ec);
+	if (ret < 0)
+		goto free_request;
 
 	char *b64data = NULL;
-
 	ret = base64_encode(output->data, output->data_size, &b64data);
 	if (ret < 0)
-		return ret;
-
-	const char *argv[] = {CMD_FINISHED, id, ec, b64data, NULL};
+		goto free_request;
 
-	ret = msg_from_argv(msg, argv);
+	ret = jsonrpc_request_set_param_string(*request, finished_key_data, b64data);
+	free(b64data);
 	if (ret < 0)
-		goto free_b64data;
+		goto free_request;
 
-free_b64data:
-	free(b64data);
+	return ret;
+
+free_request:
+	jsonrpc_request_destroy(*request);
 
 	return ret;
 }
 
-int msg_finished_parse(const struct msg *msg, int *run_id, struct proc_output **_output)
+int finished_request_parse(const struct jsonrpc_request *request, int *_run_id,
+                           struct proc_output **_output)
 {
 	int ret = 0;
 
-	ret = check_msg_length(msg, 4);
-	if (ret < 0)
-		return ret;
-
-	const char **argv = msg_get_strings(msg);
-
 	struct proc_output *output = NULL;
 	ret = proc_output_create(&output);
 	if (ret < 0)
 		return ret;
 
-	ret = string_to_int(argv[1], run_id);
+	int64_t run_id = 0;
+	ret = jsonrpc_request_get_param_int(request, finished_key_run_id, &run_id);
 	if (ret < 0)
 		goto free_output;
-	ret = string_to_int(argv[2], &output->ec);
+
+	int64_t ec = -1;
+	ret = jsonrpc_request_get_param_int(request, finished_key_ec, &ec);
 	if (ret < 0)
 		goto free_output;
+	output->ec = (int)ec;
 
-	const char *b64data = argv[3];
+	const char *b64data = NULL;
+	ret = jsonrpc_request_get_param_string(request, finished_key_data, &b64data);
+	if (ret < 0)
+		goto free_output;
 
 	ret = base64_decode(b64data, &output->data, &output->data_size);
 	if (ret < 0)
 		goto free_output;
 
+	*_run_id = (int)run_id;
 	*_output = output;
 	return ret;
 
diff --git a/src/protocol.h b/src/protocol.h
index 99ec3ee..765e6cf 100644
--- a/src/protocol.h
+++ b/src/protocol.h
@@ -8,18 +8,20 @@
 #ifndef __PROTOCOL_H__
 #define __PROTOCOL_H__
 
-#include "msg.h"
+#include "json_rpc.h"
 #include "process.h"
 #include "run_queue.h"
 
-int msg_run_parse(const struct msg *, struct run **);
+int run_request_create(struct jsonrpc_request **, const struct run *);
+int run_request_parse(const struct jsonrpc_request *, struct run **);
 
-int msg_new_worker_create(struct msg **);
+int new_worker_request_create(struct jsonrpc_request **);
+int new_worker_request_parse(const struct jsonrpc_request *);
 
-int msg_start_create(struct msg **, const struct run *);
-int msg_start_parse(const struct msg *, struct run **);
+int start_request_create(struct jsonrpc_request **, const struct run *);
+int start_request_parse(const struct jsonrpc_request *, struct run **);
 
-int msg_finished_create(struct msg **, int run_id, const struct proc_output *);
-int msg_finished_parse(const struct msg *, int *run_id, struct proc_output **);
+int finished_request_create(struct jsonrpc_request **, int run_id, const struct proc_output *);
+int finished_request_parse(const struct jsonrpc_request *, int *run_id, struct proc_output **);
 
 #endif
diff --git a/src/server.c b/src/server.c
index 8e2fab3..61e3136 100644
--- a/src/server.c
+++ b/src/server.c
@@ -11,8 +11,8 @@
 #include "const.h"
 #include "event_loop.h"
 #include "file.h"
+#include "json_rpc.h"
 #include "log.h"
-#include "msg.h"
 #include "net.h"
 #include "process.h"
 #include "protocol.h"
@@ -169,15 +169,15 @@ static void server_assign_run(struct server *server)
 	struct worker *worker = worker_queue_remove_first(&server->worker_queue);
 	log("Removed worker %d from the queue\n", worker_get_fd(worker));
 
-	struct msg *start_msg = NULL;
+	struct jsonrpc_request *start_request = NULL;
 	int ret = 0;
 
-	ret = msg_start_create(&start_msg, run);
+	ret = start_request_create(&start_request, run);
 	if (ret < 0)
 		goto exit;
 
-	ret = msg_talk(worker_get_fd(worker), start_msg, NULL);
-	msg_free(start_msg);
+	ret = jsonrpc_request_send(start_request, worker_get_fd(worker));
+	jsonrpc_request_destroy(start_request);
 	if (ret < 0)
 		goto exit;
 
@@ -222,8 +222,8 @@ exit:
 	return NULL;
 }
 
-static int server_handle_cmd_new_worker(UNUSED const struct msg *request,
-                                        UNUSED struct msg **response, void *_ctx)
+static int server_handle_cmd_new_worker(UNUSED const struct jsonrpc_request *request,
+                                        UNUSED struct jsonrpc_response **response, void *_ctx)
 {
 	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
 	struct server *server = (struct server *)ctx->arg;
@@ -255,7 +255,8 @@ close:
 	return ret;
 }
 
-static int server_handle_cmd_run(const struct msg *request, struct msg **response, void *_ctx)
+static int server_handle_cmd_run(const struct jsonrpc_request *request,
+                                 struct jsonrpc_response **response, void *_ctx)
 {
 	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
 	struct server *server = (struct server *)ctx->arg;
@@ -263,11 +264,11 @@ static int server_handle_cmd_run(const struct msg *request, struct msg **respons
 
 	struct run *run = NULL;
 
-	ret = msg_run_parse(request, &run);
+	ret = run_request_parse(request, &run);
 	if (ret < 0)
 		return ret;
 
-	ret = msg_success(response);
+	ret = jsonrpc_response_create(response, request, NULL);
 	if (ret < 0)
 		goto destroy_run;
 
@@ -278,7 +279,7 @@ static int server_handle_cmd_run(const struct msg *request, struct msg **respons
 	return ret;
 
 free_response:
-	msg_free(*response);
+	jsonrpc_response_destroy(*response);
 	*response = NULL;
 
 destroy_run:
@@ -287,8 +288,8 @@ destroy_run:
 	return ret;
 }
 
-static int server_handle_cmd_finished(const struct msg *request, UNUSED struct msg **response,
-                                      void *_ctx)
+static int server_handle_cmd_finished(const struct jsonrpc_request *request,
+                                      UNUSED struct jsonrpc_response **response, void *_ctx)
 {
 	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
 	struct server *server = (struct server *)ctx->arg;
@@ -297,7 +298,7 @@ static int server_handle_cmd_finished(const struct msg *request, UNUSED struct m
 	int run_id = 0;
 	struct proc_output *output;
 
-	ret = msg_finished_parse(request, &run_id, &output);
+	ret = finished_request_parse(request, &run_id, &output);
 	if (ret < 0)
 		return ret;
 
diff --git a/src/worker.c b/src/worker.c
index e45bfba..a6ccc89 100644
--- a/src/worker.c
+++ b/src/worker.c
@@ -13,7 +13,6 @@
 #include "event_loop.h"
 #include "git.h"
 #include "log.h"
-#include "msg.h"
 #include "net.h"
 #include "process.h"
 #include "protocol.h"
@@ -83,14 +82,14 @@ static int worker_set_stopping(UNUSED struct event_loop *loop, UNUSED int fd, UN
 	return 0;
 }
 
-static int worker_handle_cmd_start(const struct msg *request, UNUSED struct msg **response,
-                                   void *_ctx)
+static int worker_handle_cmd_start(const struct jsonrpc_request *request,
+                                   UNUSED struct jsonrpc_response **response, void *_ctx)
 {
 	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
 	struct worker *worker = (struct worker *)ctx->arg;
 	int ret = 0;
 
-	ret = msg_start_parse(request, &worker->run);
+	ret = start_request_parse(request, &worker->run);
 	if (ret < 0)
 		return ret;
 
@@ -193,19 +192,26 @@ static int worker_do_run(struct worker *worker)
 
 	proc_output_dump(result);
 
-	struct msg *finished_msg = NULL;
+	struct jsonrpc_request *finished_request = NULL;
 
-	ret = msg_finished_create(&finished_msg, run_get_id(worker->run), result);
+	ret = finished_request_create(&finished_request, run_get_id(worker->run), result);
 	if (ret < 0)
 		goto free_output;
 
-	ret = msg_connect_and_talk(worker->settings->host, worker->settings->port, finished_msg,
-	                           NULL);
+	ret = net_connect(worker->settings->host, worker->settings->port);
+	if (ret < 0)
+		goto free_request;
+	int fd = ret;
+
+	ret = jsonrpc_request_send(finished_request, fd);
 	if (ret < 0)
-		goto free_finished_msg;
+		goto close_conn;
 
-free_finished_msg:
-	msg_free(finished_msg);
+close_conn:
+	net_close(fd);
+
+free_request:
+	jsonrpc_request_destroy(finished_request);
 
 free_output:
 	proc_output_destroy(result);
@@ -224,14 +230,13 @@ static int worker_get_run(struct worker *worker)
 		return ret;
 	fd = ret;
 
-	struct msg *new_worker_msg = NULL;
-
-	ret = msg_new_worker_create(&new_worker_msg);
+	struct jsonrpc_request *new_worker_request = NULL;
+	ret = new_worker_request_create(&new_worker_request);
 	if (ret < 0)
 		goto close;
 
-	ret = msg_send(fd, new_worker_msg);
-	msg_free(new_worker_msg);
+	ret = jsonrpc_request_send(new_worker_request, fd);
+	jsonrpc_request_destroy(new_worker_request);
 	if (ret < 0)
 		goto close;
 
diff --git a/test/py/test_basic.py b/test/py/test_basic.py
index 7ac038c..eda12d0 100644
--- a/test/py/test_basic.py
+++ b/test/py/test_basic.py
@@ -47,18 +47,18 @@ def test_cmd_line_invalid_option(server_exe, worker_exe, client_exe):
     _test_cmd_line_invalid_option_internal(client_exe, 'client')
 
 
-def test_run_client_no_msg(client):
+def test_run_client_no_action(client):
     ec, output = client.try_run()
     assert ec != 0, f'Invalid exit code {ec}, output:\n{output}'
-    prefix = 'usage error: no message to send to the server\n'
+    prefix = 'usage error: no action specified\n'
     assert output.startswith(prefix), f'Invalid output:\n{output}'
 
 
-def test_run_client_invalid_msg(server, client):
+def test_run_client_invalid_request(server, client):
     ec, output = client.try_run('hello')
     assert ec != 0, f'Invalid exit code {ec}, output:\n{output}'
-    suffix = "Failed to connect to server or it couldn't process the request\n"
-    assert output.endswith(suffix), f'Invalid output:\n{output}'
+    prefix = 'usage error: invalid request\n'
+    assert output.startswith(prefix), f'Invalid output:\n{output}'
 
 
 def test_run_noop_server(server):