diff --git a/frappe/__init__.py b/frappe/__init__.py
index 43c33d436b..0b9b340c8d 100644
--- a/frappe/__init__.py
+++ b/frappe/__init__.py
@@ -169,7 +169,7 @@ lang = local("lang")
 
 # This if block is never executed when running the code. It is only used for
 # telling static code analyzer where to find dynamically defined attributes.
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: no cover
 	from werkzeug.wrappers import Request
 
 	from frappe.database.mariadb.database import MariaDBDatabase
@@ -488,9 +488,12 @@ def msgprint(
 	def _raise_exception():
 		if raise_exception:
 			if inspect.isclass(raise_exception) and issubclass(raise_exception, Exception):
-				raise raise_exception(msg)
+				exc = raise_exception(msg)
 			else:
-				raise ValidationError(msg)
+				exc = ValidationError(msg)
+			if out.__frappe_exc_id:
+				exc.__frappe_exc_id = out.__frappe_exc_id
+			raise exc
 
 	if flags.mute_messages:
 		_raise_exception()
@@ -527,6 +530,7 @@ def msgprint(
 
 	if raise_exception:
 		out.raise_exception = 1
+		out.__frappe_exc_id = generate_hash()
 
 	if primary_action:
 		out.primary_action = primary_action
@@ -534,11 +538,7 @@ def msgprint(
 	if wide:
 		out.wide = wide
 
-	message_log.append(json.dumps(out))
-
-	if raise_exception and hasattr(raise_exception, "__name__"):
-		local.response["exc_type"] = raise_exception.__name__
-
+	message_log.append(out)
 	_raise_exception()
 
 
diff --git a/frappe/api.py b/frappe/api.py
deleted file mode 100644
index 084bee060b..0000000000
--- a/frappe/api.py
+++ /dev/null
@@ -1,306 +0,0 @@
-# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
-# License: MIT. See LICENSE
-import base64
-import binascii
-import json
-from typing import Literal
-from urllib.parse import urlencode, urlparse
-
-import frappe
-import frappe.client
-import frappe.handler
-from frappe import _
-from frappe.utils.data import sbool
-from frappe.utils.response import build_response
-
-
-def handle():
-	"""
-	Handler for `/api` methods
-
-	### Examples:
-
-	`/api/method/{methodname}` will call a whitelisted method
-
-	`/api/resource/{doctype}` will query a table
-	        examples:
-	        - `?fields=["name", "owner"]`
-	        - `?filters=[["Task", "name", "like", "%005"]]`
-	        - `?limit_start=0`
-	        - `?limit_page_length=20`
-
-	`/api/resource/{doctype}/{name}` will point to a resource
-	        `GET` will return doclist
-	        `POST` will insert
-	        `PUT` will update
-	        `DELETE` will delete
-
-	`/api/resource/{doctype}/{name}?run_method={method}` will run a whitelisted controller method
-	"""
-
-	parts = frappe.request.path[1:].split("/", 3)
-	call = doctype = name = None
-
-	if len(parts) > 1:
-		call = parts[1]
-
-	if len(parts) > 2:
-		doctype = parts[2]
-
-	if len(parts) > 3:
-		name = parts[3]
-
-	return _RESTAPIHandler(call, doctype, name).get_response()
-
-
-class _RESTAPIHandler:
-	def __init__(self, call: Literal["method", "resource"], doctype: str | None, name: str | None):
-		self.call = call
-		self.doctype = doctype
-		self.name = name
-
-	def get_response(self):
-		"""Prepare and get response based on URL and form body.
-
-		Note: most methods of this class directly operate on the response local.
-		"""
-		match self.call:
-			case "method":
-				return self.handle_method()
-			case "resource":
-				self.handle_resource()
-			case _:
-				raise frappe.DoesNotExistError
-
-		return build_response("json")
-
-	def handle_method(self):
-		frappe.local.form_dict.cmd = self.doctype
-		return frappe.handler.handle()
-
-	def handle_resource(self):
-		if self.doctype and self.name:
-			self.handle_document_resource()
-		elif self.doctype:
-			self.handle_doctype_resource()
-		else:
-			raise frappe.DoesNotExistError
-
-	def handle_document_resource(self):
-		if "run_method" in frappe.local.form_dict:
-			self.execute_doc_method()
-			return
-
-		match frappe.local.request.method:
-			case "GET":
-				self.get_doc()
-			case "PUT":
-				self.update_doc()
-			case "DELETE":
-				self.delete_doc()
-			case _:
-				raise frappe.DoesNotExistError
-
-	def handle_doctype_resource(self):
-		match frappe.local.request.method:
-			case "GET":
-				self.get_doc_list()
-			case "POST":
-				self.create_doc()
-			case _:
-				raise frappe.DoesNotExistError
-
-	def execute_doc_method(self):
-		method = frappe.local.form_dict.pop("run_method")
-		doc = frappe.get_doc(self.doctype, self.name)
-		doc.is_whitelisted(method)
-
-		if frappe.local.request.method == "GET":
-			if not doc.has_permission("read"):
-				frappe.throw(_("Not permitted"), frappe.PermissionError)
-			frappe.local.response.update({"data": doc.run_method(method, **frappe.local.form_dict)})
-
-		elif frappe.local.request.method == "POST":
-			if not doc.has_permission("write"):
-				frappe.throw(_("Not permitted"), frappe.PermissionError)
-
-			frappe.local.response.update({"data": doc.run_method(method, **frappe.local.form_dict)})
-			frappe.db.commit()
-
-	def get_doc(self):
-		doc = frappe.get_doc(self.doctype, self.name)
-		if not doc.has_permission("read"):
-			raise frappe.PermissionError
-		doc.apply_fieldlevel_read_permissions()
-		frappe.local.response.update({"data": doc})
-
-	def update_doc(self):
-		data = get_request_form_data()
-
-		doc = frappe.get_doc(self.doctype, self.name, for_update=True)
-
-		if "flags" in data:
-			del data["flags"]
-
-		# Not checking permissions here because it's checked in doc.save
-		doc.update(data)
-
-		frappe.local.response.update({"data": doc.save().as_dict()})
-
-		# check for child table doctype
-		if doc.get("parenttype"):
-			frappe.get_doc(doc.parenttype, doc.parent).save()
-		frappe.db.commit()
-
-	def delete_doc(self):
-		# Not checking permissions here because it's checked in delete_doc
-		frappe.delete_doc(self.doctype, self.name, ignore_missing=False)
-		frappe.local.response.http_status_code = 202
-		frappe.local.response.message = "ok"
-		frappe.db.commit()
-
-	def get_doc_list(self):
-		if frappe.local.form_dict.get("fields"):
-			frappe.local.form_dict["fields"] = json.loads(frappe.local.form_dict["fields"])
-
-		# set limit of records for frappe.get_list
-		frappe.local.form_dict.setdefault(
-			"limit_page_length",
-			frappe.local.form_dict.limit or frappe.local.form_dict.limit_page_length or 20,
-		)
-
-		# convert strings to native types - only as_dict and debug accept bool
-		for param in ["as_dict", "debug"]:
-			param_val = frappe.local.form_dict.get(param)
-			if param_val is not None:
-				frappe.local.form_dict[param] = sbool(param_val)
-
-		# evaluate frappe.get_list
-		data = frappe.call(frappe.client.get_list, self.doctype, **frappe.local.form_dict)
-
-		# set frappe.get_list result to response
-		frappe.local.response.update({"data": data})
-
-	def create_doc(self):
-		data = get_request_form_data()
-		data.update({"doctype": self.doctype})
-
-		# insert document from request data
-		doc = frappe.get_doc(data).insert()
-
-		# set response data
-		frappe.local.response.update({"data": doc.as_dict()})
-
-		# commit for POST requests
-		frappe.db.commit()
-
-
-def get_request_form_data():
-	if frappe.local.form_dict.data is None:
-		data = frappe.safe_decode(frappe.local.request.get_data())
-	else:
-		data = frappe.local.form_dict.data
-
-	try:
-		return frappe.parse_json(data)
-	except ValueError:
-		return frappe.local.form_dict
-
-
-def validate_auth():
-	"""
-	Authenticate and sets user for the request.
-	"""
-	authorization_header = frappe.get_request_header("Authorization", "").split(" ")
-
-	if len(authorization_header) == 2:
-		validate_oauth(authorization_header)
-		validate_auth_via_api_keys(authorization_header)
-
-	validate_auth_via_hooks()
-
-
-def validate_oauth(authorization_header):
-	"""
-	Authenticate request using OAuth and set session user
-
-	Args:
-	        authorization_header (list of str): The 'Authorization' header containing the prefix and token
-	"""
-
-	from frappe.integrations.oauth2 import get_oauth_server
-	from frappe.oauth import get_url_delimiter
-
-	form_dict = frappe.local.form_dict
-	token = authorization_header[1]
-	req = frappe.request
-	parsed_url = urlparse(req.url)
-	access_token = {"access_token": token}
-	uri = (
-		parsed_url.scheme + "://" + parsed_url.netloc + parsed_url.path + "?" + urlencode(access_token)
-	)
-	http_method = req.method
-	headers = req.headers
-	body = req.get_data()
-	if req.content_type and "multipart/form-data" in req.content_type:
-		body = None
-
-	try:
-		required_scopes = frappe.db.get_value("OAuth Bearer Token", token, "scopes").split(
-			get_url_delimiter()
-		)
-		valid, oauthlib_request = get_oauth_server().verify_request(
-			uri, http_method, body, headers, required_scopes
-		)
-		if valid:
-			frappe.set_user(frappe.db.get_value("OAuth Bearer Token", token, "user"))
-			frappe.local.form_dict = form_dict
-	except AttributeError:
-		pass
-
-
-def validate_auth_via_api_keys(authorization_header):
-	"""
-	Authenticate request using API keys and set session user
-
-	Args:
-	        authorization_header (list of str): The 'Authorization' header containing the prefix and token
-	"""
-
-	try:
-		auth_type, auth_token = authorization_header
-		authorization_source = frappe.get_request_header("Frappe-Authorization-Source")
-		if auth_type.lower() == "basic":
-			api_key, api_secret = frappe.safe_decode(base64.b64decode(auth_token)).split(":")
-			validate_api_key_secret(api_key, api_secret, authorization_source)
-		elif auth_type.lower() == "token":
-			api_key, api_secret = auth_token.split(":")
-			validate_api_key_secret(api_key, api_secret, authorization_source)
-	except binascii.Error:
-		frappe.throw(
-			_("Failed to decode token, please provide a valid base64-encoded token."),
-			frappe.InvalidAuthorizationToken,
-		)
-	except (AttributeError, TypeError, ValueError):
-		pass
-
-
-def validate_api_key_secret(api_key, api_secret, frappe_authorization_source=None):
-	"""frappe_authorization_source to provide api key and secret for a doctype apart from User"""
-	doctype = frappe_authorization_source or "User"
-	doc = frappe.db.get_value(doctype=doctype, filters={"api_key": api_key}, fieldname=["name"])
-	form_dict = frappe.local.form_dict
-	doc_secret = frappe.utils.password.get_decrypted_password(doctype, doc, fieldname="api_secret")
-	if api_secret == doc_secret:
-		if doctype == "User":
-			user = frappe.db.get_value(doctype="User", filters={"api_key": api_key}, fieldname=["name"])
-		else:
-			user = frappe.db.get_value(doctype, doc, "user")
-		if frappe.local.login_manager.user in ("", "Guest"):
-			frappe.set_user(user)
-		frappe.local.form_dict = form_dict
-
-
-def validate_auth_via_hooks():
-	for auth_hook in frappe.get_hooks("auth_hooks", []):
-		frappe.get_attr(auth_hook)()
diff --git a/frappe/api/__init__.py b/frappe/api/__init__.py
new file mode 100644
index 0000000000..5c504b2512
--- /dev/null
+++ b/frappe/api/__init__.py
@@ -0,0 +1,80 @@
+# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
+# License: MIT. See LICENSE
+from enum import Enum
+
+from werkzeug.exceptions import NotFound
+from werkzeug.routing import Map, Submount
+from werkzeug.wrappers import Request, Response
+
+import frappe
+import frappe.client
+from frappe import _
+from frappe.utils.response import build_response
+
+
+class ApiVersion(str, Enum):
+	V1 = "v1"
+	V2 = "v2"
+
+
+def handle(request: Request):
+	"""
+	Entry point for `/api` methods.
+
+	APIs are versioned using second part of path.
+	v1 -> `/api/v1/*`
+	v2 -> `/api/v2/*`
+
+	Different versions have different specification but broadly following things are supported:
+
+	- `/api/method/{methodname}` will call a whitelisted method
+	- `/api/resource/{doctype}` will query a table
+	        examples:
+	        - `?fields=["name", "owner"]`
+	        - `?filters=[["Task", "name", "like", "%005"]]`
+	        - `?limit_start=0`
+	        - `?limit_page_length=20`
+	- `/api/resource/{doctype}/{name}` will point to a resource
+	        `GET` will return document
+	        `POST` will insert
+	        `PUT` will update
+	        `DELETE` will delete
+	"""
+
+	try:
+		endpoint, arguments = API_URL_MAP.bind_to_environ(request.environ).match()
+	except NotFound:  # Wrap 404 - backward compatiblity
+		raise frappe.DoesNotExistError
+
+	data = endpoint(**arguments)
+	if isinstance(data, Response):
+		return data
+
+	if data is not None:
+		frappe.response["data"] = data
+	return build_response("json")
+
+
+# Merge all API version routing rules
+from frappe.api.v1 import url_rules as v1_rules
+from frappe.api.v2 import url_rules as v2_rules
+
+API_URL_MAP = Map(
+	[
+		# V1 routes
+		Submount("/api", v1_rules),
+		Submount(f"/api/{ApiVersion.V1.value}", v1_rules),
+		Submount(f"/api/{ApiVersion.V2.value}", v2_rules),
+	],
+	strict_slashes=False,  # Allows skipping trailing slashes
+	merge_slashes=False,
+)
+
+
+def get_api_version() -> ApiVersion | None:
+	if not frappe.request:
+		return
+
+	if frappe.request.path.startswith(f"/api/{ApiVersion.V2.value}"):
+		return ApiVersion.V2
+	return ApiVersion.V1
diff --git a/frappe/api/utils.py b/frappe/api/utils.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/frappe/api/v1.py b/frappe/api/v1.py
new file mode 100644
index 0000000000..d2758f45d5
--- /dev/null
+++ b/frappe/api/v1.py
@@ -0,0 +1,118 @@
+import json
+
+from werkzeug.routing import Rule
+
+import frappe
+from frappe import _
+from frappe.utils.data import sbool
+
+
+def document_list(doctype: str):
+	if frappe.form_dict.get("fields"):
+		frappe.form_dict["fields"] = json.loads(frappe.form_dict["fields"])
+
+	# set limit of records for frappe.get_list
+	frappe.form_dict.setdefault(
+		"limit_page_length",
+		frappe.form_dict.limit or frappe.form_dict.limit_page_length or 20,
+	)
+
+	# convert strings to native types - only as_dict and debug accept bool
+	for param in ["as_dict", "debug"]:
+		param_val = frappe.form_dict.get(param)
+		if param_val is not None:
+			frappe.form_dict[param] = sbool(param_val)
+
+	# evaluate frappe.get_list
+	return frappe.call(frappe.client.get_list, doctype, **frappe.form_dict)
+
+
+def handle_rpc_call(method: str):
+	import frappe.handler
+
+	method = method.split("/")[0]  # for backward compatiblity
+
+	frappe.form_dict.cmd = method
+	return frappe.handler.handle()
+
+
+def create_doc(doctype: str):
+	data = get_request_form_data()
+	data.pop("doctype", None)
+	return frappe.new_doc(doctype, **data).insert()
+
+
+def update_doc(doctype: str, name: str):
+	data = get_request_form_data()
+
+	doc = frappe.get_doc(doctype, name, for_update=True)
+	if "flags" in data:
+		del data["flags"]
+
+	doc.update(data)
+	doc.save()
+
+	# check for child table doctype
+	if doc.get("parenttype"):
+		frappe.get_doc(doc.parenttype, doc.parent).save()
+
+	return doc
+
+
+def delete_doc(doctype: str, name: str):
+	# TODO: child doc handling
+	frappe.delete_doc(doctype, name, ignore_missing=False)
+	frappe.response.http_status_code = 202
+	return "ok"
+
+
+def read_doc(doctype: str, name: str):
+	# Backward compatiblity
+	if "run_method" in frappe.form_dict:
+		return execute_doc_method(doctype, name)
+
+	doc = frappe.get_doc(doctype, name)
+	if not doc.has_permission("read"):
+		raise frappe.PermissionError
+	doc.apply_fieldlevel_read_permissions()
+	return doc
+
+
+def execute_doc_method(doctype: str, name: str, method: str | None = None):
+	method = method or frappe.form_dict.pop("run_method")
+	doc = frappe.get_doc(doctype, name)
+	doc.is_whitelisted(method)
+
+	if frappe.request.method == "GET":
+		if not doc.has_permission("read"):
+			frappe.throw(_("Not permitted"), frappe.PermissionError)
+		return doc.run_method(method, **frappe.form_dict)
+
+	elif frappe.request.method == "POST":
+		if not doc.has_permission("write"):
+			frappe.throw(_("Not permitted"), frappe.PermissionError)
+
+		return doc.run_method(method, **frappe.form_dict)
+
+
+def get_request_form_data():
+	if frappe.form_dict.data is None:
+		data = frappe.safe_decode(frappe.request.get_data())
+	else:
+		data = frappe.form_dict.data
+
+	try:
+		return frappe.parse_json(data)
+	except ValueError:
+		return frappe.form_dict
+
+
+url_rules = [
+	Rule("/method/<path:method>", endpoint=handle_rpc_call),
+	Rule("/resource/<doctype>", methods=["GET"], endpoint=document_list),
+	Rule("/resource/<doctype>", methods=["POST"], endpoint=create_doc),
+	Rule("/resource/<doctype>/<path:name>/", methods=["GET"], endpoint=read_doc),
+	Rule("/resource/<doctype>/<path:name>/", methods=["PUT"], endpoint=update_doc),
+	Rule("/resource/<doctype>/<path:name>/", methods=["DELETE"], endpoint=delete_doc),
+	Rule("/resource/<doctype>/<path:name>/", methods=["POST"], endpoint=execute_doc_method),
+]
diff --git a/frappe/api/v2.py b/frappe/api/v2.py
new file mode 100644
index 0000000000..06b6eab04e
--- /dev/null
+++ b/frappe/api/v2.py
@@ -0,0 +1,193 @@
+"""REST API v2
+
+This file defines routes and implementation for REST API.
+
+Note:
+	- All functions in this file should be treated as "whitelisted" as they are exposed via routes
+	- None of the functions present here should be called from python code, their location and
+	  internal implementation can change without treating it as "breaking change".
+"""
+import json
+from typing import Any
+
+from werkzeug.routing import Rule
+
+import frappe
+import frappe.client
+from frappe import _, get_newargs, is_whitelisted
+from frappe.core.doctype.server_script.server_script_utils import get_server_script_map
+from frappe.handler import is_valid_http_method, run_server_script, upload_file
+
+PERMISSION_MAP = {
+	"GET": "read",
+	"POST": "write",
+}
+
+
+def handle_rpc_call(method: str, doctype: str | None = None):
+	from frappe.modules.utils import load_doctype_module
+
+	if doctype:
+		# Expand to run actual method from doctype controller
+		module = load_doctype_module(doctype)
+		method = module.__name__ + "." + method
+
+	for hook in reversed(frappe.get_hooks("override_whitelisted_methods", {}).get(method, [])):
+		# override using the last hook
+		method = hook
+		break
+
+	# via server script
+	server_script = get_server_script_map().get("_api", {}).get(method)
+	if server_script:
+		return run_server_script(server_script)
+
+	try:
+		method = frappe.get_attr(method)
+	except Exception as e:
+		frappe.throw(_("Failed to get method {0} with {1}").format(method, e))
+
+	is_whitelisted(method)
+	is_valid_http_method(method)
+
+	return frappe.call(method, **frappe.form_dict)
+
+
+def login():
+	"""Login happens implicitly, this function doesn't do anything."""
+	pass
+
+
+def logout():
+	frappe.local.login_manager.logout()
+	frappe.db.commit()
+
+
+def read_doc(doctype: str, name: str):
+	doc = frappe.get_doc(doctype, name)
+	doc.check_permission("read")
+	doc.apply_fieldlevel_read_permissions()
+	return doc
+
+
+def document_list(doctype: str):
+	if frappe.form_dict.get("fields"):
+		frappe.form_dict["fields"] = json.loads(frappe.form_dict["fields"])
+
+	# set limit of records for frappe.get_list
+	frappe.form_dict.limit_page_length = frappe.form_dict.limit or 20
+	# evaluate frappe.get_list
+	return frappe.call(frappe.client.get_list, doctype, **frappe.form_dict)
+
+
+def count(doctype: str) -> int:
+	from frappe.desk.reportview import get_count
+
+	frappe.form_dict.doctype = doctype
+
+	return get_count()
+
+
+def create_doc(doctype: str):
+	data = frappe.form_dict
+	data.pop("doctype", None)
+	return frappe.new_doc(doctype, **data).insert()
+
+
+def update_doc(doctype: str, name: str):
+	data = frappe.form_dict
+
+	doc = frappe.get_doc(doctype, name, for_update=True)
+	data.pop("flags", None)
+	doc.update(data)
+	doc.save()
+
+	# check for child table doctype
+	if doc.get("parenttype"):
+		frappe.get_doc(doc.parenttype, doc.parent).save()
+
+	return doc
+
+
+def delete_doc(doctype: str, name: str):
+	frappe.client.delete_doc(doctype, name)
+	frappe.response.http_status_code = 202
+	return "ok"
+
+
+def execute_doc_method(doctype: str, name: str, method: str | None = None):
+	"""Get a document from DB and execute method on it.
+
+	Use cases:
+	- Submitting/cancelling document
+	- Triggering some kind of update on a document
+	"""
+	method = method or frappe.form_dict.pop("run_method")
+	doc = frappe.get_doc(doctype, name)
+	doc.is_whitelisted(method)
+
+	doc.check_permission(PERMISSION_MAP[frappe.request.method])
+	return doc.run_method(method, **frappe.form_dict)
+
+
+def run_doc_method(method: str, document: dict[str, Any] | str, kwargs=None):
+	"""run a whitelisted controller method on in-memory document.
+
+
+	This is useful for building clients that don't necessarily encode all the business logic but
+	call server side function on object to validate and modify the doc.
+
+	The doc CAN exists in DB too and can write to DB as well if method is POST.
+	"""
+
+	if isinstance(document, str):
+		document = frappe.parse_json(document)
+
+	if kwargs is None:
+		kwargs = {}
+
+	doc = frappe.get_doc(document)
+	doc._original_modified = doc.modified
+	doc.check_if_latest()
+
+	doc.check_permission(PERMISSION_MAP[frappe.request.method])
+
+	method_obj = getattr(doc, method)
+	fn = getattr(method_obj, "__func__", method_obj)
+	is_whitelisted(fn)
+	is_valid_http_method(fn)
+
+	new_kwargs = get_newargs(fn, kwargs)
+	response = doc.run_method(method, **new_kwargs)
+	frappe.response.docs.append(doc)  # send modified document and result both.
+	return response
+
+
+url_rules = [
+	# RPC calls
+	Rule("/method/login", endpoint=login),
+	Rule("/method/logout", endpoint=logout),
+	Rule("/method/ping", endpoint=frappe.ping),
+	Rule("/method/upload_file", endpoint=upload_file),
+	Rule("/method/<method>", endpoint=handle_rpc_call),
+	Rule(
+		"/method/run_doc_method",
+		methods=["GET", "POST"],
+		endpoint=lambda: frappe.call(run_doc_method, **frappe.form_dict),
+	),
+	Rule("/method/<doctype>/<method>", endpoint=handle_rpc_call),
+	# Document level APIs
+	Rule("/document/<doctype>", methods=["GET"], endpoint=document_list),
+	Rule("/document/<doctype>", methods=["POST"], endpoint=create_doc),
+	Rule("/document/<doctype>/<path:name>/", methods=["GET"], endpoint=read_doc),
+	Rule("/document/<doctype>/<path:name>/", methods=["PATCH", "PUT"], endpoint=update_doc),
+	Rule("/document/<doctype>/<path:name>/", methods=["DELETE"], endpoint=delete_doc),
+	Rule(
+		"/document/<doctype>/<path:name>/method/<method>/",
+		methods=["GET", "POST"],
+		endpoint=execute_doc_method,
+	),
+	# Collection level APIs
+	Rule("/doctype/<doctype>/meta", methods=["GET"], endpoint=frappe.get_meta),
+	Rule("/doctype/<doctype>/count", methods=["GET"], endpoint=count),
+]
diff --git a/frappe/app.py b/frappe/app.py
index 28fa9c2426..add62c2bbd 100644
--- a/frappe/app.py
+++ b/frappe/app.py
@@ -22,10 +22,11 @@ import frappe.rate_limiter
 import frappe.recorder
 import frappe.utils.response
 from frappe import _
-from frappe.auth import SAFE_HTTP_METHODS, UNSAFE_HTTP_METHODS, HTTPRequest
+from frappe.auth import SAFE_HTTP_METHODS, UNSAFE_HTTP_METHODS, HTTPRequest, validate_auth
 from frappe.middlewares import StaticDataMiddleware
-from frappe.utils import CallbackManager, cint, get_site_name, sanitize_html
+from frappe.utils import CallbackManager, cint, get_site_name
 from frappe.utils.data import escape_html
+from frappe.utils.deprecations import deprecation_warning
 from frappe.utils.error import log_error_snapshot
 from frappe.website.serve import get_response
 
@@ -93,16 +94,20 @@ def application(request: Request):
 
 		init_request(request)
 
-		frappe.api.validate_auth()
+		validate_auth()
 
 		if request.method == "OPTIONS":
 			response = Response()
 
 		elif frappe.form_dict.cmd:
-			response = frappe.handler.handle()
+			deprecation_warning(
+				f"{frappe.form_dict.cmd}: Sending `cmd` for RPC calls is deprecated, call REST API instead `/api/method/cmd`"
+			)
+			frappe.handler.handle()
+			response = frappe.utils.response.build_response("json")
 
 		elif request.path.startswith("/api/"):
-			response = frappe.api.handle()
+			response = frappe.api.handle(request)
 
 		elif request.path.startswith("/backups"):
 			response = frappe.utils.response.download_backup(request.path)
diff --git a/frappe/auth.py b/frappe/auth.py
index d1259e1aaf..01a1323a5b 100644
--- a/frappe/auth.py
+++ b/frappe/auth.py
@@ -1,6 +1,8 @@
 # Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and Contributors
 # MIT License. See LICENSE
-from urllib.parse import quote
+import base64
+import binascii
+from urllib.parse import quote, urlencode, urlparse
 
 import frappe
 import frappe.database
@@ -547,3 +549,102 @@ class LoginAttemptTracker:
 		):
 			return False
 		return True
+
+
+def validate_auth():
+	"""
+	Authenticate and sets user for the request.
+	"""
+	authorization_header = frappe.get_request_header("Authorization", "").split(" ")
+
+	if len(authorization_header) == 2:
+		validate_oauth(authorization_header)
+		validate_auth_via_api_keys(authorization_header)
+
+	validate_auth_via_hooks()
+
+
+def validate_oauth(authorization_header):
+	"""
+	Authenticate request using OAuth and set session user
+
+	Args:
+	        authorization_header (list of str): The 'Authorization' header containing the prefix and token
+	"""
+
+	from frappe.integrations.oauth2 import get_oauth_server
+	from frappe.oauth import get_url_delimiter
+
+	form_dict = frappe.local.form_dict
+	token = authorization_header[1]
+	req = frappe.request
+	parsed_url = urlparse(req.url)
+	access_token = {"access_token": token}
+	uri = (
+		parsed_url.scheme + "://" + parsed_url.netloc + parsed_url.path + "?" + urlencode(access_token)
+	)
+	http_method = req.method
+	headers = req.headers
+	body = req.get_data()
+	if req.content_type and "multipart/form-data" in req.content_type:
+		body = None
+
+	try:
+		required_scopes = frappe.db.get_value("OAuth Bearer Token", token, "scopes").split(
+			get_url_delimiter()
+		)
+		valid, oauthlib_request = get_oauth_server().verify_request(
+			uri, http_method, body, headers, required_scopes
+		)
+		if valid:
+			frappe.set_user(frappe.db.get_value("OAuth Bearer Token", token, "user"))
+			frappe.local.form_dict = form_dict
+	except AttributeError:
+		pass
+
+
+def validate_auth_via_api_keys(authorization_header):
+	"""
+	Authenticate request using API keys and set session user
+
+	Args:
+	        authorization_header (list of str): The 'Authorization' header containing the prefix and token
+	"""
+
+	try:
+		auth_type, auth_token = authorization_header
+		authorization_source = frappe.get_request_header("Frappe-Authorization-Source")
+		if auth_type.lower() == "basic":
+			api_key, api_secret = frappe.safe_decode(base64.b64decode(auth_token)).split(":")
+			validate_api_key_secret(api_key, api_secret, authorization_source)
+		elif auth_type.lower() == "token":
+			api_key, api_secret = auth_token.split(":")
+			validate_api_key_secret(api_key, api_secret, authorization_source)
+	except binascii.Error:
+		frappe.throw(
+			_("Failed to decode token, please provide a valid base64-encoded token."),
+			frappe.InvalidAuthorizationToken,
+		)
+	except (AttributeError, TypeError, ValueError):
+		pass
+
+
+def validate_api_key_secret(api_key, api_secret, frappe_authorization_source=None):
+	"""frappe_authorization_source to provide api key and secret for a doctype apart from User"""
+	doctype = frappe_authorization_source or "User"
+	doc = frappe.db.get_value(doctype=doctype, filters={"api_key": api_key}, fieldname=["name"])
+	form_dict = frappe.local.form_dict
+	doc_secret = frappe.utils.password.get_decrypted_password(doctype, doc, fieldname="api_secret")
+	if api_secret == doc_secret:
+		if doctype == "User":
+			user = frappe.db.get_value(doctype="User", filters={"api_key": api_key}, fieldname=["name"])
+		else:
+			user = frappe.db.get_value(doctype, doc, "user")
+		if frappe.local.login_manager.user in ("", "Guest"):
+			frappe.set_user(user)
+		frappe.local.form_dict = form_dict
+
+
+def validate_auth_via_hooks():
+	for auth_hook in frappe.get_hooks("auth_hooks", []):
+		frappe.get_attr(auth_hook)()
diff --git a/frappe/boot.py b/frappe/boot.py
index 3f6e3a6a39..c36927637a 100644
--- a/frappe/boot.py
+++ b/frappe/boot.py
@@ -295,8 +295,7 @@ def add_home_page(bootinfo, docs):
 		docs.append(page)
 		bootinfo["home_page"] = page.name
 	except (frappe.DoesNotExistError, frappe.PermissionError):
-		if frappe.message_log:
-			frappe.message_log.pop()
+		frappe.clear_last_message()
 		bootinfo["home_page"] = "Workspaces"
 
 
diff --git a/frappe/client.py b/frappe/client.py
index 85e99a6534..6439e9d71d 100644
--- a/frappe/client.py
+++ b/frappe/client.py
@@ -32,8 +32,8 @@ def get_list(
 	limit_start=None,
 	limit_page_length=20,
 	parent=None,
-	debug=False,
-	as_dict=True,
+	debug: bool = False,
+	as_dict: bool = True,
 	or_filters=None,
 ):
 	"""Returns a list of records by filters, fields, ordering and limit
diff --git a/frappe/core/doctype/data_import/test_importer.py b/frappe/core/doctype/data_import/test_importer.py
index 978f5792dd..965e34c3e6 100644
--- a/frappe/core/doctype/data_import/test_importer.py
+++ b/frappe/core/doctype/data_import/test_importer.py
@@ -63,7 +63,7 @@ class TestImporter(FrappeTestCase):
 	def test_data_import_without_mandatory_values(self):
 		import_file = get_import_file("sample_import_file_without_mandatory")
 		data_import = self.get_importer(doctype_name, import_file)
-		frappe.local.message_log = []
+		frappe.clear_messages()
 		data_import.start_import()
 		data_import.reload()
 
diff --git a/frappe/core/doctype/deleted_document/deleted_document.py b/frappe/core/doctype/deleted_document/deleted_document.py
index b5b35206e9..aa6239c279 100644
--- a/frappe/core/doctype/deleted_document/deleted_document.py
+++ b/frappe/core/doctype/deleted_document/deleted_document.py
@@ -73,11 +73,11 @@ def bulk_restore(docnames):
 			restored.append(d)
 
 		except frappe.DocumentAlreadyRestored:
-			frappe.message_log.pop()
+			frappe.clear_last_message()
 			invalid.append(d)
 
 		except Exception:
-			frappe.message_log.pop()
+			frappe.clear_last_message()
 			failed.append(d)
 			frappe.db.rollback()
 
diff --git a/frappe/core/doctype/document_naming_settings/document_naming_settings.py b/frappe/core/doctype/document_naming_settings/document_naming_settings.py
index 3ec4147ec7..ddb25dd262 100644
--- a/frappe/core/doctype/document_naming_settings/document_naming_settings.py
+++ b/frappe/core/doctype/document_naming_settings/document_naming_settings.py
@@ -248,8 +248,7 @@ class DocumentNamingSettings(Document):
 			doc = self._fetch_last_doc_if_available()
 			return "\n".join(NamingSeries(series).get_preview(doc=doc))
 		except Exception as e:
-			if frappe.message_log:
-				frappe.message_log.pop()
+			frappe.clear_last_message()
 			return _("Failed to generate names from the series") + f"\n{str(e)}"
 
 	def _fetch_last_doc_if_available(self):
diff --git a/frappe/core/doctype/file/test_file.py b/frappe/core/doctype/file/test_file.py
index 42f349aef4..43dc51c8b1 100644
--- a/frappe/core/doctype/file/test_file.py
+++ b/frappe/core/doctype/file/test_file.py
@@ -501,7 +501,7 @@ class TestFile(FrappeTestCase):
 		test_file.file_url = frappe.utils.get_url("unknown.jpg")
 		test_file.make_thumbnail(suffix="xs")
 		self.assertEqual(
-			json.loads(frappe.message_log[0]).get("message"),
+			frappe.message_log[0].get("message"),
 			f"File '{frappe.utils.get_url('unknown.jpg')}' not found",
 		)
 		self.assertEqual(test_file.thumbnail_url, None)
diff --git a/frappe/core/doctype/user/test_user.py b/frappe/core/doctype/user/test_user.py
index 1ca0a56ec0..dc973c9e8f 100644
--- a/frappe/core/doctype/user/test_user.py
+++ b/frappe/core/doctype/user/test_user.py
@@ -420,7 +420,7 @@ class TestUser(FrappeTestCase):
 			self.assertEqual(update_password(new_password, key=test_user.reset_password_key), "/")
 			update_password(old_password, old_password=new_password)
 			self.assertEqual(
-				json.loads(frappe.message_log[0]).get("message"),
+				frappe.message_log[0].get("message"),
 				"Password reset instructions have been sent to your email",
 			)
 
diff --git a/frappe/core/doctype/user/user.py b/frappe/core/doctype/user/user.py
index 1af7af72e5..f22c62050e 100644
--- a/frappe/core/doctype/user/user.py
+++ b/frappe/core/doctype/user/user.py
@@ -549,6 +549,10 @@ class User(Document):
 		# delete user permissions
 		frappe.db.delete("User Permission", {"user": self.name})
 
+		# Delete OAuth data
+		frappe.db.delete("OAuth Authorization Code", {"user": self.name})
+		frappe.db.delete("Token Cache", {"user": self.name})
+
 	def before_rename(self, old_name, new_name, merge=False):
 		frappe.clear_cache(user=old_name)
 		self.validate_rename(old_name, new_name)
@@ -775,7 +779,7 @@ def get_timezones():
 
 
 @frappe.whitelist()
-def get_all_roles(arg=None):
+def get_all_roles():
 	"""return all roles"""
 	active_domains = frappe.get_active_domains()
 
@@ -789,7 +793,7 @@ def get_all_roles(arg=None):
 		order_by="name",
 	)
 
-	return [role.get("name") for role in roles]
+	return sorted([role.get("name") for role in roles])
 
 
 @frappe.whitelist()
diff --git a/frappe/desk/desktop.py b/frappe/desk/desktop.py
index 4701879982..8ae20f7bb0 100644
--- a/frappe/desk/desktop.py
+++ b/frappe/desk/desktop.py
@@ -22,8 +22,7 @@ def handle_not_exist(fn):
 		try:
 			return fn(*args, **kwargs)
 		except DoesNotExistError:
-			if frappe.message_log:
-				frappe.message_log.pop()
+			frappe.clear_last_message()
 			return []
 
 	return wrapper
diff --git a/frappe/desk/doctype/desktop_icon/desktop_icon.py b/frappe/desk/doctype/desktop_icon/desktop_icon.py
index 524285f85d..7901ef9500 100644
--- a/frappe/desk/doctype/desktop_icon/desktop_icon.py
+++ b/frappe/desk/doctype/desktop_icon/desktop_icon.py
@@ -283,8 +283,7 @@ def set_desktop_icons(visible_list, ignore_duplicate=True):
 						raise e
 					else:
 						visible_list.remove(module_name)
-						if frappe.message_log:
-							frappe.message_log.pop()
+						frappe.clear_last_message()
 
 	# set the order
 	set_order(visible_list)
diff --git a/frappe/desk/form/linked_with.py b/frappe/desk/form/linked_with.py
index 8f569b5a9e..d9414ef706 100644
--- a/frappe/desk/form/linked_with.py
+++ b/frappe/desk/form/linked_with.py
@@ -427,8 +427,7 @@ def get_linked_docs(doctype: str, name: str, linkinfo: dict | None = None) -> di
 			link_meta_bundle = frappe.desk.form.load.get_meta_bundle(dt)
 		except Exception as e:
 			if isinstance(e, frappe.DoesNotExistError):
-				if frappe.local.message_log:
-					frappe.local.message_log.pop()
+				frappe.clear_last_message()
 			continue
 		linkmeta = link_meta_bundle[0]
 
@@ -502,8 +501,7 @@ def get_linked_docs(doctype: str, name: str, linkinfo: dict | None = None) -> di
 						ret = None
 
 			except frappe.PermissionError:
-				if frappe.local.message_log:
-					frappe.local.message_log.pop()
+				frappe.clear_last_message()
 
 				continue
 
diff --git a/frappe/desk/reportview.py b/frappe/desk/reportview.py
index 102a708895..fd299af819 100644
--- a/frappe/desk/reportview.py
+++ b/frappe/desk/reportview.py
@@ -46,7 +46,7 @@ def get_list():
 
 @frappe.whitelist()
 @frappe.read_only()
-def get_count():
+def get_count() -> int:
 	args = get_form_params()
 
 	if is_virtual_doctype(args.doctype):
@@ -65,7 +65,7 @@ def execute(doctype, *args, **kwargs):
 
 
 def get_form_params():
-	"""Stringify GET request parameters."""
+	"""parse GET request parameters."""
 	data = frappe._dict(frappe.local.form_dict)
 	clean_params(data)
 	validate_args(data)
diff --git a/frappe/email/doctype/email_account/email_account.py b/frappe/email/doctype/email_account/email_account.py
index 149e42dd80..24d548cb37 100755
--- a/frappe/email/doctype/email_account/email_account.py
+++ b/frappe/email/doctype/email_account/email_account.py
@@ -309,7 +309,7 @@ class EmailAccount(Document):
 		except OSError:
 			if in_receive:
 				# timeout while connecting, see receive.py connect method
-				description = frappe.message_log.pop() if frappe.message_log else "Socket Error"
+				description = frappe.clear_last_message() if frappe.message_log else "Socket Error"
 				if test_internet():
 					self.db_set("no_failed", self.no_failed + 1)
 					if self.no_failed > 2:
@@ -496,7 +496,7 @@ class EmailAccount(Document):
 							}
 						)
 					except assign_to.DuplicateToDoError:
-						frappe.message_log.pop()
+						frappe.clear_last_message()
 						pass
 			else:
 				self.set_failed_attempts_count(self.get_failed_attempts_count() + 1)
diff --git a/frappe/handler.py b/frappe/handler.py
index c6e7d79878..db99a47a43 100644
--- a/frappe/handler.py
+++ b/frappe/handler.py
@@ -15,6 +15,7 @@ from frappe.core.doctype.server_script.server_script_utils import get_server_scr
 from frappe.monitor import add_data_to_monitor
 from frappe.utils import cint
 from frappe.utils.csvutils import build_csv_response
+from frappe.utils.deprecations import deprecation_warning
 from frappe.utils.image import optimize_image
 from frappe.utils.response import build_response
 
@@ -56,13 +57,11 @@ def handle():
 		# add the response to `message` label
 		frappe.response["message"] = data
 
-	return build_response("json")
-
 
 def execute_cmd(cmd, from_async=False):
 	"""execute a request as python module"""
 	for hook in reversed(frappe.get_hooks("override_whitelisted_methods", {}).get(cmd, [])):
-		# override using the first hook
+		# override using the last hook
 		cmd = hook
 		break
 
@@ -273,6 +272,9 @@ def get_attr(cmd):
 	if "." in cmd:
 		method = frappe.get_attr(cmd)
 	else:
+		deprecation_warning(
+			f"Calling shorthand for {cmd} is deprecated, please specify full path in RPC call."
+		)
 		method = globals()[cmd]
 	frappe.log("method:" + cmd)
 	return method
diff --git a/frappe/integrations/doctype/connected_app/connected_app.py b/frappe/integrations/doctype/connected_app/connected_app.py
index d571b2ba00..d6b173d040 100644
--- a/frappe/integrations/doctype/connected_app/connected_app.py
+++ b/frappe/integrations/doctype/connected_app/connected_app.py
@@ -48,7 +48,8 @@ class ConnectedApp(Document):
 	def validate(self):
 		base_url = frappe.utils.get_url()
 		callback_path = (
-			"/api/method/frappe.integrations.doctype.connected_app.connected_app.callback/" + self.name
+			"/api/method/frappe.integrations.doctype.connected_app.connected_app.callback"
+			+ f"?app={self.name}"
 		)
 		self.redirect_uri = urljoin(base_url, callback_path)
 
@@ -148,7 +149,7 @@ class ConnectedApp(Document):
 
 
 @frappe.whitelist(methods=["GET"], allow_guest=True)
-def callback(code=None, state=None):
+def callback(code=None, state=None, app=None):
 	"""Handle client's code.
 
 	Called during the oauthorization flow by the remote oAuth2 server to
@@ -161,11 +162,7 @@ def callback(code=None, state=None):
 		frappe.local.response["location"] = "/login?" + urlencode({"redirect-to": frappe.request.url})
 		return
 
-	path = frappe.request.path[1:].split("/")
-	if len(path) != 4 or not path[3]:
-		frappe.throw(_("Invalid Parameters."))
-
-	connected_app = frappe.get_doc("Connected App", path[3])
+	connected_app = frappe.get_doc("Connected App", app)
 	token_cache = frappe.get_doc("Token Cache", connected_app.name + "-" + frappe.session.user)
 
 	if state != token_cache.state:
diff --git a/frappe/integrations/doctype/connected_app/test_connected_app.py b/frappe/integrations/doctype/connected_app/test_connected_app.py
index 88441db6b2..49ed6236ff 100644
--- a/frappe/integrations/doctype/connected_app/test_connected_app.py
+++ b/frappe/integrations/doctype/connected_app/test_connected_app.py
@@ -126,7 +126,7 @@ class TestConnectedApp(FrappeTestCase):
 		def delete_if_exists(attribute):
 			doc = getattr(self, attribute, None)
 			if doc:
-				doc.delete()
+				doc.delete(force=True)
 
 		delete_if_exists("token_cache")
 		delete_if_exists("connected_app")
diff --git a/frappe/model/document.py b/frappe/model/document.py
index 410d889e58..212ae1e6f4 100644
--- a/frappe/model/document.py
+++ b/frappe/model/document.py
@@ -22,6 +22,7 @@ from frappe.model.workflow import set_workflow_state_on_action, validate_workflo
 from frappe.types import DF
 from frappe.utils import compare, cstr, date_diff, file_lock, flt, get_datetime_str, now
 from frappe.utils.data import get_absolute_url
+from frappe.utils.deprecations import deprecated
 from frappe.utils.global_search import update_global_search
 
 if TYPE_CHECKING:
@@ -139,12 +140,6 @@ class Document(BaseDocument):
 	def is_locked(self):
 		return file_lock.lock_exists(self.get_signature())
 
-	@staticmethod
-	def whitelist(fn):
-		"""Decorator: Whitelist method to be called remotely via REST API."""
-		frappe.whitelist()(fn)
-		return fn
-
 	def load_from_db(self):
 		"""Load document and children from database and create properties
 		from fields"""
@@ -1016,19 +1011,16 @@ class Document(BaseDocument):
 			elif alert.event == "Method" and method == alert.method:
 				_evaluate_alert(alert)
 
-	@whitelist.__func__
 	def _submit(self):
 		"""Submit the document. Sets `docstatus` = 1, then saves."""
 		self.docstatus = DocStatus.submitted()
 		return self.save()
 
-	@whitelist.__func__
 	def _cancel(self):
 		"""Cancel the document. Sets `docstatus` = 2, then saves."""
 		self.docstatus = DocStatus.cancelled()
 		return self.save()
 
-	@whitelist.__func__
 	def _rename(
 		self, name: str, merge: bool = False, force: bool = False, validate_rename: bool = True
 	):
@@ -1038,20 +1030,18 @@ class Document(BaseDocument):
 		self.name = rename_doc(doc=self, new=name, merge=merge, force=force, validate=validate_rename)
 		self.reload()
 
-	@whitelist.__func__
+	@frappe.whitelist()
 	def submit(self):
 		"""Submit the document. Sets `docstatus` = 1, then saves."""
 		return self._submit()
 
-	@whitelist.__func__
+	@frappe.whitelist()
 	def cancel(self):
 		"""Cancel the document. Sets `docstatus` = 2, then saves."""
 		return self._cancel()
 
-	@whitelist.__func__
-	def rename(
-		self, name: str, merge: bool = False, force: bool = False, validate_rename: bool = True
-	):
+	@frappe.whitelist()
+	def rename(self, name: str, merge=False, force=False, validate_rename=True):
 		"""Rename the document to `name`. This transforms the current object."""
 		return self._rename(name=name, merge=merge, force=force, validate_rename=validate_rename)
 
@@ -1638,8 +1628,8 @@ def execute_action(__doctype, __name, __action, **kwargs):
 		frappe.db.rollback()
 
 		# add a comment (?)
-		if frappe.local.message_log:
-			msg = json.loads(frappe.local.message_log[-1]).get("message")
+		if frappe.message_log:
+			msg = frappe.message_log[-1].get("message")
 		else:
 			msg = "<pre><code>" + frappe.get_traceback() + "</pre></code>"
 
diff --git a/frappe/public/js/frappe/request.js b/frappe/public/js/frappe/request.js
index b622037338..10a7158e4a 100644
--- a/frappe/public/js/frappe/request.js
+++ b/frappe/public/js/frappe/request.js
@@ -88,7 +88,11 @@ frappe.call = function (opts) {
 
 	let url = opts.url;
 	if (!url) {
-		url = "/api/method/" + args.cmd;
+		let prefix = "/api/method/";
+		if (opts.api_version) {
+			prefix = `/api/${opts.api_version}/method/`;
+		}
+		url = prefix + args.cmd;
 		if (window.cordova) {
 			let host = frappe.request.url;
 			host = host.slice(0, host.length - 1);
@@ -116,6 +120,7 @@ frappe.call = function (opts) {
 		// show_spinner: !opts.no_spinner,
 		async: opts.async,
 		silent: opts.silent,
+		api_version: opts.api_version,
 		url,
 	});
 };
@@ -444,12 +449,18 @@ frappe.request.cleanup = function (opts, r) {
 		}
 
 		// show messages
-		if (r._server_messages && !opts.silent) {
+		//
+		let messages;
+		if (opts.api_version == "v2") {
+			messages = r.messages;
+		} else if (r._server_messages) {
+			messages = JSON.parse(r._server_messages);
+		}
+		if (messages && !opts.silent) {
 			// show server messages if no handlers exist
 			if (handlers.length === 0) {
-				r._server_messages = JSON.parse(r._server_messages);
 				frappe.hide_msgprint();
-				frappe.msgprint(r._server_messages);
+				frappe.msgprint(messages);
 			}
 		}
 
diff --git a/frappe/public/js/frappe/ui/messages.js b/frappe/public/js/frappe/ui/messages.js
index b8ed34d59d..c3fae2c9af 100644
--- a/frappe/public/js/frappe/ui/messages.js
+++ b/frappe/public/js/frappe/ui/messages.js
@@ -144,7 +144,15 @@ frappe.msgprint = function (msg, title, is_minimizable) {
 
 	if (data.message instanceof Array) {
 		let messages = data.message;
-		const exceptions = messages.map((m) => JSON.parse(m)).filter((m) => m.raise_exception);
+		const exceptions = messages
+			.map((m) => {
+				if (typeof m == "string") {
+					return JSON.parse(m);
+				} else {
+					return m;
+				}
+			})
+			.filter((m) => m.raise_exception);
 
 		// only show exceptions if any exceptions exist
 		if (exceptions.length) {
diff --git a/frappe/tests/test_api.py b/frappe/tests/test_api.py
index 5098158e0f..9b14b308ed 100644
--- a/frappe/tests/test_api.py
+++ b/frappe/tests/test_api.py
@@ -1,20 +1,21 @@
 import json
 import sys
 from contextlib import contextmanager
+from functools import cached_property
 from random import choice
 from threading import Thread
 from time import time
 from unittest.mock import patch
+from urllib.parse import urljoin
 
 import requests
 from filetype import guess_mime
-from semantic_version import Version
 from werkzeug.test import TestResponse
 
 import frappe
 from frappe.installer import update_site_config
 from frappe.tests.utils import FrappeTestCase, patch_hooks
-from frappe.utils import cint, get_site_url, get_test_client
+from frappe.utils import cint, get_test_client, get_url
 
 try:
 	_site = frappe.local.site
@@ -73,34 +74,55 @@ class ThreadWithReturnValue(Thread):
 		return self._return
 
 
+resource_key = {
+	"": "resource",
+	"v1": "resource",
+	"v2": "document",
+}
+
+
 class FrappeAPITestCase(FrappeTestCase):
-	SITE = frappe.local.site
-	SITE_URL = get_site_url(SITE)
-	RESOURCE_URL = f"{SITE_URL}/api/resource"
+	version = ""  # Empty implies v1
 	TEST_CLIENT = get_test_client()
 
 	@property
+	def site_url(self):
+		return get_url()
+
+	def resource_path(self, *parts):
+		return self.get_path(resource_key[self.version], *parts)
+
+	def method_path(self, *method):
+		return self.get_path("method", *method)
+
+	def doctype_path(self, *method):
+		return self.get_path("doctype", *method)
+
+	def get_path(self, *parts):
+		return urljoin(self.site_url, "/".join(("api", self.version, *parts)))
+
+	@cached_property
 	def sid(self) -> str:
-		if not getattr(self, "_sid", None):
-			from frappe.auth import CookieManager, LoginManager
-			from frappe.utils import set_request
+		from frappe.auth import CookieManager, LoginManager
+		from frappe.utils import set_request
 
-			set_request(path="/")
-			frappe.local.cookie_manager = CookieManager()
-			frappe.local.login_manager = LoginManager()
-			frappe.local.login_manager.login_as("Administrator")
-			self._sid = frappe.session.sid
-
-		return self._sid
+		set_request(path="/")
+		frappe.local.cookie_manager = CookieManager()
+		frappe.local.login_manager = LoginManager()
+		frappe.local.login_manager.login_as("Administrator")
+		return frappe.session.sid
 
 	def get(self, path: str, params: dict | None = None, **kwargs) -> TestResponse:
-		return make_request(target=self.TEST_CLIENT.get, args=(path,), kwargs={"data": params, **kwargs})
+		return make_request(target=self.TEST_CLIENT.get, args=(path,), kwargs={"json": params, **kwargs})
 
 	def post(self, path, data, **kwargs) -> TestResponse:
-		return make_request(target=self.TEST_CLIENT.post, args=(path,), kwargs={"data": data, **kwargs})
+		return make_request(target=self.TEST_CLIENT.post, args=(path,), kwargs={"json": data, **kwargs})
 
 	def put(self, path, data, **kwargs) -> TestResponse:
-		return make_request(target=self.TEST_CLIENT.put, args=(path,), kwargs={"data": data, **kwargs})
+		return make_request(target=self.TEST_CLIENT.put, args=(path,), kwargs={"json": data, **kwargs})
+
+	def patch(self, path, data, **kwargs) -> TestResponse:
+		return make_request(target=self.TEST_CLIENT.patch, args=(path,), kwargs={"json": data, **kwargs})
 
 	def delete(self, path, **kwargs) -> TestResponse:
 		return make_request(target=self.TEST_CLIENT.delete, args=(path,), kwargs=kwargs)
@@ -113,8 +135,9 @@ class TestResourceAPI(FrappeAPITestCase):
 	@classmethod
 	def setUpClass(cls):
 		super().setUpClass()
-		for _ in range(10):
+		for _ in range(20):
 			doc = frappe.get_doc({"doctype": "ToDo", "description": frappe.mock("paragraph")}).insert()
+			cls.GENERATED_DOCUMENTS = []
 			cls.GENERATED_DOCUMENTS.append(doc.name)
 		frappe.db.commit()
 
@@ -126,31 +149,31 @@ class TestResourceAPI(FrappeAPITestCase):
 
 	def test_unauthorized_call(self):
 		# test 1: fetch documents without auth
-		response = requests.get(f"{self.RESOURCE_URL}/{self.DOCTYPE}")
+		response = requests.get(self.resource_path(self.DOCTYPE))
 		self.assertEqual(response.status_code, 403)
 
 	def test_get_list(self):
 		# test 2: fetch documents without params
-		response = self.get(f"/api/resource/{self.DOCTYPE}", {"sid": self.sid})
+		response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid})
 		self.assertEqual(response.status_code, 200)
 		self.assertIsInstance(response.json, dict)
 		self.assertIn("data", response.json)
 
 	def test_get_list_limit(self):
 		# test 3: fetch data with limit
-		response = self.get(f"/api/resource/{self.DOCTYPE}", {"sid": self.sid, "limit": 2})
+		response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "limit": 2})
 		self.assertEqual(response.status_code, 200)
 		self.assertEqual(len(response.json["data"]), 2)
 
 	def test_get_list_dict(self):
 		# test 4: fetch response as (not) dict
-		response = self.get(f"/api/resource/{self.DOCTYPE}", {"sid": self.sid, "as_dict": True})
+		response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "as_dict": True})
 		json = frappe._dict(response.json)
 		self.assertEqual(response.status_code, 200)
 		self.assertIsInstance(json.data, list)
 		self.assertIsInstance(json.data[0], dict)
 
-		response = self.get(f"/api/resource/{self.DOCTYPE}", {"sid": self.sid, "as_dict": False})
+		response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "as_dict": False})
 		json = frappe._dict(response.json)
 		self.assertEqual(response.status_code, 200)
 		self.assertIsInstance(json.data, list)
