diff --git a/src/master/http.cpp b/src/master/http.cpp
index 16dbed3..8a58863 100644
--- a/src/master/http.cpp
+++ b/src/master/http.cpp
@@ -539,7 +539,7 @@
     field = v1::master::Event::Subscribed::kGetStateFieldNumber;
     writer->field(
         descriptor->FindFieldByNumber(field)->name(),
-        jsonifyGetState(master, approvers));
+        Master::ReadOnlyHandler(master).jsonifyGetState(approvers));
 
     field = v1::master::Event::Subscribed::kHeartbeatIntervalSecondsFieldNumber;
     writer->field(
@@ -565,7 +565,7 @@
 
   WireFormatLite::WriteBytes(
       mesos::v1::master::Event::Subscribed::kGetStateFieldNumber,
-      serializeGetState(approvers),
+      Master::ReadOnlyHandler(master).serializeGetState(approvers),
       &writer);
 
   WireFormatLite::WriteDouble(
@@ -1392,6 +1392,7 @@
           return deferBatchedRequest(
               &Master::ReadOnlyHandler::frameworks,
               principal,
+              ContentType::JSON,
               request.url.query,
               approvers);
         }));
@@ -1451,7 +1452,7 @@
 Future<Response> Master::Http::getFrameworks(
     const mesos::master::Call& call,
     const Option<Principal>& principal,
-    ContentType contentType) const
+    ContentType outputContentType) const
 {
   CHECK_EQ(mesos::master::Call::GET_FRAMEWORKS, call.type());
 
@@ -1460,182 +1461,16 @@
       principal,
       {VIEW_FRAMEWORK})
     .then(defer(
-        master->self(),
-        [=](const Owned<ObjectApprovers>& approvers) -> Future<Response> {
-          // Serialize the following message:
-          //
-          //   mesos::master::Response response;
-          //   response.set_type(mesos::master::Response::GET_FRAMEWORKS);
-          //   *response.mutable_get_frameworks() = _getFrameworks(approvers);
-
-          switch (contentType) {
-            case ContentType::PROTOBUF: {
-              string output;
-              google::protobuf::io::StringOutputStream stream(&output);
-              google::protobuf::io::CodedOutputStream writer(&stream);
-
-              WireFormatLite::WriteEnum(
-                  mesos::v1::master::Response::kTypeFieldNumber,
-                  mesos::v1::master::Response::GET_FRAMEWORKS,
-                  &writer);
-
-              WireFormatLite::WriteBytes(
-                  mesos::v1::master::Response::kGetFrameworksFieldNumber,
-                  serializeGetFrameworks(approvers),
-                  &writer);
-
-              // We must manually trim the unused buffer space since
-              // we use the string before the coded output stream is
-              // destructed.
-              writer.Trim();
-
-              return OK(std::move(output), stringify(contentType));
-            }
-
-            case ContentType::JSON: {
-              string body = jsonify([&](JSON::ObjectWriter* writer) {
-                const google::protobuf::Descriptor* descriptor =
-                  v1::master::Response::descriptor();
-
-                int field;
-
-                field = v1::master::Response::kTypeFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    v1::master::Response::Type_Name(
-                        v1::master::Response::GET_FRAMEWORKS));
-
-                field = v1::master::Response::kGetFrameworksFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    jsonifyGetFrameworks(master, approvers));
-              });
-
-              // TODO(bmahler): Pass jsonp query parameter through here.
-              return OK(std::move(body), stringify(contentType));
-            }
-
-            default:
-              return NotAcceptable("Request must accept json or protobuf");
-          }
-      }));
-}
-
-
-function<void(JSON::ObjectWriter*)> Master::Http::jsonifyGetFrameworks(
-    const Master* master,
-    const Owned<ObjectApprovers>& approvers)
-{
-  // Serialize the following:
-  //
-  //   mesos::master::Response::GetFrameworks getFrameworks;
-  //   for each framework:
-  //     *getFrameworks.add_frameworks() = model(*framework);
-  //   for each completed framework:
-  //     *getFrameworks.add_completed_frameworks() = model(*framework);
-
-  // TODO(bmahler): Consider not constructing the temporary framework
-  // objects and instead serialize directly, but since we don't
-  // expect a large number of pending tasks, we currently don't
-  // bother with the more efficient approach.
-
-  // TODO(bmahler): This copies the Owned object approvers.
-  return [=](JSON::ObjectWriter* writer) {
-    const google::protobuf::Descriptor* descriptor =
-      v1::master::Response::GetFrameworks::descriptor();
-
-    int field;
-
-    field = v1::master::Response::GetFrameworks::kFrameworksFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-      foreachvalue (const Framework* framework,
-                    master->frameworks.registered) {
-        // Skip unauthorized frameworks.
-        if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-          continue;
-        }
-
-        mesos::master::Response::GetFrameworks::Framework f = model(*framework);
-        writer->element(asV1Protobuf(f));
-      }
-    });
-
-    field =
-      v1::master::Response::GetFrameworks::kCompletedFrameworksFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-      foreachvalue (const Owned<Framework>& framework,
-                    master->frameworks.completed) {
-        // Skip unauthorized frameworks.
-        if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-          continue;
-        }
-
-        mesos::master::Response::GetFrameworks::Framework f = model(*framework);
-        writer->element(asV1Protobuf(f));
-      }
-    });
-  };
-}
-
-
-string Master::Http::serializeGetFrameworks(
-    const Owned<ObjectApprovers>& approvers) const
-{
-  // Serialize the following:
-  //
-  //   mesos::master::Response::GetFrameworks getFrameworks;
-  //   for each framework:
-  //     *getFrameworks.add_frameworks() = model(*framework);
-  //   for each completed framework:
-  //     *getFrameworks.add_completed_frameworks() = model(*framework);
-
-  string output;
-  google::protobuf::io::StringOutputStream stream(&output);
-  google::protobuf::io::CodedOutputStream writer(&stream);
-
-  // TODO(bmahler): Consider not constructing the temporary framework
-  // objects and instead serialize directly, but since we don't
-  // expect a large number of pending tasks, we currently don't
-  // bother with the more efficient approach.
-
-  foreachvalue (const Framework* framework,
-                master->frameworks.registered) {
-    // Skip unauthorized frameworks.
-    if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-      continue;
-    }
-
-    WireFormatLite2::WriteMessageWithoutCachedSizes(
-        mesos::master::Response::GetFrameworks::kFrameworksFieldNumber,
-        model(*framework),
-        &writer);
-  }
-
-  foreachvalue (const Owned<Framework>& framework,
-                master->frameworks.completed) {
-    // Skip unauthorized frameworks.
-    if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-      continue;
-    }
-
-    WireFormatLite2::WriteMessageWithoutCachedSizes(
-        mesos::master::Response::GetFrameworks::kCompletedFrameworksFieldNumber,
-        model(*framework),
-        &writer);
-  }
-
-  // While an explicit Trim() isn't necessary (since the coded
-  // output stream is destructed before the string is returned),
-  // it's a quite tricky bug to diagnose if Trim() is missed, so
-  // we always do it explicitly to signal the reader about this
-  // subtlety.
-  writer.Trim();
-
-  return output;
+          master->self(),
+          [this, principal, outputContentType](
+              const Owned<ObjectApprovers>& approvers) {
+            return deferBatchedRequest(
+                &Master::ReadOnlyHandler::getFrameworks,
+                principal,
+                outputContentType,
+                {},
+                approvers);
+          }));
 }
 
 
@@ -1670,7 +1505,7 @@
 Future<Response> Master::Http::getExecutors(
     const mesos::master::Call& call,
     const Option<Principal>& principal,
-    ContentType contentType) const
+    ContentType outputContentType) const
 {
   CHECK_EQ(mesos::master::Call::GET_EXECUTORS, call.type());
 
@@ -1679,243 +1514,16 @@
       principal,
       {VIEW_FRAMEWORK, VIEW_EXECUTOR})
     .then(defer(
-        master->self(),
-        [=](const Owned<ObjectApprovers>& approvers) -> Response {
-          // Serialize the following message:
-          //
-          //   mesos::master::Response response;
-          //   response.set_type(mesos::master::Response::GET_EXECUTORS);
-          //   *response.mutable_get_executors() = _getExecutors(approvers);
-
-          switch (contentType) {
-            case ContentType::PROTOBUF: {
-              string output;
-              google::protobuf::io::StringOutputStream stream(&output);
-              google::protobuf::io::CodedOutputStream writer(&stream);
-
-              WireFormatLite::WriteEnum(
-                  mesos::v1::master::Response::kTypeFieldNumber,
-                  mesos::v1::master::Response::GET_EXECUTORS,
-                  &writer);
-
-              WireFormatLite::WriteBytes(
-                  mesos::v1::master::Response::kGetExecutorsFieldNumber,
-                  serializeGetExecutors(approvers),
-                  &writer);
-
-              // We must manually trim the unused buffer space since
-              // we use the string before the coded output stream is
-              // destructed.
-              writer.Trim();
-
-              return OK(std::move(output), stringify(contentType));
-            }
-
-            case ContentType::JSON: {
-              string body = jsonify([&](JSON::ObjectWriter* writer) {
-                const google::protobuf::Descriptor* descriptor =
-                  v1::master::Response::descriptor();
-
-                int field;
-
-                field = v1::master::Response::kTypeFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    v1::master::Response::Type_Name(
-                        v1::master::Response::GET_EXECUTORS));
-
-                field = v1::master::Response::kGetExecutorsFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    jsonifyGetExecutors(master, approvers));
-              });
-
-              // TODO(bmahler): Pass jsonp query parameter through here.
-              return OK(std::move(body), stringify(contentType));
-            }
-
-            default:
-              return NotAcceptable("Request must accept json or protobuf");
-          }
-      }));
-}
-
-
-function<void(JSON::ObjectWriter*)> Master::Http::jsonifyGetExecutors(
-    const Master* master,
-    const Owned<ObjectApprovers>& approvers)
-{
-  // Serialize the following:
-  //
-  //   mesos::master::Response::GetExecutors getExecutors;
-  //
-  //   for each (executor, agent):
-  //     mesos::master::Response::GetExecutors::Executor* executor =
-  //       getExecutors.add_executors();
-  //     *executor->mutable_executor_info() = executorInfo;
-  //     *executor->mutable_slave_id() = slaveId;
-
-  // TODO(bmahler): This copies the owned object approvers.
-  return [=](JSON::ObjectWriter* writer) {
-    // Construct framework list with both active and completed frameworks.
-    vector<const Framework*> frameworks;
-    foreachvalue (Framework* framework, master->frameworks.registered) {
-      // Skip unauthorized frameworks.
-      if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-        frameworks.push_back(framework);
-      }
-    }
-    foreachvalue (const Owned<Framework>& framework,
-                  master->frameworks.completed) {
-      // Skip unauthorized frameworks.
-      if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-        frameworks.push_back(framework.get());
-      }
-    }
-
-    const google::protobuf::Descriptor* descriptor =
-      v1::master::Response::GetExecutors::descriptor();
-
-    int field;
-
-    field = v1::master::Response::GetExecutors::kExecutorsFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-      foreach (const Framework* framework, frameworks) {
-        foreachpair (const SlaveID& slaveId,
-                     const auto& executorsMap,
-                     framework->executors) {
-          foreachvalue (const ExecutorInfo& executorInfo, executorsMap) {
-            // Skip unauthorized executors.
-            if (!approvers->approved<VIEW_EXECUTOR>(
-                    executorInfo, framework->info)) {
-              continue;
-            }
-
-            writer->element([&](JSON::ObjectWriter* writer) {
-              const google::protobuf::Descriptor* descriptor =
-                v1::master::Response::GetExecutors::Executor::descriptor();
-
-              // Serialize the following message:
-              //
-              //   mesos::master::Response::GetExecutors::Executor executor;
-              //   *executor.mutable_executor_info() = executorInfo;
-              //   *executor.mutable_slave_id() = slaveId;
-              int field;
-
-              field = v1::master::Response::GetExecutors::Executor
-                ::kExecutorInfoFieldNumber;
-              writer->field(
-                  descriptor->FindFieldByNumber(field)->name(),
-                  asV1Protobuf(executorInfo));
-
-              field = v1::master::Response::GetExecutors::Executor
-                ::kAgentIdFieldNumber;
-              writer->field(
-                  descriptor->FindFieldByNumber(field)->name(),
-                  asV1Protobuf(slaveId));
-            });
-          }
-        }
-      }
-    });
-  };
-}
-
-
-string Master::Http::serializeGetExecutors(
-    const Owned<ObjectApprovers>& approvers) const
-{
-  // Construct framework list with both active and completed frameworks.
-  vector<const Framework*> frameworks;
-  foreachvalue (Framework* framework, master->frameworks.registered) {
-    // Skip unauthorized frameworks.
-    if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-      frameworks.push_back(framework);
-    }
-  }
-  foreachvalue (const Owned<Framework>& framework,
-                master->frameworks.completed) {
-    // Skip unauthorized frameworks.
-    if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-      frameworks.push_back(framework.get());
-    }
-  }
-
-  // Lambda for serializing the following message:
-  //
-  //   mesos::master::Response::GetExecutors::Executor executor;
-  //   *executor.mutable_executor_info() = executorInfo;
-  //   *executor.mutable_slave_id() = slaveId;
-  auto serializeExecutor = [](const ExecutorInfo& e, const SlaveID& s) {
-    string output;
-    google::protobuf::io::StringOutputStream stream(&output);
-    google::protobuf::io::CodedOutputStream writer(&stream);
-
-    WireFormatLite2::WriteMessageWithoutCachedSizes(
-        mesos::v1::master::Response::GetExecutors::Executor
-          ::kExecutorInfoFieldNumber,
-        e,
-        &writer);
-
-    WireFormatLite2::WriteMessageWithoutCachedSizes(
-        mesos::v1::master::Response::GetExecutors::Executor
-          ::kAgentIdFieldNumber,
-        s,
-        &writer);
-
-    // While an explicit Trim() isn't necessary (since the coded
-    // output stream is destructed before the string is returned),
-    // it's a quite tricky bug to diagnose if Trim() is missed, so
-    // we always do it explicitly to signal the reader about this
-    // subtlety.
-    writer.Trim();
-
-    return output;
-  };
-
-  string output;
-  google::protobuf::io::StringOutputStream stream(&output);
-  google::protobuf::io::CodedOutputStream writer(&stream);
-
-  // Serialize the following:
-  //
-  //   mesos::master::Response::GetExecutors getExecutors;
-  //
-  //   for each (executor, agent):
-  //     mesos::master::Response::GetExecutors::Executor* executor =
-  //       getExecutors.add_executors();
-  //     *executor->mutable_executor_info() = executorInfo;
-  //     *executor->mutable_slave_id() = slaveId;
-
-  foreach (const Framework* framework, frameworks) {
-    foreachpair (const SlaveID& slaveId,
-                 const auto& executorsMap,
-                 framework->executors) {
-      foreachvalue (const ExecutorInfo& executorInfo, executorsMap) {
-        // Skip unauthorized executors.
-        if (!approvers->approved<VIEW_EXECUTOR>(
-                executorInfo, framework->info)) {
-          continue;
-        }
-
-        WireFormatLite::WriteBytes(
-            mesos::v1::master::Response::GetExecutors::kExecutorsFieldNumber,
-            serializeExecutor(executorInfo, slaveId),
-            &writer);
-      }
-    }
-  }
-
-  // While an explicit Trim() isn't necessary (since the coded
-  // output stream is destructed before the string is returned),
-  // it's a quite tricky bug to diagnose if Trim() is missed, so
-  // we always do it explicitly to signal the reader about this
-  // subtlety.
-  writer.Trim();
-
-  return output;
+          master->self(),
+          [this, principal, outputContentType](
+             const Owned<ObjectApprovers>& approvers) {
+            return deferBatchedRequest(
+                &Master::ReadOnlyHandler::getExecutors,
+                principal,
+                outputContentType,
+                {},
+                approvers);
+          }));
 }
 
 
