diff --git a/brew_view/__init__.py b/brew_view/__init__.py
index 2de3c12c..17e5e5e0 100644
--- a/brew_view/__init__.py
+++ b/brew_view/__init__.py
@@ -249,6 +249,8 @@ def _setup_tornado_app():
         CommandAPI,
         CommandListAPI,
         ConfigHandler,
+        FileListAPI,
+        FileAPI,
         InstanceAPI,
         QueueAPI,
         QueueListAPI,
@@ -303,6 +305,8 @@ def _setup_tornado_app():
         (r"{0}api/v1/config/logging/?".format(prefix), LoggingConfigAPI),
         # Beta
         (r"{0}api/vbeta/events/?".format(prefix), EventPublisherAPI),
+        (r"{0}api/vbeta/files/?".format(prefix), FileListAPI),
+        (r"{0}api/vbeta/files/(\w+)/?".format(prefix), FileAPI),
         # Deprecated
         (r"{0}api/v1/admin/system/?".format(prefix), OldAdminAPI),
         (r"{0}api/v1/admin/queues/?".format(prefix), OldQueueListAPI),
diff --git a/brew_view/authorization.py b/brew_view/authorization.py
index 603b7f56..36905d4d 100644
--- a/brew_view/authorization.py
+++ b/brew_view/authorization.py
@@ -23,6 +23,10 @@ class Permissions(Enum):
     EVENT_READ = "bg-event-read"
     EVENT_UPDATE = "bg-event-update"
     EVENT_DELETE = "bg-event-delete"
+    FILE_CREATE = "bg-file-create"
+    FILE_READ = "bg-file-read"
+    FILE_UPDATE = "bg-file-update"
+    FILE_DELETE = "bg-file-delete"
     INSTANCE_CREATE = "bg-instance-create"
     INSTANCE_READ = "bg-instance-read"
     INSTANCE_UPDATE = "bg-instance-update"
diff --git a/brew_view/base_handler.py b/brew_view/base_handler.py
index 5fd35ae9..a0049afc 100644
--- a/brew_view/base_handler.py
+++ b/brew_view/base_handler.py
@@ -58,6 +58,13 @@ class BaseHandler(AuthMixin, RequestHandler):
         socket.timeout: {"status_code": 504, "message": "Backend request timed out"},
     }
 
+    def get_bool_header(self, key, default_value):
+        value = self.request.headers.get(key)
+        if value is None:
+            return default_value
+
+        return str(value).lower() in ["1", "true", "t", "yes", "y"]
+
     def get_refresh_id_from_cookie(self):
         token_id = self.get_secure_cookie(self.REFRESH_COOKIE_NAME)
         if token_id:
@@ -125,21 +132,27 @@ def prepare(self):
             content_type = content_type.split(";")
 
             self.request.mime_type = content_type[0]