@@ -158,7 +181,8 @@ class TestResourceAPI(FrappeAPITestCase):
 
 	def test_get_list_debug(self):
 		# test 5: fetch response with debug
-		response = self.get(f"/api/resource/{self.DOCTYPE}", {"sid": self.sid, "debug": True})
+		with suppress_stdout():
+			response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "debug": True})
 		self.assertEqual(response.status_code, 200)
 		self.assertIn("exc", response.json)
 		self.assertIsInstance(response.json["exc"], str)
@@ -167,52 +191,48 @@ class TestResourceAPI(FrappeAPITestCase):
 	def test_get_list_fields(self):
 		# test 6: fetch response with fields
 		response = self.get(
-			f"/api/resource/{self.DOCTYPE}", {"sid": self.sid, "fields": '["description"]'}
+			self.resource_path(self.DOCTYPE), {"sid": self.sid, "fields": '["description"]'}
 		)
 		self.assertEqual(response.status_code, 200)
 		json = frappe._dict(response.json)
 		self.assertIn("description", json.data[0])
 
 	def test_create_document(self):
-		# test 7: POST method on /api/resource to create doc
 		data = {"description": frappe.mock("paragraph"), "sid": self.sid}
-		response = self.post(f"/api/resource/{self.DOCTYPE}", data)
+		response = self.post(self.resource_path(self.DOCTYPE), data)
 		self.assertEqual(response.status_code, 200)
 		docname = response.json["data"]["name"]
 		self.assertIsInstance(docname, str)
 		self.GENERATED_DOCUMENTS.append(docname)
 
 	def test_update_document(self):