@@ -1972,7 +1580,7 @@
 Future<Response> Master::Http::getState(
     const mesos::master::Call& call,
     const Option<Principal>& principal,
-    ContentType contentType) const
+    ContentType outputContentType) const
 {
   CHECK_EQ(mesos::master::Call::GET_STATE, call.type());
 
@@ -1981,153 +1589,16 @@
       principal,
       {VIEW_FRAMEWORK, VIEW_TASK, VIEW_EXECUTOR, VIEW_ROLE})
     .then(defer(
-        master->self(),
-        [=](const Owned<ObjectApprovers>& approvers) -> Response {
-          // Serialize the following message:
-          //
-          //   mesos::master::Response response;
-          //   response.set_type(mesos::master::Response::GET_STATE);
-          //   *response.mutable_get_state() = _getState(approvers);
-
-          switch (contentType) {
-            case ContentType::PROTOBUF: {
-              string output;
-              google::protobuf::io::StringOutputStream stream(&output);
-              google::protobuf::io::CodedOutputStream writer(&stream);
-
-              WireFormatLite::WriteEnum(
-                  mesos::v1::master::Response::kTypeFieldNumber,
-                  mesos::v1::master::Response::GET_STATE,
-                  &writer);
-
-              WireFormatLite::WriteBytes(
-                  mesos::v1::master::Response::kGetStateFieldNumber,
-                  serializeGetState(approvers),
-                  &writer);
-
-              // We must manually trim the unused buffer space since
-              // we use the string before the coded output stream is
-              // destructed.
-              writer.Trim();
-
-              return OK(std::move(output), stringify(contentType));
-            }
-
-            case ContentType::JSON: {
-              string body = jsonify([&](JSON::ObjectWriter* writer) {
-                const google::protobuf::Descriptor* descriptor =
-                  v1::master::Response::descriptor();
-
-                int field;
-
-                field = v1::master::Response::kTypeFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    v1::master::Response::Type_Name(
-                        v1::master::Response::GET_STATE));
-
-                field = v1::master::Response::kGetStateFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    jsonifyGetState(master, approvers));
-              });
-
-              // TODO(bmahler): Pass jsonp query parameter through here.
-              return OK(std::move(body), stringify(contentType));
-            }
-
-            default:
-              return NotAcceptable("Request must accept json or protobuf");
-          }
-        }));
-}
-
-
-function<void(JSON::ObjectWriter*)> Master::Http::jsonifyGetState(
-    const Master* master,
-    const Owned<ObjectApprovers>& approvers)
-{
-  // Jsonify the following message:
-  //
-  //   mesos::master::Response::GetState getState;
-  //   *getState.mutable_get_tasks() = _getTasks(approvers);
-  //   *getState.mutable_get_executors() = _getExecutors(approvers);
-  //   *getState.mutable_get_frameworks() = _getFrameworks(approvers);
-  //   *getState.mutable_get_agents() = _getAgents(approvers);
-
-  // TODO(bmahler): This copies the Owned object approvers.
-  return [=](JSON::ObjectWriter* writer) {
-    const google::protobuf::Descriptor* descriptor =
-      v1::master::Response::GetState::descriptor();
-
-    int field;
-
-    field = v1::master::Response::GetState::kGetTasksFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        jsonifyGetTasks(master, approvers));
-
-    field = v1::master::Response::GetState::kGetExecutorsFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        jsonifyGetExecutors(master, approvers));
-
-    field = v1::master::Response::GetState::kGetFrameworksFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        jsonifyGetFrameworks(master, approvers));
-
-    field = v1::master::Response::GetState::kGetAgentsFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        jsonifyGetAgents(master, approvers));
-  };
-}
-
-
-string Master::Http::serializeGetState(
-    const Owned<ObjectApprovers>& approvers) const
-{
-  // Serialize the following message:
-  //
-  //   mesos::master::Response::GetState getState;
-  //   *getState.mutable_get_tasks() = _getTasks(approvers);
-  //   *getState.mutable_get_executors() = _getExecutors(approvers);
-  //   *getState.mutable_get_frameworks() = _getFrameworks(approvers);
-  //   *getState.mutable_get_agents() = _getAgents(approvers);
-
-  string output;
-  google::protobuf::io::StringOutputStream stream(&output);
-  google::protobuf::io::CodedOutputStream writer(&stream);
-
-  WireFormatLite::WriteBytes(
-      mesos::v1::master::Response::GetState::kGetTasksFieldNumber,
-      serializeGetTasks(approvers),
-      &writer);
-
-  WireFormatLite::WriteBytes(
-      mesos::v1::master::Response::GetState::kGetExecutorsFieldNumber,
-      serializeGetExecutors(approvers),
-      &writer);
-
-  WireFormatLite::WriteBytes(
-      mesos::v1::master::Response::GetState::kGetFrameworksFieldNumber,
-      serializeGetFrameworks(approvers),
-      &writer);
-
-  WireFormatLite::WriteBytes(
-      mesos::v1::master::Response::GetState::kGetAgentsFieldNumber,
-      serializeGetAgents(approvers),
-      &writer);
-
-  // While an explicit Trim() isn't necessary (since the coded
-  // output stream is destructed before the string is returned),
-  // it's a quite tricky bug to diagnose if Trim() is missed, so
-  // we always do it explicitly to signal the reader about this
-  // subtlety.
-  writer.Trim();
-
-  return output;
+          master->self(),
+          [this, principal, outputContentType](
+              const Owned<ObjectApprovers>& approvers) {
+            return deferBatchedRequest(
+                &Master::ReadOnlyHandler::getState,
+                principal,
+                outputContentType,
+                {},
+                approvers);
+          }));
 }
 
 
@@ -2896,6 +2367,7 @@
           return deferBatchedRequest(
               &Master::ReadOnlyHandler::slaves,
               principal,
+              ContentType::JSON,
               request.url.query,
               approvers);
         }));