-            if self.request.mime_type not in [
+            if self.request.mime_type in [
                 "application/json",
                 "application/x-www-form-urlencoded",
             ]:
+                # Attempt to parse out the charset and decode the body, default to utf-8
+                charset = "utf-8"
+                if len(content_type) > 1:
+                    search_result = self.charset_re.search(content_type[1])
+                    if search_result:
+                        charset = search_result.group(1)
+                self.request.charset = charset
+                self.request.decoded_body = self.request.body.decode(charset)
+            elif (
+                self.request.mime_type == "multipart/form-data"
+                and "files" in self.request.path
+            ):
+                # No need to decode the request body, we're just reading bytes
+                pass
+            else:
                 raise ModelValidationError("Unsupported or missing content-type header")
 
-            # Attempt to parse out the charset and decode the body, default to utf-8
-            charset = "utf-8"
-            if len(content_type) > 1:
-                search_result = self.charset_re.search(content_type[1])
-                if search_result:
-                    charset = search_result.group(1)
-            self.request.charset = charset
-            self.request.decoded_body = self.request.body.decode(charset)
-
     def on_finish(self):
         """Called after a handler completes processing"""
         # This is gross, but in some cases we have to do these in the handler
diff --git a/brew_view/controllers/__init__.py b/brew_view/controllers/__init__.py
index 6d056db7..55b7c615 100644
--- a/brew_view/controllers/__init__.py
+++ b/brew_view/controllers/__init__.py
@@ -24,3 +24,4 @@
 from brew_view.controllers.system_list_api import SystemListAPI
 from brew_view.controllers.token_api import TokenAPI, TokenListAPI
 from brew_view.controllers.users_api import UserAPI, UsersAPI
+from brew_view.controllers.file_api import FileListAPI, FileAPI
diff --git a/brew_view/controllers/file_api.py b/brew_view/controllers/file_api.py
new file mode 100644
index 00000000..194ee338
--- /dev/null
+++ b/brew_view/controllers/file_api.py
@@ -0,0 +1,125 @@
+import logging
+
+from tornado.gen import coroutine
+
+from brew_view import file_writers
+from brew_view import file_readers
+from bg_utils.mongo.parser import MongoParser
+from brew_view.authorization import authenticated, Permissions
+from brew_view.base_handler import BaseHandler
+from brewtils.models import Events
+
+
+class FileListAPI(BaseHandler):
+
+    parser = MongoParser()
+    logger = logging.getLogger(__name__)
+
+    @coroutine
+    @authenticated(permissions=[Permissions.FILE_CREATE])
+    def post(self):
+        """
+        ---
+        summary: Create a new file
+        description: |
+            Creates a new file based on the default driver. You can specify
+            a different driver with a header.
+        consumes:
+          - multipart/form-data
+        parameters:
+          - name: file
+            in: formData
+            type: file
+            description: The file to upload.
+        responses:
+          201:
+            description: A new file has been created
+          400:
+            $ref: '#/definitions/400Error'
+          50x:
+            $ref: '#/definitions/50xError'
+        tags:
+          - Beta
+        """
+        file_writer = self._get_file_writer()
+
+        response = {}
+        for key in self.request.files:
+            self.request.event.name = Events.FILE_CREATED.name
+            fileinfo = self.request.files[key][0]
+            filename = fileinfo["filename"]
+            body = fileinfo["body"]
+            file_id = file_writer.write(body, filename=filename)
+            response[key] = {
+                "id": file_id,
+                "storage_type": "gridfs",
+                "filename": filename,
+            }
+
+        self.set_status(201)
+        self.write(response)
+
+    def _get_file_writer(self):
+        storage_type = self.request.headers.get("X-BG-Storage-Type", "gridfs")
+        return file_writers.get(storage_type)
+
+
+class FileAPI(BaseHandler):
+
+    parser = MongoParser()
+    logger = logging.getLogger(__name__)
+
+    @authenticated(permissions=[Permissions.FILE_READ])
+    def get(self, file_id):
+        """
+        ---
+        summary: Retrieve a specific file, or its metadata
+        description: |
+            If your storage_type is anything other than gridfs, you probably
+            only want to use this endpoint to retrieve metadata, however if
+            your storage type is gridfs, then you want to retrieve the actual
+            file.
+        parameters:
+          - name: file_id
+            in: path
+            required: true
+            description: The ID of the file
+            type: string
+          - name: X-BG-file-meta-only
+            in: header
+            required: false
+            description: If true, will only return JSON metadata about file
+            type: string
+        produces:
+          - application/json
+          - application/octet-stream
+        responses:
+          200:
+            description: Will return JSON if X-BG-file-meta-only is true,
+              otherwise, it will return the actual file.
+          404:
+            $ref: '#/definitions/404Error'
+          50x:
+            $ref: '#/definitions/50xError'
+        tags:
+          - Beta
+        """
+        meta_only = self.get_bool_header("X-BG-file-meta-only", False)
+        request_file, body = file_readers.get("gridfs").read(
+            file_id, meta_only=meta_only
+        )
+        if meta_only:
+            self.write(
+                self.parser.serialize_request_file(request_file, to_string=False)
+            )
+        else:
+            size = 4096
+            self.set_header("Content-Type", "application/octet-stream")
+            self.set_header(
+                "Content-Disposition", "attachment; filename=%s" % request_file.filename
+            )
+            while True:
+                data = body.read(size)
+                if not data:
+                    break
+                self.write(data)
diff --git a/brew_view/file_readers/__init__.py b/brew_view/file_readers/__init__.py
new file mode 100644
index 00000000..06b7a31d
--- /dev/null
+++ b/brew_view/file_readers/__init__.py
@@ -0,0 +1,10 @@
+from brew_view.file_readers import gridfs
+
+
+def get(name):
+    """Get a file_reader"""
+    fname = str(name).lower()
+    if fname == "gridfs":
+        return gridfs
+    else:
+        raise TypeError("Invalid reader type: %s" % name)
diff --git a/brew_view/file_readers/gridfs.py b/brew_view/file_readers/gridfs.py
new file mode 100644
index 00000000..0b5a3260
--- /dev/null
+++ b/brew_view/file_readers/gridfs.py
@@ -0,0 +1,10 @@
+from bg_utils.mongo.models import RequestFile
+
+
+def read(file_id, **kwargs):
+    """Read a file from gridfs."""
+    r_file = RequestFile.objects.get(id=file_id)
+    if kwargs.get("meta_only", False):
+        return r_file, None
+
+    return r_file, r_file.body
diff --git a/brew_view/file_writers/__init__.py b/brew_view/file_writers/__init__.py
new file mode 100644
index 00000000..f4ff6d46
--- /dev/null
+++ b/brew_view/file_writers/__init__.py
@@ -0,0 +1,10 @@
+from brew_view.file_writers import gridfs
+
+
+def get(name):
+    """Get a file writer"""
+    fname = str(name).lower()
+    if fname == "gridfs":
+        return gridfs
+    else:
+        raise TypeError("Invalid writer type: %s" % name)
diff --git a/brew_view/file_writers/gridfs.py b/brew_view/file_writers/gridfs.py
new file mode 100644
index 00000000..d08adefa
--- /dev/null
+++ b/brew_view/file_writers/gridfs.py
@@ -0,0 +1,28 @@
+from bg_utils.mongo.models import RequestFile
+
+
+def write(body, **kwargs):
+    """Write the given body to gridfs.
+
+    Args:
+        body: Bytes
+
+    Keyword Args:
+        filename: required, name of the file.
+        encoding: default is None
+
+    Returns:
+        The ID of the RequestFile that was written to the database.
+
+    """
+    filename = kwargs.get("filename")
+    if not filename:
+        raise ValueError("Cannot write a gridfs document without a filename.")
+
+    encoding = kwargs.get("encoding")
+    r_file = RequestFile(filename=filename)
+    r_file.body.new_file(encoding=encoding)
+    r_file.body.write(body)
+    r_file.body.close()
+    r_file.save()
+    return str(r_file.id)
diff --git a/test/conftest.py b/test/conftest.py
index 9e3b0377..ac9f61d1 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -69,6 +69,11 @@ def thrift_context(thrift_client):
     )
 
 