-		# test 8: PUT method on /api/resource to update doc
 		generated_desc = frappe.mock("paragraph")
 		data = {"description": generated_desc, "sid": self.sid}
 		random_doc = choice(self.GENERATED_DOCUMENTS)
-		desc_before_update = frappe.db.get_value(self.DOCTYPE, random_doc, "description")
 
-		response = self.put(f"/api/resource/{self.DOCTYPE}/{random_doc}", data=data)
+		response = self.put(self.resource_path(self.DOCTYPE, random_doc), data=data)
 		self.assertEqual(response.status_code, 200)
-		self.assertNotEqual(response.json["data"]["description"], desc_before_update)
+		self.assertEqual(response.json["data"]["description"], generated_desc)
+
+		response = self.get(self.resource_path(self.DOCTYPE, random_doc))
 		self.assertEqual(response.json["data"]["description"], generated_desc)
 
 	def test_delete_document(self):
-		# test 9: DELETE method on /api/resource
 		doc_to_delete = choice(self.GENERATED_DOCUMENTS)
-		response = self.delete(f"/api/resource/{self.DOCTYPE}/{doc_to_delete}")
+		response = self.delete(self.resource_path(self.DOCTYPE, doc_to_delete))
 		self.assertEqual(response.status_code, 202)
-		self.assertDictEqual(response.json, {"message": "ok"})
-		self.GENERATED_DOCUMENTS.remove(doc_to_delete)
+		self.assertDictEqual(response.json, {"data": "ok"})
 