@@ -2905,204 +2377,22 @@
 Future<Response> Master::Http::getAgents(
     const mesos::master::Call& call,
     const Option<Principal>& principal,
-    ContentType contentType) const
+    ContentType outputContentType) const
 {
   CHECK_EQ(mesos::master::Call::GET_AGENTS, call.type());
 
   return ObjectApprovers::create(master->authorizer, principal, {VIEW_ROLE})
     .then(defer(
-        master->self(),
-        [=](const Owned<ObjectApprovers>& approvers) -> Response {
-          // Serialize the following message:
-          //
-          //   mesos::master::Response response;
-          //   response.set_type(mesos::master::Response::GET_AGENTS);
-          //   *response.mutable_get_agents() = _getAgents(approvers);
-
-          switch (contentType) {
-            case ContentType::PROTOBUF: {
-              string output;
-              google::protobuf::io::StringOutputStream stream(&output);
-              google::protobuf::io::CodedOutputStream writer(&stream);
-
-              WireFormatLite::WriteEnum(
-                  mesos::v1::master::Response::kTypeFieldNumber,
-                  mesos::v1::master::Response::GET_AGENTS,
-                  &writer);
-
-              WireFormatLite::WriteBytes(
-                  mesos::v1::master::Response::kGetAgentsFieldNumber,
-                  serializeGetAgents(approvers),
-                  &writer);
-
-              // We must manually trim the unused buffer space since
-              // we use the string before the coded output stream is
-              // destructed.
-              writer.Trim();
-
-              return OK(std::move(output), stringify(contentType));
-            }
-
-            case ContentType::JSON: {
-              string body = jsonify([&](JSON::ObjectWriter* writer) {
-                const google::protobuf::Descriptor* descriptor =
-                  v1::master::Response::descriptor();
-
-                int field;
-
-                field = v1::master::Response::kTypeFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    v1::master::Response::Type_Name(
-                        v1::master::Response::GET_AGENTS));
-
-                field = v1::master::Response::kGetAgentsFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    jsonifyGetAgents(master, approvers));
-              });
-
-              // TODO(bmahler): Pass jsonp query parameter through here.
-              return OK(std::move(body), stringify(contentType));
-            }
-
-            default:
-              return NotAcceptable("Request must accept json or protobuf");
-          }
-    }));
-}
-
-
-function<void(JSON::ObjectWriter*)> Master::Http::jsonifyGetAgents(
-    const Master* master,
-    const Owned<ObjectApprovers>& approvers)
-{
-  // Serialize the following:
-  //
-  //   mesos::master::Response::GetAgents getAgents;
-  //   for each registered agent:
-  //     *getAgents.add_agents() = protobuf::master::event::createAgentResponse(
-  //         agent,
-  //         master->slaves.draining.get(slave->id),
-  //         master->slaves.deactivated.contains(slave->id),
-  //         approvers);
-  //   for each recovered agent:
-  //     SlaveInfo* agent = getAgents.add_recovered_agents();
-  //     agent->CopyFrom(slaveInfo);
-  //     agent->clear_resources();
-  //     foreach (const Resource& resource, slaveInfo.resources()):
-  //       if (approvers->approved<VIEW_ROLE>(resource)):
-  //         *agent->add_resources() = resource;
-
-  // TODO(bmahler): This copies the Owned object approvers.
-  return [=](JSON::ObjectWriter* writer) {
-    const google::protobuf::Descriptor* descriptor =
-      v1::master::Response::GetAgents::descriptor();
-
-    int field;
-
-    field = v1::master::Response::GetAgents::kAgentsFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-      foreachvalue (const Slave* slave, master->slaves.registered) {
-        // TODO(bmahler): Consider not constructing the temporary
-        // agent object and instead serialize directly.
-        mesos::master::Response::GetAgents::Agent agent =
-            protobuf::master::event::createAgentResponse(
-                *slave,
-                master->slaves.draining.get(slave->id),
-                master->slaves.deactivated.contains(slave->id),
+          master->self(),
+          [this, principal, outputContentType](
+              const Owned<ObjectApprovers>& approvers) {
+            return deferBatchedRequest(
+                &Master::ReadOnlyHandler::getAgents,
+                principal,
+                outputContentType,
+                {},
                 approvers);
-
-        writer->element(asV1Protobuf(agent));
-      }
-    });
-
-    field = v1::master::Response::GetAgents::kRecoveredAgentsFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-      foreachvalue (const SlaveInfo& slaveInfo, master->slaves.recovered) {
-        // TODO(bmahler): Consider not constructing the temporary
-        // SlaveInfo object and instead serialize directly.
-        SlaveInfo agent = slaveInfo;
-        agent.clear_resources();
-        foreach (const Resource& resource, slaveInfo.resources()) {
-          if (approvers->approved<VIEW_ROLE>(resource)) {
-            *agent.add_resources() = resource;
-          }
-        }
-
-        writer->element(asV1Protobuf(agent));
-      }
-    });
-  };
-}
-
-
-string Master::Http::serializeGetAgents(
-    const Owned<ObjectApprovers>& approvers) const
-{
-  // Serialize the following:
-  //
-  //   mesos::master::Response::GetAgents getAgents;
-  //   for each registered agent:
-  //     *getAgents.add_agents() = protobuf::master::event::createAgentResponse(
-  //         agent,
-  //         master->slaves.draining.get(slave->id),
-  //         master->slaves.deactivated.contains(slave->id),
-  //         approvers);
-  //   for each recovered agent:
-  //     SlaveInfo* agent = getAgents.add_recovered_agents();
-  //     agent->CopyFrom(slaveInfo);
-  //     agent->clear_resources();
-  //     foreach (const Resource& resource, slaveInfo.resources()):
-  //       if (approvers->approved<VIEW_ROLE>(resource)):
-  //         *agent->add_resources() = resource;
-
-  string output;
-  google::protobuf::io::StringOutputStream stream(&output);
-  google::protobuf::io::CodedOutputStream writer(&stream);
-
-  foreachvalue (const Slave* slave, master->slaves.registered) {
-    // TODO(bmahler): Consider not constructing the temporary
-    // agent object and instead serialize directly.
-    WireFormatLite2::WriteMessageWithoutCachedSizes(
-        mesos::master::Response::GetAgents::kAgentsFieldNumber,
-        protobuf::master::event::createAgentResponse(
-            *slave,
-            master->slaves.draining.get(slave->id),
-            master->slaves.deactivated.contains(slave->id),
-            approvers),
-        &writer);
-  }
-
-  foreachvalue (const SlaveInfo& slaveInfo, master->slaves.recovered) {
-    // TODO(bmahler): Consider not constructing the temporary
-    // SlaveInfo object and instead serialize directly.
-    SlaveInfo agent = slaveInfo;
-    agent.clear_resources();
-    foreach (const Resource& resource, slaveInfo.resources()) {
-      if (approvers->approved<VIEW_ROLE>(resource)) {
-        *agent.add_resources() = resource;
-      }
-    }
-
-    WireFormatLite2::WriteMessageWithoutCachedSizes(
-        mesos::master::Response::GetAgents::kRecoveredAgentsFieldNumber,
-        agent,
-        &writer);
-  }
-
-  // While an explicit Trim() isn't necessary (since the coded
-  // output stream is destructed before the string is returned),
-  // it's a quite tricky bug to diagnose if Trim() is missed, so
-  // we always do it explicitly to signal the reader about this
-  // subtlety.
-  writer.Trim();
-
-  return output;
+          }));
 }
 
 
@@ -3392,6 +2682,7 @@
           return deferBatchedRequest(
               &Master::ReadOnlyHandler::state,
               principal,
+              ContentType::JSON,
               request.url.query,
               approvers);
         }));
@@ -3401,13 +2692,14 @@
 Future<Response> Master::Http::deferBatchedRequest(
     ReadOnlyRequestHandler handler,
     const Option<Principal>& principal,
+    ContentType outputContentType,
     const hashmap<std::string, std::string>& queryParameters,
     const Owned<ObjectApprovers>& approvers) const
 {
   bool scheduleBatch = batchedRequests.empty();
 
   auto it = std::find_if(batchedRequests.begin(), batchedRequests.end(),
-      [handler, &principal, &queryParameters](
+      [handler, &principal, &queryParameters, &outputContentType](
           const BatchedRequest& batchedRequest) {
         // NOTE: This is not a general-purpose request comparison, but
         // specific to the batched requests which are always members of
@@ -3415,6 +2707,7 @@
         // on query parameters and the current master state.
         return handler == batchedRequest.handler &&
                principal == batchedRequest.principal &&
+               outputContentType == batchedRequest.outputContentType &&
                queryParameters == batchedRequest.queryParameters;
       });
 
@@ -3434,6 +2727,7 @@
     future = promise.future();
     batchedRequests.push_back(BatchedRequest{
         handler,
+        outputContentType,
         queryParameters,
         principal,
         approvers,
@@ -3466,11 +2760,16 @@
   foreach (BatchedRequest& request, batchedRequests) {
     request.promise.associate(process::async(
         [this](ReadOnlyRequestHandler handler,
+               ContentType outputContentType,
                const hashmap<std::string, std::string>& queryParameters,
                const process::Owned<ObjectApprovers>& approvers) {
-          return (readonlyHandler.*handler)(queryParameters, approvers);
+          return (readonlyHandler.*handler)(
+              outputContentType,
+              queryParameters,
+              approvers);
         },
         request.handler,
+        request.outputContentType,
         request.queryParameters,
         request.approvers));
   }
@@ -3597,6 +2896,7 @@
           return deferBatchedRequest(
               &Master::ReadOnlyHandler::stateSummary,
               principal,
+              ContentType::JSON,
               request.url.query,
               approvers);
         }));
@@ -3650,6 +2950,7 @@
             return deferBatchedRequest(
                 &Master::ReadOnlyHandler::roles,
                 principal,
+                ContentType::JSON,
                 request.url.query,
                 approvers);
           }));
@@ -3711,64 +3012,22 @@
 Future<Response> Master::Http::getRoles(
     const mesos::master::Call& call,
     const Option<Principal>& principal,
-    ContentType contentType) const
+    ContentType outputContentType) const
 {
   CHECK_EQ(mesos::master::Call::GET_ROLES, call.type());
+
   return ObjectApprovers::create(master->authorizer, principal, {VIEW_ROLE})
-    .then(defer(master->self(),
-        [this, contentType](const Owned<ObjectApprovers>& approvers)
-          -> Response {
-      const vector<string> knownRoles = master->knownRoles();
-
-      mesos::master::Response response;
-      response.set_type(mesos::master::Response::GET_ROLES);
-
-      mesos::master::Response::GetRoles* getRoles =
-        response.mutable_get_roles();
-
-      foreach (const string& name, knownRoles) {
-        if (!approvers->approved<VIEW_ROLE>(name)) {
-          continue;
-        }
-
-        mesos::Role* role = getRoles->add_roles();
-
-        role->set_name(name);
-
-        role->set_weight(master->weights.get(name).getOrElse(DEFAULT_WEIGHT));
-
-        RoleResourceBreakdown resourceBreakdown(master, name);
-
-        ResourceQuantities allocatedAndOffered =
-          resourceBreakdown.allocated() + resourceBreakdown.offered();
-
-        // `resources` will be deprecated in favor of
-        // `offered`, `allocated`, `reserved`, and quota consumption.
-        // As a result, we don't bother trying to expose more
-        // than {cpus, mem, disk, gpus} since we don't know if
-        // anything outside this set is of type SCALAR.
-        foreach (const auto& quantity, allocatedAndOffered) {
-          if (quantity.first == "cpus" || quantity.first == "mem" ||
-              quantity.first == "disk" || quantity.first == "gpus") {
-            Resource* resource = role->add_resources();
-            resource->set_name(quantity.first);
-            resource->set_type(Value::SCALAR);
-            *resource->mutable_scalar() = quantity.second;
-          }
-        }
-
-        Option<Role*> role_ = master->roles.get(name);
-
-        if (role_.isSome()) {
-          foreachkey (const FrameworkID& frameworkId, (*role_)->frameworks) {
-            *role->add_frameworks() = frameworkId;
-          }
-        }
-      }
-
-      return OK(serialize(contentType, evolve(response)),
-                stringify(contentType));
-    }));
+    .then(defer(
+          master->self(),
+          [this, principal, outputContentType](
+              const Owned<ObjectApprovers>& approvers) {
+            return deferBatchedRequest(
+                &Master::ReadOnlyHandler::getRoles,
+                principal,
+                outputContentType,
+                {},
+                approvers);
+          }));
 }
 
 
@@ -3913,66 +3172,22 @@
 Future<Response> Master::Http::getOperations(
     const mesos::master::Call& call,
     const Option<Principal>& principal,
-    ContentType contentType) const
+    ContentType outputContentType) const
 {
   CHECK_EQ(mesos::master::Call::GET_OPERATIONS, call.type());
 
   return ObjectApprovers::create(master->authorizer, principal, {VIEW_ROLE})
     .then(defer(
-        master->self(),
-        [=](const Owned<ObjectApprovers>& approvers) -> Response {
-          // We consider a principal to be authorized to view an operation if it
-          // is authorized to view the resources the operation is performed on.
-          auto approved = [&approvers](const Operation& operation) {
-            Try<Resources> consumedResources =
-              protobuf::getConsumedResources(operation.info());
-
-            if (consumedResources.isError()) {
-              LOG(WARNING)
-                << "Could not approve operation " << operation.uuid()
-                << " since its consumed resources could not be determined:"
-                << consumedResources.error();
-
-              return false;
-            }
-
-            foreach (const Resource& resource, consumedResources.get()) {
-              if (!approvers->approved<VIEW_ROLE>(resource)) {
-                return false;
-              }
-            }
-
-            return true;
-          };
-
-          mesos::master::Response response;
-          response.set_type(mesos::master::Response::GET_OPERATIONS);
-
-          mesos::master::Response::GetOperations* operations =
-            response.mutable_get_operations();
-
-          foreachvalue (const Slave* slave, master->slaves.registered) {
-            foreachvalue (Operation* operation, slave->operations) {
-              if (approved(*operation)) {
-                operations->add_operations()->CopyFrom(*operation);
-              }
-            }
-
-            foreachvalue (
-                const Slave::ResourceProvider resourceProvider,
-                slave->resourceProviders) {
-              foreachvalue (Operation* operation, resourceProvider.operations) {
-                if (approved(*operation)) {
-                  operations->add_operations()->CopyFrom(*operation);
-                }
-              }
-            }
-          }
-
-          return OK(
-              serialize(contentType, evolve(response)),
-              stringify(contentType));
-        }));
+          master->self(),
+          [this, principal, outputContentType](
+             const Owned<ObjectApprovers>& approvers) {
+            return deferBatchedRequest(
+                &Master::ReadOnlyHandler::getOperations,
+                principal,
+                outputContentType,
+                {},
+                approvers);
+          }));
 }
 
 