+@pytest.fixture
+def mongo_request_file(bg_request_file):
+    return brew2mongo(bg_request_file)
+
+
 @pytest.fixture
 def mongo_system(bg_system):
     return brew2mongo(bg_system)
diff --git a/test/unit/controllers/file_api_test.py b/test/unit/controllers/file_api_test.py
new file mode 100644
index 00000000..33f4c309
--- /dev/null
+++ b/test/unit/controllers/file_api_test.py
@@ -0,0 +1,95 @@
+from uuid import uuid4
+
+import pytest
+import json
+from functools import partial
+from tornado import gen
+
+from bg_utils.mongo.models import RequestFile
+
+
+@pytest.fixture(autouse=True)
+def drop_files(app):
+    RequestFile.drop_collection()
+
+
+class TestFileAPI(object):
+    @pytest.mark.gen_test
+    def test_get_meta_only(
+        self, http_client, base_url, request_file_dict, mongo_request_file
+    ):
+        mongo_request_file.save()
+
+        response = yield http_client.fetch(
+            base_url + "/api/vbeta/files/" + str(mongo_request_file.id),
+            headers={"X-BG-file-meta-only": "true"},
+        )
+        assert 200 == response.code
+        assert request_file_dict == json.loads(response.body.decode("utf-8"))
+
+    @pytest.mark.gen_test
+    def test_get_file(self, http_client, base_url, mongo_request_file):
+        mongo_request_file.save()
+        response = yield http_client.fetch(
+            base_url + "/api/vbeta/files/" + str(mongo_request_file.id)
+        )
+        assert 200 == response.code
+        assert "Content-Type" in response.headers
+        assert "Content-Disposition" in response.headers
+        assert (
+            response.headers["Content-Disposition"]
+            == "attachment; filename=%s" % mongo_request_file.filename
+        )
+        assert response.body == b""
+
+    @pytest.mark.gen_test
+    def test_get_404(self, http_client, base_url, system_id):
+        response = yield http_client.fetch(
+            base_url + "/api/vbeta/files/" + system_id, raise_error=False
+        )
+        assert 404 == response.code
+
+    @pytest.fixture
+    def boundary(self):
+        return uuid4().hex
+
+    @staticmethod
+    @gen.coroutine
+    def multipart_producer(boundary, filename, body, write):
+        boundary_bytes = boundary.encode()
+        filename_bytes = filename.encode()
+        mtype = "application/octet-stream"
+        buf = (
+            (b"--%s\r\n" % boundary_bytes)
+            + (
+                b'Content-Disposition: form-data; name="%s"; filename="%s"\r\n'
+                % (filename_bytes, filename_bytes)
+            )
+            + (b"Content-Type: %s\r\n" % mtype.encode())
+            + b"\r\n"
+        )
+        yield write(buf)
+        yield write(body)
+        yield write(b"\r\n")
+        yield write(b"--%s--\r\n" % boundary_bytes)
+
+    @pytest.mark.gen_test
+    @pytest.mark.skip
+    def test_post(self, http_client, base_url, boundary, request_file_dict):
+        """Skipping until mongomock supports gridfs"""
+        headers = {"Content-Type": "multipart/form-data; boundary=%s" % boundary}
+        producer = partial(
+            self.multipart_producer, boundary, request_file_dict["filename"], b"0x01"
+        )
+        response = yield http_client.fetch(
+            base_url + "/api/vbeta/files/",
+            method="POST",
+            headers=headers,
+            body_producer=producer,
+        )
+        assert 201 == response.code
+        json_response = json.loads(response.body.decode("utf-8"))
+        assert "id" in json_response
+        assert "storage_type" in json_response
+        assert "filename" in json_response
+        assert "content_type" in json_response
diff --git a/test/unit/file_readers_test.py b/test/unit/file_readers_test.py
new file mode 100644
index 00000000..2a7fa46a
--- /dev/null
+++ b/test/unit/file_readers_test.py
@@ -0,0 +1,45 @@
+import pytest
+from mongoengine import connect
+
+import brew_view.file_readers as readers
+from bg_utils.mongo.models import RequestFile
+from brewtils.test.fixtures import request_file_dict
+
+
+def test_get():
+    reader = readers.get("gridfs")
+    assert hasattr(reader, "read")
+
+
+def test_get_fail():
+    with pytest.raises(TypeError):
+        readers.get("INVALID")
+
+
+class TestGridfs(object):
+    @pytest.fixture
+    def gridfs(self):
+        return readers.get("gridfs")
+
+    @pytest.fixture(scope="session", autouse=True)
+    def run_around(self):
+        connect("test", host="mongomock://localhost")
+
+    @pytest.fixture(autouse=True)
+    def remove_all(self):
+        RequestFile.objects().delete()
+        assert RequestFile.objects.count() == 0
+
+    def test_read_meta_only(self, gridfs, request_file_dict):
+        to_save = RequestFile(**request_file_dict)
+        to_save.save()
+        r_file, body = gridfs.read(to_save.id, meta_only=True)
+        assert isinstance(r_file, RequestFile)
+        assert body is None
+
+    def test_read(self, gridfs, request_file_dict):
+        to_save = RequestFile(**request_file_dict)
+        to_save.save()
+        r_file, body = gridfs.read(to_save.id, meta_only=False)
+        assert body == to_save.body
+        assert r_file == to_save
diff --git a/test/unit/file_writers_test.py b/test/unit/file_writers_test.py
new file mode 100644
index 00000000..5b2d9021
--- /dev/null
+++ b/test/unit/file_writers_test.py
@@ -0,0 +1,44 @@
+import pytest
+from mongoengine import connect
+import mongomock.gridfs
+
+import brew_view.file_writers as writers
+from bg_utils.mongo.models import RequestFile
+
+
+def test_get():
+    writer = writers.get("gridfs")
+    assert hasattr(writer, "write")
+
+
+def test_get_fail():
+    with pytest.raises(TypeError):
+        writers.get("INVALID")
+
+
+class TestGridfsWriter(object):
+    @pytest.fixture
+    def gridfs(self):
+        return writers.get("gridfs")
+
+    @pytest.fixture(scope="session", autouse=True)
+    def run_around(self):
+        connect("test", host="mongomock://localhost")
+        mongomock.gridfs.enable_gridfs_integration()
+
+    @pytest.fixture(autouse=True)
+    def remove_all(self):
+        RequestFile.objects().delete()
+        assert RequestFile.objects.count() == 0
+
+    def test_write_no_filename(self, gridfs):
+        with pytest.raises(ValueError):
+            gridfs.write("body")
+
+    @pytest.mark.skip
+    def test_write(self, gridfs):
+        """Skipping this test until mongomock supports gridfs"""
+        mongo_id = gridfs.write("BODY", filename="some_name")
+        assert mongo_id is not None
+        r_file = RequestFile.objects.get(id=mongo_id)
+        assert r_file.body == "BODY"