-		non_existent_doc = frappe.generate_hash(length=12)
-		with suppress_stdout():
-			response = self.delete(f"/api/resource/{self.DOCTYPE}/{non_existent_doc}")
+		response = self.get(self.resource_path(self.DOCTYPE, doc_to_delete))
 		self.assertEqual(response.status_code, 404)
-		self.assertDictEqual(response.json, {})
+		self.GENERATED_DOCUMENTS.remove(doc_to_delete)
 
 	def test_run_doc_method(self):
 		# test 10: Run whitelisted method on doc via /api/resource
 		# status_code is 403 if no other tests are run before this - it's not logged in
-		self.post("/api/resource/Website Theme/Standard", {"run_method": "get_apps"})
-		response = self.get("/api/resource/Website Theme/Standard", {"run_method": "get_apps"})
+		self.post(self.resource_path("Website Theme", "Standard"), {"run_method": "get_apps"})
+		response = self.get(self.resource_path("Website Theme", "Standard"), {"run_method": "get_apps"})
+
 		self.assertIn(response.status_code, (403, 200))
 
 		if response.status_code == 403:
@@ -232,25 +252,16 @@ class TestResourceAPI(FrappeAPITestCase):
 
 
 class TestMethodAPI(FrappeAPITestCase):
-	METHOD_PATH = "/api/method"
-
-	def setUp(self):
-		if self._testMethodName == "test_auth_cycle":
-			from frappe.core.doctype.user.user import generate_keys
-
-			generate_keys("Administrator")
-			frappe.db.commit()
-
 	def test_ping(self):
 		# test 2: test for /api/method/ping