@@ -4043,6 +3258,7 @@
           return deferBatchedRequest(
               &Master::ReadOnlyHandler::tasks,
               principal,
+              ContentType::JSON,
               request.url.query,
               approvers);
         }));
@@ -4052,7 +3268,7 @@
 Future<Response> Master::Http::getTasks(
     const mesos::master::Call& call,
     const Option<Principal>& principal,
-    ContentType contentType) const
+    ContentType outputContentType) const
 {
   CHECK_EQ(mesos::master::Call::GET_TASKS, call.type());
 
@@ -4061,290 +3277,16 @@
       principal,
       {VIEW_FRAMEWORK, VIEW_TASK})
     .then(defer(
-        master->self(),
-        [=](const Owned<ObjectApprovers>& approvers) -> Response {
-          // Serialize the following message:
-          //
-          //    mesos::master::Response response;
-          //    response.set_type(mesos::master::Response::GET_TASKS);
-          //    *response.mutable_get_tasks() = _getTasks(approvers);
-
-          switch (contentType) {
-            case ContentType::PROTOBUF: {
-              string output;
-              google::protobuf::io::StringOutputStream stream(&output);
-              google::protobuf::io::CodedOutputStream writer(&stream);
-
-              WireFormatLite::WriteEnum(
-                  mesos::v1::master::Response::kTypeFieldNumber,
-                  mesos::v1::master::Response::GET_TASKS,
-                  &writer);
-
-              WireFormatLite::WriteBytes(
-                  mesos::v1::master::Response::kGetTasksFieldNumber,
-                  serializeGetTasks(approvers),
-                  &writer);
-
-              // We must manually trim the unused buffer space since
-              // we use the string before the coded output stream is
-              // destructed.
-              writer.Trim();
-
-              return OK(std::move(output), stringify(contentType));
-            }
-
-            case ContentType::JSON: {
-              string body = jsonify([&](JSON::ObjectWriter* writer) {
-                const google::protobuf::Descriptor* descriptor =
-                  v1::master::Response::descriptor();
-
-                int field;
-
-                field = v1::master::Response::kTypeFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    v1::master::Response::Type_Name(
-                        v1::master::Response::GET_TASKS));
-
-                field = v1::master::Response::kGetTasksFieldNumber;
-                writer->field(
-                    descriptor->FindFieldByNumber(field)->name(),
-                    jsonifyGetTasks(master, approvers));
-              });
-
-              // TODO(bmahler): Pass jsonp query parameter through here.
-              return OK(std::move(body), stringify(contentType));
-            }
-
-            default:
-              return NotAcceptable("Request must accept json or protobuf");
-          }
-    }));
-}
-
-
-function<void(JSON::ObjectWriter*)> Master::Http::jsonifyGetTasks(
-    const Master* master,
-    const Owned<ObjectApprovers>& approvers)
-{
-  // Jsonify the following message:
-  //
-  //  master::Response::GetTasks getTasks;
-  //  for each pending task:
-  //    *getTasks.add_pending_tasks() =
-  //      protobuf::createTask(taskInfo, TASK_STAGING, framework->id());
-  //  for each task:
-  //    *getTasks.add_tasks() = *task;
-  //  for each unreachable task:
-  //    *getTasks.add_unreachable_tasks() = *task;
-  //  for each completed task:
-  //    *getTasks.add_completed_tasks() = *task;
-
-  // TODO(bmahler): This copies the Owned object approvers.
-  return [=](JSON::ObjectWriter* writer) {
-    // Construct framework list with both active and completed frameworks.
-    vector<const Framework*> frameworks;
-    foreachvalue (Framework* framework, master->frameworks.registered) {
-      // Skip unauthorized frameworks.
-      if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-        frameworks.push_back(framework);
-      }
-    }
-    foreachvalue (const Owned<Framework>& framework,
-                  master->frameworks.completed) {
-      // Skip unauthorized frameworks.
-      if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-        frameworks.push_back(framework.get());
-      }
-    }
-
-    const google::protobuf::Descriptor* descriptor =
-      v1::master::Response::GetTasks::descriptor();
-
-    int field;
-
-    // Pending tasks.
-    field = v1::master::Response::GetTasks::kPendingTasksFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-          foreach (const Framework* framework, frameworks) {
-            foreachvalue (const TaskInfo& t, framework->pendingTasks) {
-              // Skip unauthorized tasks.
-              if (!approvers->approved<VIEW_TASK>(t, framework->info)) {
-                continue;
-              }
-
-              Task task =
-                  protobuf::createTask(t, TASK_STAGING, framework->id());
-
-              writer->element(asV1Protobuf(task));
-            }
-          }
-        });
-
-    // Active tasks.
-    field = v1::master::Response::GetTasks::kTasksFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-          foreach (const Framework* framework, frameworks) {
-            foreachvalue (Task* task, framework->tasks) {
-              CHECK_NOTNULL(task);
-              // Skip unauthorized tasks.
-              if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
-                continue;
-              }
-
-              writer->element(asV1Protobuf(*task));
-            }
-          }
-        });
-
-    // Unreachable tasks.
-    field = v1::master::Response::GetTasks::kUnreachableTasksFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-          foreach (const Framework* framework, frameworks) {
-            foreachvalue (const Owned<Task>& task,
-                          framework->unreachableTasks) {
-              // Skip unauthorized tasks.
-              if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
-                continue;
-              }
-
-              writer->element(asV1Protobuf(*task));
-            }
-          }
-        });
-
-    // Completed tasks.
-    field = v1::master::Response::GetTasks::kCompletedTasksFieldNumber;
-    writer->field(
-        descriptor->FindFieldByNumber(field)->name(),
-        [&](JSON::ArrayWriter* writer) {
-          foreach (const Framework* framework, frameworks) {
-            foreach (const Owned<Task>& task, framework->completedTasks) {
-              // Skip unauthorized tasks.
-              if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
-                continue;
-              }
-
-              writer->element(asV1Protobuf(*task));
-            }
-          }
-        });
-  };
-}
-
-
-string Master::Http::serializeGetTasks(
-    const Owned<ObjectApprovers>& approvers) const
-{
-  // Construct framework list with both active and completed frameworks.
-  vector<const Framework*> frameworks;
-  foreachvalue (Framework* framework, master->frameworks.registered) {
-    // Skip unauthorized frameworks.
-    if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-      frameworks.push_back(framework);
-    }
-  }
-  foreachvalue (const Owned<Framework>& framework,
-                master->frameworks.completed) {
-    // Skip unauthorized frameworks.
-    if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
-      frameworks.push_back(framework.get());
-    }
-  }
-
-  // Serialize the following message:
-  //
-  //  mesos::master::Response::GetTasks getTasks;
-  //  for each pending task:
-  //    *getTasks.add_pending_tasks() =
-  //      protobuf::createTask(taskInfo, TASK_STAGING, framework->id());
-  //  for each task:
-  //    *getTasks.add_tasks() = *task;
-  //  for each unreachable task:
-  //    *getTasks.add_unreachable_tasks() = *task;
-  //  for each completed task:
-  //    *getTasks.add_completed_tasks() = *task;
-
-  string output;
-  google::protobuf::io::StringOutputStream stream(&output);
-  google::protobuf::io::CodedOutputStream writer(&stream);
-
-  foreach (const Framework* framework, frameworks) {
-    // Pending tasks.
-    foreachvalue (const TaskInfo& taskInfo, framework->pendingTasks) {
-      // Skip unauthorized tasks.
-      if (!approvers->approved<VIEW_TASK>(taskInfo, framework->info)) {
-        continue;
-      }
-
-      // TODO(bmahler): Consider not constructing the temporary task
-      // object and instead serialize directly. Since we don't expect
-      // a large number of pending tasks, we currently don't bother
-      // with the more efficient approach.
-      //
-      // *getTasks.add_pending_tasks() =
-      //   protobuf::createTask(taskInfo, TASK_STAGING, framework->id());
-      WireFormatLite2::WriteMessageWithoutCachedSizes(
-          mesos::v1::master::Response::GetTasks::kPendingTasksFieldNumber,
-          protobuf::createTask(taskInfo, TASK_STAGING, framework->id()),
-          &writer);
-    }
-
-    // Active tasks.
-    foreachvalue (Task* task, framework->tasks) {
-      CHECK_NOTNULL(task);
-      // Skip unauthorized tasks.
-      if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
-        continue;
-      }
-
-      WireFormatLite2::WriteMessageWithoutCachedSizes(
-          mesos::v1::master::Response::GetTasks::kTasksFieldNumber,
-          *task,
-          &writer);
-    }
-
-    // Unreachable tasks.
-    foreachvalue (const Owned<Task>& task, framework->unreachableTasks) {
-      // Skip unauthorized tasks.
-      if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
-        continue;
-      }
-
-      WireFormatLite2::WriteMessageWithoutCachedSizes(
-          mesos::v1::master::Response::GetTasks::kUnreachableTasksFieldNumber,
-          *task,
-          &writer);
-    }
-
-    // Completed tasks.
-    foreach (const Owned<Task>& task, framework->completedTasks) {
-      // Skip unauthorized tasks.
-      if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
-        continue;
-      }
-
-      WireFormatLite2::WriteMessageWithoutCachedSizes(
-          mesos::v1::master::Response::GetTasks::kCompletedTasksFieldNumber,
-          *task,
-          &writer);
-    }
-  }
-
-  // While an explicit Trim() isn't necessary (since the coded
-  // output stream is destructed before the string is returned),
-  // it's a quite tricky bug to diagnose if Trim() is missed, so
-  // we always do it explicitly to signal the reader about this
-  // subtlety.
-  writer.Trim();
-
-  return output;
+          master->self(),
+          [this, principal, outputContentType](
+             const Owned<ObjectApprovers>& approvers) {
+            return deferBatchedRequest(
+                &Master::ReadOnlyHandler::getTasks,
+                principal,
+                outputContentType,
+                {},
+                approvers);
+          }));
 }
 
 
diff --git a/src/master/master.hpp b/src/master/master.hpp
index c9d417c..3074918 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -1309,34 +1309,117 @@
 
     // /frameworks
     process::http::Response frameworks(
+        ContentType outputContentType,
         const hashmap<std::string, std::string>& queryParameters,
         const process::Owned<ObjectApprovers>& approvers) const;
 
     // /roles
     process::http::Response roles(
+        ContentType outputContentType,
         const hashmap<std::string, std::string>& queryParameters,
         const process::Owned<ObjectApprovers>& approvers) const;
 
     // /slaves
     process::http::Response slaves(
+        ContentType outputContentType,
         const hashmap<std::string, std::string>& queryParameters,
         const process::Owned<ObjectApprovers>& approvers) const;
 
     // /state
     process::http::Response state(
+        ContentType outputContentType,
         const hashmap<std::string, std::string>& queryParameters,
         const process::Owned<ObjectApprovers>& approvers) const;
 
     // /state-summary
     process::http::Response stateSummary(
+        ContentType outputContentType,
         const hashmap<std::string, std::string>& queryParameters,
         const process::Owned<ObjectApprovers>& approvers) const;
 
     // /tasks
     process::http::Response tasks(
+        ContentType outputContentType,
         const hashmap<std::string, std::string>& queryParameters,
         const process::Owned<ObjectApprovers>& approvers) const;
 