-		response = self.get(f"{self.METHOD_PATH}/ping")
+		response = self.get(self.method_path("ping"))
 		self.assertEqual(response.status_code, 200)
 		self.assertIsInstance(response.json, dict)
 		self.assertEqual(response.json["message"], "pong")
 
 	def test_get_user_info(self):
 		# test 3: test for /api/method/frappe.realtime.get_user_info
-		response = self.get(f"{self.METHOD_PATH}/frappe.realtime.get_user_info")
+		response = self.get(self.method_path("frappe.realtime.get_user_info"))
 		self.assertEqual(response.status_code, 200)
 		self.assertIsInstance(response.json, dict)
 		self.assertIn(response.json.get("message").get("user"), ("Administrator", "Guest"))
@@ -258,10 +269,11 @@ class TestMethodAPI(FrappeAPITestCase):
 	def test_auth_cycle(self):
 		# test 4: Pass authorization token in request
 		global authorization_token
+		generate_admin_keys()
 		user = frappe.get_doc("User", "Administrator")
 		api_key, api_secret = user.api_key, user.get_password("api_secret")
 		authorization_token = f"{api_key}:{api_secret}"
-		response = self.get("/api/method/frappe.auth.get_logged_user")
+		response = self.get(self.method_path("frappe.auth.get_logged_user"))
 
 		self.assertEqual(response.status_code, 200)
 		self.assertEqual(response.json["message"], "Administrator")
@@ -269,18 +281,45 @@ class TestMethodAPI(FrappeAPITestCase):
 		authorization_token = None
 
 	def test_404s(self):
-		response = self.get("/api/rest", {"sid": self.sid})
+		response = self.get(self.get_path("rest"), {"sid": self.sid})
 		self.assertEqual(response.status_code, 404)
-		response = self.get("/api/resource/User/NonExistent@s.com", {"sid": self.sid})
+		response = self.get(self.resource_path("User", "NonExistent@s.com"), {"sid": self.sid})
 		self.assertEqual(response.status_code, 404)
 
+	def test_logs(self):
+		method = "frappe.tests.test_api.test"
+
+		def get_message(resp, msg_type):
+			return frappe.parse_json(frappe.parse_json(frappe.parse_json(resp.json)[msg_type])[0])
+
+		expected_message = "Failed"
+		response = self.get(self.method_path(method), {"sid": self.sid, "message": expected_message})
+		self.assertEqual(get_message(response, "_server_messages").message, expected_message)
+
+		# Cause handled failured
+		with suppress_stdout():
+			response = self.get(
+				self.method_path(method), {"sid": self.sid, "message": expected_message, "fail": True}
+			)
+		self.assertEqual(get_message(response, "_server_messages").message, expected_message)
+		self.assertEqual(response.json["exc_type"], "ValidationError")
+		self.assertIn("Traceback", response.json["exc"])
+
+		# Cause handled failured
+		with suppress_stdout():
+			response = self.get(
+				self.method_path(method),
+				{"sid": self.sid, "message": expected_message, "fail": True, "handled": False},
+			)
+		self.assertNotIn("_server_messages", response.json)
+		self.assertIn("ZeroDivisionError", response.json["exception"])  # WHY?
+		self.assertIn("Traceback", response.json["exc"])
+
 
 class TestReadOnlyMode(FrappeAPITestCase):
 	"""During migration if read only mode can be enabled.
 	Test if reads work well and writes are blocked"""
 