+    // master::Call::GET_STATE
+    process::http::Response getState(
+        ContentType outputContentType,
+        const hashmap<std::string, std::string>& queryParameters,
+        const process::Owned<ObjectApprovers>& approvers) const;
+
+    // master::Call::GET_AGENTS
+    process::http::Response getAgents(
+        ContentType outputContentType,
+        const hashmap<std::string, std::string>& queryParameters,
+        const process::Owned<ObjectApprovers>& approvers) const;
+
+    // master::Call::GET_FRAMEWORKS
+    process::http::Response getFrameworks(
+        ContentType outputContentType,
+        const hashmap<std::string, std::string>& queryParameters,
+        const process::Owned<ObjectApprovers>& approvers) const;
+
+    // master::Call::GET_EXECUTORS
+    process::http::Response getExecutors(
+        ContentType outputContentType,
+        const hashmap<std::string, std::string>& queryParameters,
+        const process::Owned<ObjectApprovers>& approvers) const;
+
+    // master::Call::GET_OPERATIONS
+    process::http::Response getOperations(
+        ContentType outputContentType,
+        const hashmap<std::string, std::string>& queryParameters,
+        const process::Owned<ObjectApprovers>& approvers) const;
+
+    // master::Call::GET_TASKS
+    process::http::Response getTasks(
+        ContentType outputContentType,
+        const hashmap<std::string, std::string>& queryParameters,
+        const process::Owned<ObjectApprovers>& approvers) const;
+
+    // master::Call::GET_ROLES
+    process::http::Response getRoles(
+        ContentType outputContentType,
+        const hashmap<std::string, std::string>& queryParameters,
+        const process::Owned<ObjectApprovers>& approvers) const;
+
+    // TODO(bmahler): These could just live in the .cpp file,
+    // however they are shared with SUBSCRIBE which currently
+    // is not implemented as a read only handler here. Make these
+    // private or only in the .cpp file once SUBSCRIBE is moved
+    // into readonly_handler.cpp.
+    std::string serializeGetState(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::string serializeGetAgents(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::string serializeGetFrameworks(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::string serializeGetExecutors(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::string serializeGetOperations(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::string serializeGetTasks(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::string serializeGetRoles(
+        const process::Owned<ObjectApprovers>& approvers) const;
+
+    std::function<void(JSON::ObjectWriter*)> jsonifyGetState(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::function<void(JSON::ObjectWriter*)> jsonifyGetAgents(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::function<void(JSON::ObjectWriter*)> jsonifyGetFrameworks(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::function<void(JSON::ObjectWriter*)> jsonifyGetExecutors(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::function<void(JSON::ObjectWriter*)> jsonifyGetOperations(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::function<void(JSON::ObjectWriter*)> jsonifyGetTasks(
+        const process::Owned<ObjectApprovers>& approvers) const;
+    std::function<void(JSON::ObjectWriter*)> jsonifyGetRoles(
+        const process::Owned<ObjectApprovers>& approvers) const;
+
   private:
     const Master* master;
   };
@@ -1631,12 +1714,6 @@
         const mesos::master::Call& call,
         const Option<process::http::authentication::Principal>& principal,
         ContentType contentType) const;
-
-    static std::function<void(JSON::ObjectWriter*)> jsonifyGetAgents(
-        const Master* master,
-        const process::Owned<ObjectApprovers>& approvers);
-    std::string serializeGetAgents(
-        const process::Owned<ObjectApprovers>& approvers) const;
     mesos::master::Response::GetAgents _getAgents(
         const process::Owned<ObjectApprovers>& approvers) const;
 
@@ -1734,12 +1811,6 @@
         const mesos::master::Call& call,
         const Option<process::http::authentication::Principal>& principal,
         ContentType contentType) const;
-
-    static std::function<void(JSON::ObjectWriter*)> jsonifyGetTasks(
-        const Master* master,
-        const process::Owned<ObjectApprovers>& approvers);
-    std::string serializeGetTasks(
-        const process::Owned<ObjectApprovers>& approvers) const;
     mesos::master::Response::GetTasks _getTasks(
         const process::Owned<ObjectApprovers>& approvers) const;
 
@@ -1777,12 +1848,6 @@
         const mesos::master::Call& call,
         const Option<process::http::authentication::Principal>& principal,
         ContentType contentType) const;
-
-    static std::function<void(JSON::ObjectWriter*)> jsonifyGetFrameworks(
-        const Master* master,
-        const process::Owned<ObjectApprovers>& approvers);
-    std::string serializeGetFrameworks(
-        const process::Owned<ObjectApprovers>& approvers) const;
     mesos::master::Response::GetFrameworks _getFrameworks(
         const process::Owned<ObjectApprovers>& approvers) const;
 
@@ -1790,12 +1855,6 @@
         const mesos::master::Call& call,
         const Option<process::http::authentication::Principal>& principal,
         ContentType contentType) const;
-
-    static std::function<void(JSON::ObjectWriter*)> jsonifyGetExecutors(
-        const Master* master,
-        const process::Owned<ObjectApprovers>& approvers);
-    std::string serializeGetExecutors(
-        const process::Owned<ObjectApprovers>& approvers) const;
     mesos::master::Response::GetExecutors _getExecutors(
         const process::Owned<ObjectApprovers>& approvers) const;
 
@@ -1803,12 +1862,6 @@
         const mesos::master::Call& call,
         const Option<process::http::authentication::Principal>& principal,
         ContentType contentType) const;
-
-    static std::function<void(JSON::ObjectWriter*)> jsonifyGetState(
-        const Master* master,
-        const process::Owned<ObjectApprovers>& approvers);
-    std::string serializeGetState(
-        const process::Owned<ObjectApprovers>& approvers) const;
     mesos::master::Response::GetState _getState(
         const process::Owned<ObjectApprovers>& approvers) const;
 
@@ -1864,12 +1917,14 @@
 
     typedef process::http::Response
       (Master::ReadOnlyHandler::*ReadOnlyRequestHandler)(
+          ContentType,
           const hashmap<std::string, std::string>&,
           const process::Owned<ObjectApprovers>&) const;
 
     process::Future<process::http::Response> deferBatchedRequest(
         ReadOnlyRequestHandler handler,
         const Option<process::http::authentication::Principal>& principal,
+        ContentType outputContentType,
         const hashmap<std::string, std::string>& queryParameters,
         const process::Owned<ObjectApprovers>& approvers) const;
 
@@ -1878,6 +1933,7 @@
     struct BatchedRequest
     {
       ReadOnlyRequestHandler handler;
+      ContentType outputContentType;
       hashmap<std::string, std::string> queryParameters;
       Option<process::http::authentication::Principal> principal;
       process::Owned<ObjectApprovers> approvers;
@@ -2685,6 +2741,11 @@
   hashmap<FrameworkID, Framework*> frameworks;
 };
 
+
+mesos::master::Response::GetFrameworks::Framework model(
+    const Framework& framework);
+
+
 } // namespace master {
 } // namespace internal {
 } // namespace mesos {
diff --git a/src/master/readonly_handler.cpp b/src/master/readonly_handler.cpp
index 500ce3f..fbe748d 100644
--- a/src/master/readonly_handler.cpp
+++ b/src/master/readonly_handler.cpp
@@ -19,6 +19,12 @@
 #include <string>
 #include <vector>
 
+#include <google/protobuf/descriptor.h>
+#include <google/protobuf/wire_format_lite.h>
+
+#include <google/protobuf/io/coded_stream.h>
+#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
+
 #include <mesos/mesos.hpp>
 
 #include <mesos/authorizer/authorizer.hpp>
@@ -36,8 +42,11 @@
 #include "common/build.hpp"
 #include "common/http.hpp"
 
+using google::protobuf::internal::WireFormatLite;
+
 using process::Owned;
 
+using process::http::NotAcceptable;
 using process::http::OK;
 
 using mesos::authorization::VIEW_EXECUTOR;
@@ -46,6 +55,9 @@
 using mesos::authorization::VIEW_ROLE;
 using mesos::authorization::VIEW_TASK;
 
+using mesos::internal::protobuf::WireFormatLite2;
+
+using std::function;
 using std::vector;
 using std::string;
 
@@ -661,9 +673,12 @@
 
 
 process::http::Response Master::ReadOnlyHandler::frameworks(
+    ContentType outputContentType,
     const hashmap<std::string, std::string>& query,
     const process::Owned<ObjectApprovers>& approvers) const
 {
+  CHECK_EQ(outputContentType, ContentType::JSON);
+
   IDAcceptor<FrameworkID> selectFrameworkId(
       query.get("framework_id"));
 
@@ -719,9 +734,12 @@
 
 
 process::http::Response Master::ReadOnlyHandler::roles(
+    ContentType outputContentType,
     const hashmap<std::string, std::string>& query,
     const process::Owned<ObjectApprovers>& approvers) const
 {
+  CHECK_EQ(outputContentType, ContentType::JSON);
+
   const Master* master = this->master;
 
   const vector<string> knownRoles = master->knownRoles();
@@ -796,9 +814,12 @@
 
 
 process::http::Response Master::ReadOnlyHandler::slaves(
+    ContentType outputContentType,
     const hashmap<std::string, std::string>& query,
     const process::Owned<ObjectApprovers>& approvers) const
 {
+  CHECK_EQ(outputContentType, ContentType::JSON);
+
   IDAcceptor<SlaveID> selectSlaveId(query.get("slave_id"));
 
   return process::http::OK(
@@ -808,9 +829,12 @@
 
 
 process::http::Response Master::ReadOnlyHandler::state(
+    ContentType outputContentType,
     const hashmap<std::string, std::string>& query,
     const process::Owned<ObjectApprovers>& approvers) const
 {
+  CHECK_EQ(outputContentType, ContentType::JSON);
+
   const Master* master = this->master;
   auto calculateState = [master, &approvers](JSON::ObjectWriter* writer) {
     writer->field("version", MESOS_VERSION);
@@ -954,9 +978,12 @@
 
 
 process::http::Response Master::ReadOnlyHandler::stateSummary(
+    ContentType outputContentType,
     const hashmap<std::string, std::string>& query,
     const process::Owned<ObjectApprovers>& approvers) const
 {
+  CHECK_EQ(outputContentType, ContentType::JSON);
+
   const Master* master = this->master;
   auto stateSummary = [master, &approvers](JSON::ObjectWriter* writer) {
     writer->field("hostname", master->info().hostname());
@@ -1150,9 +1177,12 @@
 
 
 process::http::Response Master::ReadOnlyHandler::tasks(
-  const hashmap<std::string, std::string>& query,
-  const process::Owned<ObjectApprovers>& approvers) const
+    ContentType outputContentType,
+    const hashmap<std::string, std::string>& query,
+    const process::Owned<ObjectApprovers>& approvers) const
 {
+  CHECK_EQ(outputContentType, ContentType::JSON);
+
   // Get list options (limit and offset).
   Result<int> result = numify<int>(query.get("limit"));
   size_t limit = result.isSome() ? result.get() : TASK_LIMIT;
@@ -1257,6 +1287,1182 @@
   return OK(jsonify(tasksWriter), query.get("jsonp"));
 }
 
+
+function<void(JSON::ObjectWriter*)> Master::ReadOnlyHandler::jsonifyGetAgents(
+    const Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following:
+  //
+  //   mesos::master::Response::GetAgents getAgents;
+  //   for each registered agent:
+  //     *getAgents.add_agents() = protobuf::master::event::createAgentResponse(
+  //         agent,
+  //         master->slaves.draining.get(slave->id),
+  //         master->slaves.deactivated.contains(slave->id),
+  //         approvers);
+  //   for each recovered agent:
+  //     SlaveInfo* agent = getAgents.add_recovered_agents();
+  //     agent->CopyFrom(slaveInfo);
+  //     agent->clear_resources();
+  //     foreach (const Resource& resource, slaveInfo.resources()):
+  //       if (approvers->approved<VIEW_ROLE>(resource)):
+  //         *agent->add_resources() = resource;
+
+  // TODO(bmahler): This copies the Owned object approvers.
+  return [=](JSON::ObjectWriter* writer) {
+    const google::protobuf::Descriptor* descriptor =
+      v1::master::Response::GetAgents::descriptor();
+
+    int field;
+
+    field = v1::master::Response::GetAgents::kAgentsFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+      foreachvalue (const Slave* slave, master->slaves.registered) {
+        // TODO(bmahler): Consider not constructing the temporary
+        // agent object and instead serialize directly.
+        mesos::master::Response::GetAgents::Agent agent =
+            protobuf::master::event::createAgentResponse(
+                *slave,
+                master->slaves.draining.get(slave->id),
+                master->slaves.deactivated.contains(slave->id),
+                approvers);
+
+        writer->element(asV1Protobuf(agent));
+      }
+    });
+
+    field = v1::master::Response::GetAgents::kRecoveredAgentsFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+      foreachvalue (const SlaveInfo& slaveInfo, master->slaves.recovered) {
+        // TODO(bmahler): Consider not constructing the temporary
+        // SlaveInfo object and instead serialize directly.
+        SlaveInfo agent = slaveInfo;
+        agent.clear_resources();
+        foreach (const Resource& resource, slaveInfo.resources()) {
+          if (approvers->approved<VIEW_ROLE>(resource)) {
+            *agent.add_resources() = resource;
+          }
+        }
+
+        writer->element(asV1Protobuf(agent));
+      }
+    });
+  };
+}
+
+
+string Master::ReadOnlyHandler::serializeGetAgents(
+    const Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following:
+  //
+  //   mesos::master::Response::GetAgents getAgents;
+  //   for each registered agent:
+  //     *getAgents.add_agents() = protobuf::master::event::createAgentResponse(
+  //         agent,
+  //         master->slaves.draining.get(slave->id),
+  //         master->slaves.deactivated.contains(slave->id),
+  //         approvers);
+  //   for each recovered agent:
+  //     SlaveInfo* agent = getAgents.add_recovered_agents();
+  //     agent->CopyFrom(slaveInfo);
+  //     agent->clear_resources();
+  //     foreach (const Resource& resource, slaveInfo.resources()):
+  //       if (approvers->approved<VIEW_ROLE>(resource)):
+  //         *agent->add_resources() = resource;
+
+  string output;
+  google::protobuf::io::StringOutputStream stream(&output);
+  google::protobuf::io::CodedOutputStream writer(&stream);
+
+  foreachvalue (const Slave* slave, master->slaves.registered) {
+    // TODO(bmahler): Consider not constructing the temporary
+    // agent object and instead serialize directly.
+    WireFormatLite2::WriteMessageWithoutCachedSizes(
+        mesos::master::Response::GetAgents::kAgentsFieldNumber,
+        protobuf::master::event::createAgentResponse(
+            *slave,
+            master->slaves.draining.get(slave->id),
+            master->slaves.deactivated.contains(slave->id),
+            approvers),
+        &writer);
+  }
+
+  foreachvalue (const SlaveInfo& slaveInfo, master->slaves.recovered) {
+    // TODO(bmahler): Consider not constructing the temporary
+    // SlaveInfo object and instead serialize directly.
+    SlaveInfo agent = slaveInfo;
+    agent.clear_resources();
+    foreach (const Resource& resource, slaveInfo.resources()) {
+      if (approvers->approved<VIEW_ROLE>(resource)) {
+        *agent.add_resources() = resource;
+      }
+    }
+
+    WireFormatLite2::WriteMessageWithoutCachedSizes(
+        mesos::master::Response::GetAgents::kRecoveredAgentsFieldNumber,
+        agent,
+        &writer);
+  }
+
+  // While an explicit Trim() isn't necessary (since the coded
+  // output stream is destructed before the string is returned),
+  // it's a quite tricky bug to diagnose if Trim() is missed, so
+  // we always do it explicitly to signal the reader about this
+  // subtlety.
+  writer.Trim();
+
+  return output;
+}
+
+
+
+process::http::Response Master::ReadOnlyHandler::getAgents(
+    ContentType outputContentType,
+    const hashmap<std::string, std::string>& query,
+    const process::Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following message:
+  //
+  //   mesos::master::Response response;
+  //   response.set_type(mesos::master::Response::GET_AGENTS);
+  //   *response.mutable_get_agents() = _getAgents(approvers);
+
+  switch (outputContentType) {
+    case ContentType::PROTOBUF: {
+      string output;
+      google::protobuf::io::StringOutputStream stream(&output);
+      google::protobuf::io::CodedOutputStream writer(&stream);
+
+      WireFormatLite::WriteEnum(
+          mesos::v1::master::Response::kTypeFieldNumber,
+          mesos::v1::master::Response::GET_AGENTS,
+          &writer);
+
+      WireFormatLite::WriteBytes(
+          mesos::v1::master::Response::kGetAgentsFieldNumber,
+          serializeGetAgents(approvers),
+          &writer);
+
+      // We must manually trim the unused buffer space since
+      // we use the string before the coded output stream is
+      // destructed.
+      writer.Trim();
+
+      return OK(std::move(output), stringify(outputContentType));
+    }
+
+    case ContentType::JSON: {
+      string body = jsonify([&](JSON::ObjectWriter* writer) {
+        const google::protobuf::Descriptor* descriptor =
+          v1::master::Response::descriptor();
+
+        int field;
+
+        field = v1::master::Response::kTypeFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            v1::master::Response::Type_Name(
+                v1::master::Response::GET_AGENTS));
+
+        field = v1::master::Response::kGetAgentsFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            jsonifyGetAgents(approvers));
+      });
+
+      // TODO(bmahler): Pass jsonp query parameter through here.
+      return OK(std::move(body), stringify(outputContentType));
+    }
+
+    default:
+      return NotAcceptable("Request must accept json or protobuf");
+  }
+}
+
+
+function<void(JSON::ObjectWriter*)>
+  Master::ReadOnlyHandler::jsonifyGetFrameworks(
+      const Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following:
+  //
+  //   mesos::master::Response::GetFrameworks getFrameworks;
+  //   for each framework:
+  //     *getFrameworks.add_frameworks() = model(*framework);
+  //   for each completed framework:
+  //     *getFrameworks.add_completed_frameworks() = model(*framework);
+
+  // TODO(bmahler): Consider not constructing the temporary framework
+  // objects and instead serialize directly, but since we don't
+  // expect a large number of pending tasks, we currently don't
+  // bother with the more efficient approach.
+
+  // TODO(bmahler): This copies the Owned object approvers.
+  return [=](JSON::ObjectWriter* writer) {
+    const google::protobuf::Descriptor* descriptor =
+      v1::master::Response::GetFrameworks::descriptor();
+
+    int field;
+
+    field = v1::master::Response::GetFrameworks::kFrameworksFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+      foreachvalue (const Framework* framework,
+                    master->frameworks.registered) {
+        // Skip unauthorized frameworks.
+        if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+          continue;
+        }
+
+        mesos::master::Response::GetFrameworks::Framework f = model(*framework);
+        writer->element(asV1Protobuf(f));
+      }
+    });
+
+    field =
+      v1::master::Response::GetFrameworks::kCompletedFrameworksFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+      foreachvalue (const Owned<Framework>& framework,
+                    master->frameworks.completed) {
+        // Skip unauthorized frameworks.
+        if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+          continue;
+        }
+
+        mesos::master::Response::GetFrameworks::Framework f = model(*framework);
+        writer->element(asV1Protobuf(f));
+      }
+    });
+  };
+}
+
+
+string Master::ReadOnlyHandler::serializeGetFrameworks(
+    const Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following:
+  //
+  //   mesos::master::Response::GetFrameworks getFrameworks;
+  //   for each framework:
+  //     *getFrameworks.add_frameworks() = model(*framework);
+  //   for each completed framework:
+  //     *getFrameworks.add_completed_frameworks() = model(*framework);
+
+  string output;
+  google::protobuf::io::StringOutputStream stream(&output);
+  google::protobuf::io::CodedOutputStream writer(&stream);
+
+  // TODO(bmahler): Consider not constructing the temporary framework
+  // objects and instead serialize directly, but since we don't
+  // expect a large number of pending tasks, we currently don't
+  // bother with the more efficient approach.
+
+  foreachvalue (const Framework* framework,
+                master->frameworks.registered) {
+    // Skip unauthorized frameworks.
+    if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+      continue;
+    }
+
+    WireFormatLite2::WriteMessageWithoutCachedSizes(
+        mesos::master::Response::GetFrameworks::kFrameworksFieldNumber,
+        model(*framework),
+        &writer);
+  }
+
+  foreachvalue (const Owned<Framework>& framework,
+                master->frameworks.completed) {
+    // Skip unauthorized frameworks.
+    if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+      continue;
+    }
+
+    WireFormatLite2::WriteMessageWithoutCachedSizes(
+        mesos::master::Response::GetFrameworks::kCompletedFrameworksFieldNumber,
+        model(*framework),
+        &writer);
+  }
+
+  // While an explicit Trim() isn't necessary (since the coded
+  // output stream is destructed before the string is returned),
+  // it's a quite tricky bug to diagnose if Trim() is missed, so
+  // we always do it explicitly to signal the reader about this
+  // subtlety.
+  writer.Trim();
+
+  return output;
+}
+
+
+process::http::Response Master::ReadOnlyHandler::getFrameworks(
+    ContentType outputContentType,
+    const hashmap<std::string, std::string>& query,
+    const process::Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following message:
+  //
+  //   mesos::master::Response response;
+  //   response.set_type(mesos::master::Response::GET_FRAMEWORKS);
+  //   *response.mutable_get_frameworks() = _getFrameworks(approvers);
+
+  switch (outputContentType) {
+    case ContentType::PROTOBUF: {
+      string output;
+      google::protobuf::io::StringOutputStream stream(&output);
+      google::protobuf::io::CodedOutputStream writer(&stream);
+
+      WireFormatLite::WriteEnum(
+          mesos::v1::master::Response::kTypeFieldNumber,
+          mesos::v1::master::Response::GET_FRAMEWORKS,
+          &writer);
+
+      WireFormatLite::WriteBytes(
+          mesos::v1::master::Response::kGetFrameworksFieldNumber,
+          serializeGetFrameworks(approvers),
+          &writer);
+
+      // We must manually trim the unused buffer space since
+      // we use the string before the coded output stream is
+      // destructed.
+      writer.Trim();
+
+      return OK(std::move(output), stringify(outputContentType));
+    }
+
+    case ContentType::JSON: {
+      string body = jsonify([&](JSON::ObjectWriter* writer) {
+        const google::protobuf::Descriptor* descriptor =
+          v1::master::Response::descriptor();
+
+        int field;
+
+        field = v1::master::Response::kTypeFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            v1::master::Response::Type_Name(
+                v1::master::Response::GET_FRAMEWORKS));
+
+        field = v1::master::Response::kGetFrameworksFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            jsonifyGetFrameworks(approvers));
+      });
+
+      // TODO(bmahler): Pass jsonp query parameter through here.
+      return OK(std::move(body), stringify(outputContentType));
+    }
+
+    default:
+      return NotAcceptable("Request must accept json or protobuf");
+  }
+}
+
+
+function<void(JSON::ObjectWriter*)>
+  Master::ReadOnlyHandler::jsonifyGetExecutors(
+      const Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following:
+  //
+  //   mesos::master::Response::GetExecutors getExecutors;
+  //
+  //   for each (executor, agent):
+  //     mesos::master::Response::GetExecutors::Executor* executor =
+  //       getExecutors.add_executors();
+  //     *executor->mutable_executor_info() = executorInfo;
+  //     *executor->mutable_slave_id() = slaveId;
+
+  // TODO(bmahler): This copies the owned object approvers.
+  return [=](JSON::ObjectWriter* writer) {
+    // Construct framework list with both active and completed frameworks.
+    vector<const Framework*> frameworks;
+    foreachvalue (Framework* framework, master->frameworks.registered) {
+      // Skip unauthorized frameworks.
+      if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+        frameworks.push_back(framework);
+      }
+    }
+    foreachvalue (const Owned<Framework>& framework,
+                  master->frameworks.completed) {
+      // Skip unauthorized frameworks.
+      if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+        frameworks.push_back(framework.get());
+      }
+    }
+
+    const google::protobuf::Descriptor* descriptor =
+      v1::master::Response::GetExecutors::descriptor();
+
+    int field;
+
+    field = v1::master::Response::GetExecutors::kExecutorsFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+      foreach (const Framework* framework, frameworks) {
+        foreachpair (const SlaveID& slaveId,
+                     const auto& executorsMap,
+                     framework->executors) {
+          foreachvalue (const ExecutorInfo& executorInfo, executorsMap) {
+            // Skip unauthorized executors.
+            if (!approvers->approved<VIEW_EXECUTOR>(
+                    executorInfo, framework->info)) {
+              continue;
+            }
+
+            writer->element([&](JSON::ObjectWriter* writer) {
+              const google::protobuf::Descriptor* descriptor =
+                v1::master::Response::GetExecutors::Executor::descriptor();
+
+              // Serialize the following message:
+              //
+              //   mesos::master::Response::GetExecutors::Executor executor;
+              //   *executor.mutable_executor_info() = executorInfo;
+              //   *executor.mutable_slave_id() = slaveId;
+              int field;
+
+              field = v1::master::Response::GetExecutors::Executor
+                ::kExecutorInfoFieldNumber;
+              writer->field(
+                  descriptor->FindFieldByNumber(field)->name(),
+                  asV1Protobuf(executorInfo));
+
+              field = v1::master::Response::GetExecutors::Executor
+                ::kAgentIdFieldNumber;
+              writer->field(
+                  descriptor->FindFieldByNumber(field)->name(),
+                  asV1Protobuf(slaveId));
+            });
+          }
+        }
+      }
+    });
+  };
+}
+
+
+string Master::ReadOnlyHandler::serializeGetExecutors(
+    const Owned<ObjectApprovers>& approvers) const
+{
+  // Construct framework list with both active and completed frameworks.
+  vector<const Framework*> frameworks;
+  foreachvalue (Framework* framework, master->frameworks.registered) {
+    // Skip unauthorized frameworks.
+    if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+      frameworks.push_back(framework);
+    }
+  }
+  foreachvalue (const Owned<Framework>& framework,
+                master->frameworks.completed) {
+    // Skip unauthorized frameworks.
+    if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+      frameworks.push_back(framework.get());
+    }
+  }
+
+  // Lambda for serializing the following message:
+  //
+  //   mesos::master::Response::GetExecutors::Executor executor;
+  //   *executor.mutable_executor_info() = executorInfo;
+  //   *executor.mutable_slave_id() = slaveId;
+  auto serializeExecutor = [](const ExecutorInfo& e, const SlaveID& s) {
+    string output;
+    google::protobuf::io::StringOutputStream stream(&output);
+    google::protobuf::io::CodedOutputStream writer(&stream);
+
+    WireFormatLite2::WriteMessageWithoutCachedSizes(
+        mesos::v1::master::Response::GetExecutors::Executor
+          ::kExecutorInfoFieldNumber,
+        e,
+        &writer);
+
+    WireFormatLite2::WriteMessageWithoutCachedSizes(
+        mesos::v1::master::Response::GetExecutors::Executor
+          ::kAgentIdFieldNumber,
+        s,
+        &writer);
+
+    // While an explicit Trim() isn't necessary (since the coded
+    // output stream is destructed before the string is returned),
+    // it's a quite tricky bug to diagnose if Trim() is missed, so
+    // we always do it explicitly to signal the reader about this
+    // subtlety.
+    writer.Trim();
+
+    return output;
+  };
+
+  string output;
+  google::protobuf::io::StringOutputStream stream(&output);
+  google::protobuf::io::CodedOutputStream writer(&stream);
+
+  // Serialize the following:
+  //
+  //   mesos::master::Response::GetExecutors getExecutors;
+  //
+  //   for each (executor, agent):
+  //     mesos::master::Response::GetExecutors::Executor* executor =
+  //       getExecutors.add_executors();
+  //     *executor->mutable_executor_info() = executorInfo;
+  //     *executor->mutable_slave_id() = slaveId;
+
+  foreach (const Framework* framework, frameworks) {
+    foreachpair (const SlaveID& slaveId,
+                 const auto& executorsMap,
+                 framework->executors) {
+      foreachvalue (const ExecutorInfo& executorInfo, executorsMap) {
+        // Skip unauthorized executors.
+        if (!approvers->approved<VIEW_EXECUTOR>(
+                executorInfo, framework->info)) {
+          continue;
+        }
+
+        WireFormatLite::WriteBytes(
+            mesos::v1::master::Response::GetExecutors::kExecutorsFieldNumber,
+            serializeExecutor(executorInfo, slaveId),
+            &writer);
+      }
+    }
+  }
+
+  // While an explicit Trim() isn't necessary (since the coded
+  // output stream is destructed before the string is returned),
+  // it's a quite tricky bug to diagnose if Trim() is missed, so
+  // we always do it explicitly to signal the reader about this
+  // subtlety.
+  writer.Trim();
+
+  return output;
+}
+
+
+process::http::Response Master::ReadOnlyHandler::getExecutors(
+    ContentType outputContentType,
+    const hashmap<std::string, std::string>& query,
+    const process::Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following message:
+  //
+  //   mesos::master::Response response;
+  //   response.set_type(mesos::master::Response::GET_EXECUTORS);
+  //   *response.mutable_get_executors() = _getExecutors(approvers);
+
+  switch (outputContentType) {
+    case ContentType::PROTOBUF: {
+      string output;
+      google::protobuf::io::StringOutputStream stream(&output);
+      google::protobuf::io::CodedOutputStream writer(&stream);
+
+      WireFormatLite::WriteEnum(
+          mesos::v1::master::Response::kTypeFieldNumber,
+          mesos::v1::master::Response::GET_EXECUTORS,
+          &writer);
+
+      WireFormatLite::WriteBytes(
+          mesos::v1::master::Response::kGetExecutorsFieldNumber,
+          serializeGetExecutors(approvers),
+          &writer);
+
+      // We must manually trim the unused buffer space since
+      // we use the string before the coded output stream is
+      // destructed.
+      writer.Trim();
+
+      return OK(std::move(output), stringify(outputContentType));
+    }
+
+    case ContentType::JSON: {
+      string body = jsonify([&](JSON::ObjectWriter* writer) {
+        const google::protobuf::Descriptor* descriptor =
+          v1::master::Response::descriptor();
+
+        int field;
+
+        field = v1::master::Response::kTypeFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            v1::master::Response::Type_Name(
+                v1::master::Response::GET_EXECUTORS));
+
+        field = v1::master::Response::kGetExecutorsFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            jsonifyGetExecutors(approvers));
+      });
+
+      // TODO(bmahler): Pass jsonp query parameter through here.
+      return OK(std::move(body), stringify(outputContentType));
+    }
+
+    default:
+      return NotAcceptable("Request must accept json or protobuf");
+  }
+}
+
+
+function<void(JSON::ObjectWriter*)> Master::ReadOnlyHandler::jsonifyGetTasks(
+    const Owned<ObjectApprovers>& approvers) const
+{
+  // Jsonify the following message:
+  //
+  //  master::Response::GetTasks getTasks;
+  //  for each pending task:
+  //    *getTasks.add_pending_tasks() =
+  //      protobuf::createTask(taskInfo, TASK_STAGING, framework->id());
+  //  for each task:
+  //    *getTasks.add_tasks() = *task;
+  //  for each unreachable task:
+  //    *getTasks.add_unreachable_tasks() = *task;
+  //  for each completed task:
+  //    *getTasks.add_completed_tasks() = *task;
+
+  // TODO(bmahler): This copies the Owned object approvers.
+  return [=](JSON::ObjectWriter* writer) {
+    // Construct framework list with both active and completed frameworks.
+    vector<const Framework*> frameworks;
+    foreachvalue (Framework* framework, master->frameworks.registered) {
+      // Skip unauthorized frameworks.
+      if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+        frameworks.push_back(framework);
+      }
+    }
+    foreachvalue (const Owned<Framework>& framework,
+                  master->frameworks.completed) {
+      // Skip unauthorized frameworks.
+      if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+        frameworks.push_back(framework.get());
+      }
+    }
+
+    const google::protobuf::Descriptor* descriptor =
+      v1::master::Response::GetTasks::descriptor();
+
+    int field;
+
+    // Pending tasks.
+    field = v1::master::Response::GetTasks::kPendingTasksFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+          foreach (const Framework* framework, frameworks) {
+            foreachvalue (const TaskInfo& t, framework->pendingTasks) {
+              // Skip unauthorized tasks.
+              if (!approvers->approved<VIEW_TASK>(t, framework->info)) {
+                continue;
+              }
+
+              Task task =
+                  protobuf::createTask(t, TASK_STAGING, framework->id());
+
+              writer->element(asV1Protobuf(task));
+            }
+          }
+        });
+
+    // Active tasks.
+    field = v1::master::Response::GetTasks::kTasksFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+          foreach (const Framework* framework, frameworks) {
+            foreachvalue (Task* task, framework->tasks) {
+              CHECK_NOTNULL(task);
+              // Skip unauthorized tasks.
+              if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
+                continue;
+              }
+
+              writer->element(asV1Protobuf(*task));
+            }
+          }
+        });
+
+    // Unreachable tasks.
+    field = v1::master::Response::GetTasks::kUnreachableTasksFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+          foreach (const Framework* framework, frameworks) {
+            foreachvalue (const Owned<Task>& task,
+                          framework->unreachableTasks) {
+              // Skip unauthorized tasks.
+              if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
+                continue;
+              }
+
+              writer->element(asV1Protobuf(*task));
+            }
+          }
+        });
+
+    // Completed tasks.
+    field = v1::master::Response::GetTasks::kCompletedTasksFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        [&](JSON::ArrayWriter* writer) {
+          foreach (const Framework* framework, frameworks) {
+            foreach (const Owned<Task>& task, framework->completedTasks) {
+              // Skip unauthorized tasks.
+              if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
+                continue;
+              }
+
+              writer->element(asV1Protobuf(*task));
+            }
+          }
+        });
+  };
+}
+
+
+string Master::ReadOnlyHandler::serializeGetTasks(
+    const Owned<ObjectApprovers>& approvers) const
+{
+  // Construct framework list with both active and completed frameworks.
+  vector<const Framework*> frameworks;
+  foreachvalue (Framework* framework, master->frameworks.registered) {
+    // Skip unauthorized frameworks.
+    if (approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+      frameworks.push_back(framework);
+    }
+  }
+  foreachvalue (const Owned<Framework>& framework,
+                master->frameworks.completed) {
+    // Skip unauthorized frameworks.
+    if (!approvers->approved<VIEW_FRAMEWORK>(framework->info)) {
+      frameworks.push_back(framework.get());
+    }
+  }
+
+  // Serialize the following message:
+  //
+  //  mesos::master::Response::GetTasks getTasks;
+  //  for each pending task:
+  //    *getTasks.add_pending_tasks() =
+  //      protobuf::createTask(taskInfo, TASK_STAGING, framework->id());
+  //  for each task:
+  //    *getTasks.add_tasks() = *task;
+  //  for each unreachable task:
+  //    *getTasks.add_unreachable_tasks() = *task;
+  //  for each completed task:
+  //    *getTasks.add_completed_tasks() = *task;
+
+  string output;
+  google::protobuf::io::StringOutputStream stream(&output);
+  google::protobuf::io::CodedOutputStream writer(&stream);
+
+  foreach (const Framework* framework, frameworks) {
+    // Pending tasks.
+    foreachvalue (const TaskInfo& taskInfo, framework->pendingTasks) {
+      // Skip unauthorized tasks.
+      if (!approvers->approved<VIEW_TASK>(taskInfo, framework->info)) {
+        continue;
+      }
+
+      // TODO(bmahler): Consider not constructing the temporary task
+      // object and instead serialize directly. Since we don't expect
+      // a large number of pending tasks, we currently don't bother
+      // with the more efficient approach.
+      //
+      // *getTasks.add_pending_tasks() =
+      //   protobuf::createTask(taskInfo, TASK_STAGING, framework->id());
+      WireFormatLite2::WriteMessageWithoutCachedSizes(
+          mesos::v1::master::Response::GetTasks::kPendingTasksFieldNumber,
+          protobuf::createTask(taskInfo, TASK_STAGING, framework->id()),
+          &writer);
+    }
+
+    // Active tasks.
+    foreachvalue (Task* task, framework->tasks) {
+      CHECK_NOTNULL(task);
+      // Skip unauthorized tasks.
+      if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
+        continue;
+      }
+
+      WireFormatLite2::WriteMessageWithoutCachedSizes(
+          mesos::v1::master::Response::GetTasks::kTasksFieldNumber,
+          *task,
+          &writer);
+    }
+
+    // Unreachable tasks.
+    foreachvalue (const Owned<Task>& task, framework->unreachableTasks) {
+      // Skip unauthorized tasks.
+      if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
+        continue;
+      }
+
+      WireFormatLite2::WriteMessageWithoutCachedSizes(
+          mesos::v1::master::Response::GetTasks::kUnreachableTasksFieldNumber,
+          *task,
+          &writer);
+    }
+
+    // Completed tasks.
+    foreach (const Owned<Task>& task, framework->completedTasks) {
+      // Skip unauthorized tasks.
+      if (!approvers->approved<VIEW_TASK>(*task, framework->info)) {
+        continue;
+      }
+
+      WireFormatLite2::WriteMessageWithoutCachedSizes(
+          mesos::v1::master::Response::GetTasks::kCompletedTasksFieldNumber,
+          *task,
+          &writer);
+    }
+  }
+
+  // While an explicit Trim() isn't necessary (since the coded
+  // output stream is destructed before the string is returned),
+  // it's a quite tricky bug to diagnose if Trim() is missed, so
+  // we always do it explicitly to signal the reader about this
+  // subtlety.
+  writer.Trim();
+
+  return output;
+}
+
+
+process::http::Response Master::ReadOnlyHandler::getTasks(
+    ContentType outputContentType,
+    const hashmap<std::string, std::string>& query,
+    const process::Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following message:
+  //
+  //    mesos::master::Response response;
+  //    response.set_type(mesos::master::Response::GET_TASKS);
+  //    *response.mutable_get_tasks() = _getTasks(approvers);
+
+  switch (outputContentType) {
+    case ContentType::PROTOBUF: {
+      string output;
+      google::protobuf::io::StringOutputStream stream(&output);
+      google::protobuf::io::CodedOutputStream writer(&stream);
+
+      WireFormatLite::WriteEnum(
+          mesos::v1::master::Response::kTypeFieldNumber,
+          mesos::v1::master::Response::GET_TASKS,
+          &writer);
+
+      WireFormatLite::WriteBytes(
+          mesos::v1::master::Response::kGetTasksFieldNumber,
+          serializeGetTasks(approvers),
+          &writer);
+
+      // We must manually trim the unused buffer space since
+      // we use the string before the coded output stream is
+      // destructed.
+      writer.Trim();
+
+      return OK(std::move(output), stringify(outputContentType));
+    }
+
+    case ContentType::JSON: {
+      string body = jsonify([&](JSON::ObjectWriter* writer) {
+        const google::protobuf::Descriptor* descriptor =
+          v1::master::Response::descriptor();
+
+        int field;
+
+        field = v1::master::Response::kTypeFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            v1::master::Response::Type_Name(
+                v1::master::Response::GET_TASKS));
+
+        field = v1::master::Response::kGetTasksFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            jsonifyGetTasks(approvers));
+      });
+
+      // TODO(bmahler): Pass jsonp query parameter through here.
+      return OK(std::move(body), stringify(outputContentType));
+    }
+
+    default:
+      return NotAcceptable("Request must accept json or protobuf");
+  }
+}
+
+
+process::http::Response Master::ReadOnlyHandler::getOperations(
+    ContentType outputContentType,
+    const hashmap<std::string, std::string>& query,
+    const process::Owned<ObjectApprovers>& approvers) const
+{
+  // We consider a principal to be authorized to view an operation if it
+  // is authorized to view the resources the operation is performed on.
+  auto approved = [&approvers](const Operation& operation) {
+    Try<Resources> consumedResources =
+      protobuf::getConsumedResources(operation.info());
+
+    if (consumedResources.isError()) {
+      LOG(WARNING)
+        << "Could not approve operation " << operation.uuid()
+        << " since its consumed resources could not be determined:"
+        << consumedResources.error();
+
+      return false;
+    }
+
+    foreach (const Resource& resource, consumedResources.get()) {
+      if (!approvers->approved<VIEW_ROLE>(resource)) {
+        return false;
+      }
+    }
+
+    return true;
+  };
+
+  mesos::master::Response response;
+  response.set_type(mesos::master::Response::GET_OPERATIONS);
+
+  mesos::master::Response::GetOperations* operations =
+    response.mutable_get_operations();
+
+  foreachvalue (const Slave* slave, master->slaves.registered) {
+    foreachvalue (Operation* operation, slave->operations) {
+      if (approved(*operation)) {
+        operations->add_operations()->CopyFrom(*operation);
+      }
+    }
+
+    foreachvalue (
+        const Slave::ResourceProvider resourceProvider,
+        slave->resourceProviders) {
+      foreachvalue (Operation* operation, resourceProvider.operations) {
+        if (approved(*operation)) {
+          operations->add_operations()->CopyFrom(*operation);
+        }
+      }
+    }
+  }
+
+  return OK(
+      serialize(outputContentType, evolve(response)),
+      stringify(outputContentType));
+}
+
+
+process::http::Response Master::ReadOnlyHandler::getRoles(
+    ContentType outputContentType,
+    const hashmap<std::string, std::string>& query,
+    const process::Owned<ObjectApprovers>& approvers) const
+{
+  const vector<string> knownRoles = master->knownRoles();
+
+  mesos::master::Response response;
+  response.set_type(mesos::master::Response::GET_ROLES);
+
+  mesos::master::Response::GetRoles* getRoles =
+    response.mutable_get_roles();
+
+  foreach (const string& name, knownRoles) {
+    if (!approvers->approved<VIEW_ROLE>(name)) {
+      continue;
+    }
+
+    mesos::Role* role = getRoles->add_roles();
+
+    role->set_name(name);
+
+    role->set_weight(master->weights.get(name).getOrElse(DEFAULT_WEIGHT));
+
+    RoleResourceBreakdown resourceBreakdown(master, name);
+
+    ResourceQuantities allocatedAndOffered =
+      resourceBreakdown.allocated() + resourceBreakdown.offered();
+
+    // `resources` will be deprecated in favor of
+    // `offered`, `allocated`, `reserved`, and quota consumption.
+    // As a result, we don't bother trying to expose more
+    // than {cpus, mem, disk, gpus} since we don't know if
+    // anything outside this set is of type SCALAR.
+    foreach (const auto& quantity, allocatedAndOffered) {
+      if (quantity.first == "cpus" || quantity.first == "mem" ||
+          quantity.first == "disk" || quantity.first == "gpus") {
+        Resource* resource = role->add_resources();
+        resource->set_name(quantity.first);
+        resource->set_type(Value::SCALAR);
+        *resource->mutable_scalar() = quantity.second;
+      }
+    }
+
+    Option<Role*> role_ = master->roles.get(name);
+
+    if (role_.isSome()) {
+      foreachkey (const FrameworkID& frameworkId, (*role_)->frameworks) {
+        *role->add_frameworks() = frameworkId;
+      }
+    }
+  }
+
+  return OK(serialize(outputContentType, evolve(response)),
+            stringify(outputContentType));
+}
+
+
+function<void(JSON::ObjectWriter*)> Master::ReadOnlyHandler::jsonifyGetState(
+    const Owned<ObjectApprovers>& approvers) const
+{
+  // Jsonify the following message:
+  //
+  //   mesos::master::Response::GetState getState;
+  //   *getState.mutable_get_tasks() = _getTasks(approvers);
+  //   *getState.mutable_get_executors() = _getExecutors(approvers);
+  //   *getState.mutable_get_frameworks() = _getFrameworks(approvers);
+  //   *getState.mutable_get_agents() = _getAgents(approvers);
+
+  // TODO(bmahler): This copies the Owned object approvers.
+  return [=](JSON::ObjectWriter* writer) {
+    const google::protobuf::Descriptor* descriptor =
+      v1::master::Response::GetState::descriptor();
+
+    int field;
+
+    field = v1::master::Response::GetState::kGetTasksFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        jsonifyGetTasks(approvers));
+
+    field = v1::master::Response::GetState::kGetExecutorsFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        jsonifyGetExecutors(approvers));
+
+    field = v1::master::Response::GetState::kGetFrameworksFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        jsonifyGetFrameworks(approvers));
+
+    field = v1::master::Response::GetState::kGetAgentsFieldNumber;
+    writer->field(
+        descriptor->FindFieldByNumber(field)->name(),
+        jsonifyGetAgents(approvers));
+  };
+}
+
+
+string Master::ReadOnlyHandler::serializeGetState(
+    const Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following message:
+  //
+  //   mesos::master::Response::GetState getState;
+  //   *getState.mutable_get_tasks() = _getTasks(approvers);
+  //   *getState.mutable_get_executors() = _getExecutors(approvers);
+  //   *getState.mutable_get_frameworks() = _getFrameworks(approvers);
+  //   *getState.mutable_get_agents() = _getAgents(approvers);
+
+  string output;
+  google::protobuf::io::StringOutputStream stream(&output);
+  google::protobuf::io::CodedOutputStream writer(&stream);
+
+  WireFormatLite::WriteBytes(
+      mesos::v1::master::Response::GetState::kGetTasksFieldNumber,
+      serializeGetTasks(approvers),
+      &writer);
+
+  WireFormatLite::WriteBytes(
+      mesos::v1::master::Response::GetState::kGetExecutorsFieldNumber,
+      serializeGetExecutors(approvers),
+      &writer);
+
+  WireFormatLite::WriteBytes(
+      mesos::v1::master::Response::GetState::kGetFrameworksFieldNumber,
+      serializeGetFrameworks(approvers),
+      &writer);
+
+  WireFormatLite::WriteBytes(
+      mesos::v1::master::Response::GetState::kGetAgentsFieldNumber,
+      serializeGetAgents(approvers),
+      &writer);
+
+  // While an explicit Trim() isn't necessary (since the coded
+  // output stream is destructed before the string is returned),
+  // it's a quite tricky bug to diagnose if Trim() is missed, so
+  // we always do it explicitly to signal the reader about this
+  // subtlety.
+  writer.Trim();
+
+  return output;
+}
+
+
+process::http::Response Master::ReadOnlyHandler::getState(
+    ContentType outputContentType,
+    const hashmap<std::string, std::string>& query,
+    const process::Owned<ObjectApprovers>& approvers) const
+{
+  // Serialize the following message:
+  //
+  //   mesos::master::Response response;
+  //   response.set_type(mesos::master::Response::GET_STATE);
+  //   *response.mutable_get_state() = _getState(approvers);
+
+  switch (outputContentType) {
+    case ContentType::PROTOBUF: {
+      string output;
+      google::protobuf::io::StringOutputStream stream(&output);
+      google::protobuf::io::CodedOutputStream writer(&stream);
+
+      WireFormatLite::WriteEnum(
+          mesos::v1::master::Response::kTypeFieldNumber,
+          mesos::v1::master::Response::GET_STATE,
+          &writer);
+
+      WireFormatLite::WriteBytes(
+          mesos::v1::master::Response::kGetStateFieldNumber,
+          serializeGetState(approvers),
+          &writer);
+
+      // We must manually trim the unused buffer space since
+      // we use the string before the coded output stream is
+      // destructed.
+      writer.Trim();
+
+      return OK(std::move(output), stringify(outputContentType));
+    }
+
+    case ContentType::JSON: {
+      string body = jsonify([&](JSON::ObjectWriter* writer) {
+        const google::protobuf::Descriptor* descriptor =
+          v1::master::Response::descriptor();
+
+        int field;
+
+        field = v1::master::Response::kTypeFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            v1::master::Response::Type_Name(
+                v1::master::Response::GET_STATE));
+
+        field = v1::master::Response::kGetStateFieldNumber;
+        writer->field(
+            descriptor->FindFieldByNumber(field)->name(),
+            jsonifyGetState(approvers));
+      });
+
+      // TODO(bmahler): Pass jsonp query parameter through here.
+      return OK(std::move(body), stringify(outputContentType));
+    }
+
+    default:
+      return NotAcceptable("Request must accept json or protobuf");
+  }
+}
+
 } // namespace master {
 } // namespace internal {
 } // namespace mesos {
diff --git a/src/tests/master_load_tests.cpp b/src/tests/master_load_tests.cpp
index ac16fe7..6cee248 100644
--- a/src/tests/master_load_tests.cpp
+++ b/src/tests/master_load_tests.cpp
@@ -390,15 +390,20 @@
 
     Response reference;
     if (request.endpoint == "/state") {
-      reference = readOnlyHandler.state(queryParameters, approvers);
+      reference = readOnlyHandler.state(
+          ContentType::JSON, queryParameters, approvers);
     } else if (request.endpoint == "/state-summary") {
-      reference = readOnlyHandler.stateSummary(queryParameters, approvers);
+      reference = readOnlyHandler.stateSummary(
+          ContentType::JSON, queryParameters, approvers);
     } else if (request.endpoint == "/roles") {
-      reference = readOnlyHandler.roles(queryParameters, approvers);
+      reference = readOnlyHandler.roles(
+          ContentType::JSON, queryParameters, approvers);
     } else if (request.endpoint == "/frameworks") {
-      reference = readOnlyHandler.frameworks(queryParameters, approvers);
+      reference = readOnlyHandler.frameworks(
+          ContentType::JSON, queryParameters, approvers);
     } else if (request.endpoint == "/slaves") {
-      reference = readOnlyHandler.slaves(queryParameters, approvers);
+      reference = readOnlyHandler.slaves(
+          ContentType::JSON, queryParameters, approvers);
     } else {
       UNREACHABLE();
     }