-	REQ_PATH = "/api/resource/ToDo"
-
 	@classmethod
 	def setUpClass(cls):
 		super().setUpClass()
@@ -290,13 +329,16 @@ class TestReadOnlyMode(FrappeAPITestCase):
 		update_site_config("maintenance_mode", 1)
 
 	def test_reads(self):
-		response = self.get(self.REQ_PATH, {"sid": self.sid})
+		response = self.get(self.resource_path("ToDo"), {"sid": self.sid})
 		self.assertEqual(response.status_code, 200)
 		self.assertIsInstance(response.json, dict)
 		self.assertIsInstance(response.json["data"], list)
 
 	def test_blocked_writes(self):
-		response = self.post(self.REQ_PATH, {"description": frappe.mock("paragraph"), "sid": self.sid})
+		with suppress_stdout():
+			response = self.post(
+				self.resource_path("ToDo"), {"description": frappe.mock("paragraph"), "sid": self.sid}
+			)
 		self.assertEqual(response.status_code, 503)
 		self.assertEqual(response.json["exc_type"], "InReadOnlyMode")
 
@@ -368,3 +410,21 @@ class TestResponse(FrappeAPITestCase):
 		self.assertEqual(response.status_code, 200)
 		self.assertIn("text/csv", response.headers["content-type"])
 		self.assertGreater(cint(response.headers["content-length"]), 0)
+
+
+def generate_admin_keys():
+	from frappe.core.doctype.user.user import generate_keys
+
+	generate_keys("Administrator")
+	frappe.db.commit()
+
+
+@frappe.whitelist()
+def test(*, fail=False, handled=True, message="Failed"):
+	if fail:
+		if handled:
+			frappe.throw(message)
+		else:
+			1 / 0
+	else:
+		frappe.msgprint(message)
diff --git a/frappe/tests/test_api_v2.py b/frappe/tests/test_api_v2.py
new file mode 100644
index 0000000000..ba25b8b05d
--- /dev/null
+++ b/frappe/tests/test_api_v2.py
@@ -0,0 +1,297 @@
+from random import choice
+
+import requests
+
+import frappe
+from frappe.installer import update_site_config
+from frappe.tests.test_api import FrappeAPITestCase, suppress_stdout
+
+authorization_token = None
+
+
+resource_key = {
+	"": "resource",
+	"v1": "resource",
+	"v2": "document",
+}
+
+
+class TestResourceAPIV2(FrappeAPITestCase):
+	version = "v2"
+	DOCTYPE = "ToDo"
+	GENERATED_DOCUMENTS = []
+
+	@classmethod
+	def setUpClass(cls):
+		super().setUpClass()
+		for _ in range(20):
+			doc = frappe.get_doc({"doctype": "ToDo", "description": frappe.mock("paragraph")}).insert()
+			cls.GENERATED_DOCUMENTS = []
+			cls.GENERATED_DOCUMENTS.append(doc.name)
+		frappe.db.commit()
+
+	@classmethod
+	def tearDownClass(cls):
+		for name in cls.GENERATED_DOCUMENTS:
+			frappe.delete_doc_if_exists(cls.DOCTYPE, name)
+		frappe.db.commit()
+
+	def test_unauthorized_call(self):
+		# test 1: fetch documents without auth
+		response = requests.get(self.resource_path(self.DOCTYPE))
+		self.assertEqual(response.status_code, 403)
+
+	def test_get_list(self):
+		# test 2: fetch documents without params
+		response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid})
+		self.assertEqual(response.status_code, 200)
+		self.assertIsInstance(response.json, dict)
+		self.assertIn("data", response.json)
+
+	def test_get_list_limit(self):
+		# test 3: fetch data with limit
+		response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "limit": 2})
+		self.assertEqual(response.status_code, 200)
+		self.assertEqual(len(response.json["data"]), 2)
+
+	def test_get_list_dict(self):
+		# test 4: fetch response as (not) dict
+		response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "as_dict": True})
+		json = frappe._dict(response.json)
+		self.assertEqual(response.status_code, 200)
+		self.assertIsInstance(json.data, list)
+		self.assertIsInstance(json.data[0], dict)
+
+		response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "as_dict": False})
+		json = frappe._dict(response.json)
+		self.assertEqual(response.status_code, 200)
+		self.assertIsInstance(json.data, list)
+		self.assertIsInstance(json.data[0], list)
+
+	def test_get_list_fields(self):
+		# test 6: fetch response with fields
+		response = self.get(
+			self.resource_path(self.DOCTYPE), {"sid": self.sid, "fields": '["description"]'}
+		)
+		self.assertEqual(response.status_code, 200)
+		json = frappe._dict(response.json)
+		self.assertIn("description", json.data[0])
+
+	def test_create_document(self):
+		data = {"description": frappe.mock("paragraph"), "sid": self.sid}
+		response = self.post(self.resource_path(self.DOCTYPE), data)
+		self.assertEqual(response.status_code, 200)
+		docname = response.json["data"]["name"]
+		self.assertIsInstance(docname, str)
+		self.GENERATED_DOCUMENTS.append(docname)
+
+	def test_delete_document(self):
+		doc_to_delete = choice(self.GENERATED_DOCUMENTS)
+		response = self.delete(self.resource_path(self.DOCTYPE, doc_to_delete))
+		self.assertEqual(response.status_code, 202)
+		self.assertDictEqual(response.json, {"data": "ok"})
+
+		response = self.get(self.resource_path(self.DOCTYPE, doc_to_delete))
+		self.assertEqual(response.status_code, 404)
+		self.GENERATED_DOCUMENTS.remove(doc_to_delete)
+
+	def test_execute_doc_method(self):
+		response = self.get(self.resource_path("Website Theme", "Standard", "method", "get_apps"))
+		self.assertEqual(response.json["data"][0]["name"], "frappe")
+
+	def test_update_document(self):
+		generated_desc = frappe.mock("paragraph")
+		data = {"description": generated_desc, "sid": self.sid}
+		random_doc = choice(self.GENERATED_DOCUMENTS)
+
+		response = self.patch(self.resource_path(self.DOCTYPE, random_doc), data=data)
+		self.assertEqual(response.status_code, 200)
+		self.assertEqual(response.json["data"]["description"], generated_desc)
+
+		response = self.get(self.resource_path(self.DOCTYPE, random_doc))
+		self.assertEqual(response.json["data"]["description"], generated_desc)
+
+	def test_delete_document_non_existing(self):
+		non_existent_doc = frappe.generate_hash(length=12)
+		with suppress_stdout():
+			response = self.delete(self.resource_path(self.DOCTYPE, non_existent_doc))
+		self.assertEqual(response.status_code, 404)
+		self.assertEqual(response.json["errors"][0]["type"], "DoesNotExistError")
+		# 404s dont return exceptions
+		self.assertFalse(response.json["errors"][0].get("exception"))
+
+
+class TestMethodAPIV2(FrappeAPITestCase):
+	version = "v2"
+
+	def setUp(self) -> None:
+		self.post(self.method_path("login"), {"sid": self.sid})
+		return super().setUp()
+
+	def test_ping(self):
+		response = self.get(self.method_path("ping"))
+		self.assertEqual(response.status_code, 200)
+		self.assertIsInstance(response.json, dict)
+		self.assertEqual(response.json["data"], "pong")
+
+	def test_get_user_info(self):
+		response = self.get(self.method_path("frappe.realtime.get_user_info"))
+		self.assertEqual(response.status_code, 200)
+		self.assertIsInstance(response.json, dict)
+		self.assertIn(response.json.get("data").get("user"), ("Administrator", "Guest"))
+
+	def test_auth_cycle(self):
+		global authorization_token
+
+		generate_admin_keys()
+		user = frappe.get_doc("User", "Administrator")
+		api_key, api_secret = user.api_key, user.get_password("api_secret")
+		authorization_token = f"{api_key}:{api_secret}"
+		response = self.get(self.method_path("frappe.auth.get_logged_user"))
+
+		self.assertEqual(response.status_code, 200)
+		self.assertEqual(response.json["data"], "Administrator")
+
+		authorization_token = None
+
+	def test_404s(self):
+		response = self.get(self.get_path("rest"), {"sid": self.sid})
+		self.assertEqual(response.status_code, 404)
+		response = self.get(self.resource_path("User", "NonExistent@s.com"), {"sid": self.sid})
+		self.assertEqual(response.status_code, 404)
+
+	def test_shorthand_controller_methods(self):
+		shorthand_response = self.get(self.method_path("User", "get_all_roles"), {"sid": self.sid})
+		self.assertIn("Blogger", shorthand_response.json["data"])
+
+		expanded_response = self.get(
+			self.method_path("frappe.core.doctype.user.user.get_all_roles"), {"sid": self.sid}
+		)
+		self.assertEqual(expanded_response.data, shorthand_response.data)
+
+	def test_logout(self):
+		self.post(self.method_path("logout"), {"sid": self.sid})
+		response = self.get(self.method_path("ping"))
+		self.assertFalse(response.request.cookies["sid"])
+
+	def test_run_doc_method_in_memory(self):
+		dns = frappe.get_doc("Document Naming Settings")
+
+		# Check that simple API can be called.
+		response = self.get(
+			self.method_path("run_doc_method"),
+			{
+				"sid": self.sid,
+				"document": dns.as_dict(),
+				"method": "get_transactions_and_prefixes",
+			},
+		)
+		self.assertTrue(response.json["data"])
+		self.assertGreaterEqual(len(response.json["docs"]), 1)
+
+		# Call with known and unknown arguments, only known should get passed
+		response = self.get(
+			self.method_path("run_doc_method"),
+			{
+				"sid": self.sid,
+				"document": dns.as_dict(),
+				"method": "get_options",
+				"kwargs": {"doctype": "Webhook", "unknown": "what"},
+			},
+		)
+		self.assertEqual(response.status_code, 200)
+
+	def test_logs(self):
+		method = "frappe.tests.test_api.test"
+
+		expected_message = "Failed v2"
+		response = self.get(
+			self.method_path(method), {"sid": self.sid, "message": expected_message}
+		).json
+
+		self.assertIsInstance(response["messages"], list)
+		self.assertEqual(response["messages"][0]["message"], expected_message)
+
+		# Cause handled failured
+		with suppress_stdout():
+			response = self.get(
+				self.method_path(method), {"sid": self.sid, "message": expected_message, "fail": True}
+			).json
+		self.assertIsInstance(response["errors"], list)
+		self.assertEqual(response["errors"][0]["message"], expected_message)
+		self.assertEqual(response["errors"][0]["type"], "ValidationError")
+		self.assertIn("Traceback", response["errors"][0]["exception"])
+
+		# Cause handled failured
+		with suppress_stdout():
+			response = self.get(
+				self.method_path(method),
+				{"sid": self.sid, "message": expected_message, "fail": True, "handled": False},
+			).json
+
+		self.assertIsInstance(response["errors"], list)
+		self.assertEqual(response["errors"][0]["type"], "ZeroDivisionError")
+		self.assertIn("Traceback", response["errors"][0]["exception"])
+
+
+class TestDocTypeAPIV2(FrappeAPITestCase):
+	version = "v2"
+
+	def setUp(self) -> None:
+		self.post(self.method_path("login"), {"sid": self.sid})
+		return super().setUp()
+
+	def test_meta(self):
+		response = self.get(self.doctype_path("ToDo", "meta"))
+		self.assertEqual(response.status_code, 200)
+		self.assertEqual(response.json["data"]["name"], "ToDo")
+
+	def test_count(self):
+		response = self.get(self.doctype_path("ToDo", "count"))
+		self.assertIsInstance(response.json["data"], int)
+
+
+class TestReadOnlyMode(FrappeAPITestCase):
+	"""During migration if read only mode can be enabled.
+	Test if reads work well and writes are blocked"""
+
+	version = "v2"
+
+	@classmethod
+	def setUpClass(cls):
+		super().setUpClass()
+		update_site_config("allow_reads_during_maintenance", 1)
+		cls.addClassCleanup(update_site_config, "maintenance_mode", 0)
+		update_site_config("maintenance_mode", 1)
+
+	def test_reads(self):
+		response = self.get(self.resource_path("ToDo"), {"sid": self.sid})
+		self.assertEqual(response.status_code, 200)
+		self.assertIsInstance(response.json, dict)
+		self.assertIsInstance(response.json["data"], list)
+
+	def test_blocked_writes_v2(self):
+		with suppress_stdout():
+			response = self.post(
+				self.resource_path("ToDo"), {"description": frappe.mock("paragraph"), "sid": self.sid}
+			)
+		self.assertEqual(response.status_code, 503)
+		self.assertEqual(response.json["errors"][0]["type"], "InReadOnlyMode")
+
+
+def generate_admin_keys():
+	from frappe.core.doctype.user.user import generate_keys
+
+	generate_keys("Administrator")
+	frappe.db.commit()
+
+
+@frappe.whitelist()
+def test(*, fail=False, handled=True, message="Failed"):
+	if fail:
+		if handled:
+			frappe.throw(message)
+		else:
+			1 / 0
+	else:
+		frappe.msgprint(message)
diff --git a/frappe/utils/nestedset.py b/frappe/utils/nestedset.py
index 01b7576876..2beaa63447 100644
--- a/frappe/utils/nestedset.py
+++ b/frappe/utils/nestedset.py
@@ -292,7 +292,7 @@ class NestedSet(Document):
 			update_nsm(self)
 		except frappe.DoesNotExistError:
 			if self.flags.on_rollback:
-				frappe.message_log.pop()
+				frappe.clear_last_message()
 			else:
 				raise
 
diff --git a/frappe/utils/response.py b/frappe/utils/response.py
index fcd7b49a0f..3fcf1a365e 100644
--- a/frappe/utils/response.py
+++ b/frappe/utils/response.py
@@ -6,7 +6,8 @@ import decimal
 import json
 import mimetypes
 import os
-from typing import TYPE_CHECKING
+import sys
+from typing import TYPE_CHECKING, Literal
 from urllib.parse import quote
 
 import werkzeug.utils
@@ -21,7 +22,7 @@ import frappe.sessions
 import frappe.utils
 from frappe import _
 from frappe.core.doctype.access_log.access_log import make_access_log
-from frappe.utils import cint, format_timedelta
+from frappe.utils import format_timedelta
 
 if TYPE_CHECKING:
 	from frappe.core.doctype.file.file import File
@@ -29,22 +30,45 @@ if TYPE_CHECKING:
 
 def report_error(status_code):
 	"""Build error. Show traceback in developer mode"""
-	allow_traceback = frappe.get_system_settings("allow_error_traceback") if frappe.db else False
-	if (
-		allow_traceback
-		and (status_code != 404 or frappe.conf.logging)
+	from frappe.api import ApiVersion, get_api_version
+
+	allow_traceback = (
+		(frappe.get_system_settings("allow_error_traceback") if frappe.db else False)
 		and not frappe.local.flags.disable_traceback
-	):
-		traceback = frappe.utils.get_traceback()
-		if traceback:
-			frappe.errprint(traceback)
-			frappe.local.response.exception = traceback.splitlines()[-1]
+		and (status_code != 404 or frappe.conf.logging)
+	)
+
+	traceback = frappe.utils.get_traceback()
+	exc_type, exc_value, _ = sys.exc_info()
+
+	match get_api_version():
+		case ApiVersion.V1:
+			if allow_traceback:
+				frappe.errprint(traceback)
+				frappe.response.exception = traceback.splitlines()[-1]
+			frappe.response["exc_type"] = exc_type.__name__
+		case ApiVersion.V2:
+			error_log = {"type": exc_type.__name__}
+			if allow_traceback:
+				error_log["exception"] = traceback
+			_link_error_with_message_log(error_log, exc_value, frappe.message_log)
+			frappe.local.response.errors = [error_log]
 
 	response = build_response("json")
 	response.status_code = status_code
 	return response
 
 
+def _link_error_with_message_log(error_log, exception, message_logs):
+	for message in message_logs:
+		if message.get("__frappe_exc_id") == exception.__frappe_exc_id:
+			error_log.update(message)
+			message_logs.remove(message)
+			error_log.pop("raise_exception", None)
+			error_log.pop("__frappe_exc_id", None)
+			return
+
+
 def build_response(response_type=None):
 	if "docs" in frappe.local.response and not frappe.local.response.docs:
 		del frappe.local.response["docs"]
@@ -99,6 +123,7 @@ def as_raw():
 
 def as_json():
 	make_logs()
+
 	response = Response()
 	if frappe.local.response.http_status_code:
 		response.status_code = frappe.local.response["http_status_code"]
@@ -125,12 +150,22 @@ def as_binary():
 	return response
 
 
-def make_logs(response=None):
+def make_logs():
 	"""make strings for msgprint and errprint"""
+
+	from frappe.api import ApiVersion, get_api_version
+
+	match get_api_version():
+		case ApiVersion.V1:
+			_make_logs_v1()
+		case ApiVersion.V2:
+			_make_logs_v2()
+
+
+def _make_logs_v1():
 	from frappe.utils.error import guess_exception_source
 
-	if not response:
-		response = frappe.local.response
+	response = frappe.local.response
 
 	if frappe.error_log:
 		if source := guess_exception_source(frappe.local.error_log and frappe.local.error_log[0]["exc"]):
@@ -138,17 +173,25 @@ def make_logs(response=None):
 		response["exc"] = json.dumps([frappe.utils.cstr(d["exc"]) for d in frappe.local.error_log])
 
 	if frappe.local.message_log:
-		response["_server_messages"] = json.dumps(
-			[frappe.utils.cstr(d) for d in frappe.local.message_log]
-		)
+		response["_server_messages"] = json.dumps([json.dumps(d) for d in frappe.local.message_log])
 
-	if frappe.debug_log and frappe.conf.get("logging") or False:
+	if frappe.debug_log and frappe.conf.get("logging"):
 		response["_debug_messages"] = json.dumps(frappe.local.debug_log)
 
 	if frappe.flags.error_message:
 		response["_error_message"] = frappe.flags.error_message
 
 
+def _make_logs_v2():
+	response = frappe.local.response
+
+	if frappe.local.message_log:
+		response["messages"] = frappe.local.message_log
+
+	if frappe.debug_log and frappe.conf.get("logging"):
+		response["debug"] = [{"message": m} for m in frappe.local.debug_log]
+
+
 def json_handler(obj):
 	"""serialize non-serializable data for json"""
 	from collections.abc import Iterable
diff --git a/frappe/utils/typing_validations.py b/frappe/utils/typing_validations.py
index 3f24c5eb19..cd8e736fe6 100644
--- a/frappe/utils/typing_validations.py
+++ b/frappe/utils/typing_validations.py
@@ -1,7 +1,7 @@
 from collections.abc import Callable
 from functools import lru_cache, wraps
 from inspect import _empty, isclass, signature
-from types import EllipsisType
+from types import EllipsisType, NoneType
 from typing import ForwardRef, TypeVar, Union
 
 from pydantic import ConfigDict
diff --git a/frappe/website/doctype/web_page_view/web_page_view.py b/frappe/website/doctype/web_page_view/web_page_view.py
index 21b1c00a1f..7c18d1ff66 100644
--- a/frappe/website/doctype/web_page_view/web_page_view.py
+++ b/frappe/website/doctype/web_page_view/web_page_view.py
@@ -93,8 +93,7 @@ def make_view_log(
 		else:
 			view.insert(ignore_permissions=True)
 	except Exception:
-		if frappe.message_log:
-			frappe.message_log.pop()
+		frappe.clear_last_message()
 
 
 @frappe.whitelist()