Repository: bentoml/BentoML
Pull request: Refactor(adapter): generic logic
See description for #972 

Problem statement:
Refactor adapters
## Is your feature request related to a problem? Please describe.
- Allowed users to customize HTTP response/ AWS lambda event like headers, etc
- Made contribution of adapters easier.
- Drier adapter code
- A more handy CLI UX for user defined API function

## Describe the solution you'd like

- Introduce InferenceTask & InferenceResult types
- Move micro-batching related logic out of adapters

### For adapter developers
Now a basic JSON input adapter would be super simple:
```python

ApiFuncArgs = Tuple[Sequence[JsonSerializable],]

class JsonInput(BaseInputAdapter):
    def from_http_request(self, req: HTTPRequest) -> InferenceTask:
        return  InferenceTask(http_headers=r.parsed_headers, data=req.body)

    def from_aws_lambda_event(self, event: AwsLambdaEvent) -> InferenceTask:
        return InferenceTask(aws_lambda_event=event, data=event['body'].encode(JSON_CHARSET))

    def from_cli(self, cli_args: Sequence[str]) -> Iterator[InferenceTask]:
        for json_file in parse_cli_input(cli_args):
            yield InferenceTask(cli_args=cli_args, data=json_file.read())

    def extract_user_func_args(self, tasks: Sequence[InferenceTask]) -> ApiFuncArgs:
        json_inputs = []
        for task in tasks:
            try:
                json_str = task.data.decode(JSON_CHARSET)
                parsed_json = json.loads(json_str)
                json_inputs.append(parsed_json)
            except UnicodeDecodeError:
                task.discard(
                    http_status=400, err_msg=f"JSON must be encoded in {JSON_CHARSET}"
                )
            except json.JSONDecodeError:
                task.discard(http_status=400, err_msg="Not a valid JSON format")
            except Exception:  # pylint: disable=broad-except
                err = traceback.format_exc()
                task.discard(http_status=500, err_msg=f"Internal Server Error: {err}")
        return (json_inputs,)
```

And it allows user to query:

**by CLI**
```shell
bentoml run <yourservice>:latest --input {"height": 180, "weight": 80} {"height": 190, "weight": 100}
```

```shell
bentoml run <yourservice>:latest --input-file path/to/your/*.json --max-batch-size 10
```

**by HTTP**
```shell
curl -i \
    --header "Content-Type: application/json" \
    --request POST \
    --data '{"height": 190, "weight": 100}' \
    localhost:5000/predict
```

### For Users
Users could customize HTTP headers, AWS lambda events and CLI exit code by return instances of InferenceResult in their API function:

```python
import bentoml
from bentoml.types import InferenceResult, InferenceContext

class MyService(bentoml.BentoService):
    @bentoml.api(input=JsonInput())
    def predict(self, json_obj_list, contexts=contexts):
        rv = []
        predictions = user_model.run(json_obj_list)
        for context, prediction in zip(contexts, predictions):
            if context.http_headers['Accept'] == "application/json":
                rv.append(
                    InferenceResult(
                        data=prediction,
                        http_status=200,
                        http_headers={"Content-Type": "application/json"},
                        aws_lambda_event={"data": prediction},
                        cli_status=0,
                ))
        return rv
```


## TODO
decision:
- [ ] enable BATCH_MODE_SUPPORTED for all adapters; enable batching server by default?
- [x] ~~merge FileInput & MultiFileInput and so on?~~
- [x] ~~remove legacy adapters?~~

cli:
- [x] multiple cli input
- [ ] better cli error message (now just print err_msg)
- [ ] system cli args encoding?

adapters:
- [x] prototype json input adapter
- [x] annotated_image_input
- [x] file input adapter and its tests
- [x] dataframe input adapter and its tests
- [x] image input adapter and its tests
- [x] tensorflow tensor adapter and its tests
- [x] other input adapters and their tests
- [x] other output adapters

chore:
- [ ] apply `set_dependencies`
- [ ] benchmark
- [ ] test clipper handler
- [ ] 2e2 tests

Hints from issue discussion:

Commit message: Various changes and typo fixes

Patch to review:
diff --git a/bentoml/__init__.py b/bentoml/__init__.py
index 346333ecb5f..c1f09d3b4a9 100644
--- a/bentoml/__init__.py
+++ b/bentoml/__init__.py
@@ -35,7 +35,6 @@
     save,
 )
 
-
 __all__ = [
     "__version__",
     "api",
diff --git a/bentoml/_version.py b/bentoml/_version.py
index d3f3d35be71..ce8334d6647 100644
--- a/bentoml/_version.py
+++ b/bentoml/_version.py
@@ -247,10 +247,7 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, run_command=run_command):
         raise NotThisMethod("'git rev-parse' failed")
     full_out = full_out.strip()
 
-    pieces = {}
-    pieces["long"] = full_out
-    pieces["short"] = full_out[:7]  # maybe improved later
-    pieces["error"] = None
+    pieces = {"long": full_out, "short": full_out[:7], "error": None}
 
     # parse describe_out. It will be like TAG-NUM-gHEX[-dirty] or HEX[-dirty]
     # TAG might have hyphens.
diff --git a/bentoml/adapters/annotated_image_input.py b/bentoml/adapters/annotated_image_input.py
index e992f3366d7..ce47cdf4dfa 100644
--- a/bentoml/adapters/annotated_image_input.py
+++ b/bentoml/adapters/annotated_image_input.py
@@ -24,7 +24,7 @@
 
 from bentoml import config
 from bentoml.utils.lazy_loader import LazyLoader
-from bentoml.marshal.utils import SimpleRequest, SimpleResponse
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml.exceptions import BadInput
 from bentoml.adapters.base_input import BaseInputAdapter
 
@@ -127,7 +127,7 @@ class AnnotatedImageInput(BaseInputAdapter):
         >>>
         >>> CLASS_NAMES = ['cat', 'dog']
         >>>
-        >>> @artifacts([TensorflowArtifact('classifer')])
+        >>> @artifacts([TensorflowArtifact('classifier')])
         >>> class PetClassification(BentoService):
         >>>    @api(input=AnnotatedImageInput())
         >>>    def predict(self, image: Numpy.array, annotations: JsonSerializable):
@@ -276,8 +276,8 @@ def _load_image_and_json_data(self, request: Request):
         return {self.image_input_name: input_image}
 
     def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func: callable
-    ) -> Iterable[SimpleResponse]:
+        self, requests: Iterable[HTTPRequest], func: callable
+    ) -> Iterable[HTTPResponse]:
         """
         Batch version of handle_request
         """
@@ -306,7 +306,7 @@ def handle_batch_request(
 
         results = [func(**d) if d else {} for d in input_datas]
 
-        return self.output_adapter.to_batch_response(
+        return self.output_adapter.to_http_response(
             result_conc=results,
             slices=slices,
             fallbacks=[None] * len(slices),
diff --git a/bentoml/adapters/base_input.py b/bentoml/adapters/base_input.py
index 4343da30db8..9bd2b4eeaea 100644
--- a/bentoml/adapters/base_input.py
+++ b/bentoml/adapters/base_input.py
@@ -11,104 +11,99 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import argparse
+from typing import Iterable, Generic, Tuple, Iterator
 
-from typing import Iterable
+from bentoml.types import (
+    ApiFuncArgs,
+    HTTPRequest,
+    InferenceTask,
+    AwsLambdaEvent,
+)
 
 
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest, BATCH_REQUEST_HEADER
-
-
-class BaseInputAdapter:
-    """InputAdapter is an abstraction layer between user defined API callback function
+class BaseInputAdapter(Generic[ApiFuncArgs]):
+    """
+    InputAdapter is an abstraction layer between user defined API callback function
     and prediction request input in a variety of different forms, such as HTTP request
     body, command line arguments or AWS Lambda event object.
     """
 
     HTTP_METHODS = ["POST", "GET"]
-
     BATCH_MODE_SUPPORTED = False
 
-    def __init__(self, output_adapter=None, http_input_example=None, **base_config):
-        '''
-        base_configs:
-            - is_batch_input
-        '''
+    def __init__(self, http_input_example=None, **base_config):
         self._config = base_config
-        self._output_adapter = output_adapter
         self._http_input_example = http_input_example
 
     @property
     def config(self):
-        if getattr(self, '_config', None) is None:
-            self._config = {}
         return self._config
 
-    def is_batch_request(self, request):
-        if BATCH_REQUEST_HEADER in request.parsed_headers:
-            return request.parsed_headers[BATCH_REQUEST_HEADER] != 'false'
-        return self.config.get("is_batch_input", False)
+    @property
+    def request_schema(self):
+        """
+        :return: OpenAPI json schema for the HTTP API endpoint created with this input
+                 adapter
+        """
+        return {"application/json": {"schema": {"type": "object"}}}
 
     @property
-    def output_adapter(self):
-        if self._output_adapter is None:
-            from .default_output import DefaultOutput
+    def pip_dependencies(self):
+        """
+        :return: List of PyPI package names required by this InputAdapter
+        """
+        return []
 
-            self._output_adapter = DefaultOutput()
-        return self._output_adapter
+    def from_http_request(self, reqs: Iterable[HTTPRequest]) -> Iterable[InferenceTask]:
+        """
+        Handles HTTP requests, convert it into InferenceTasks
+        """
+        raise NotImplementedError()
 
-    def handle_request(self, request, func):
-        """Handles an HTTP request, convert it into corresponding data
-        format that user API function is expecting, and return API
-        function result as the HTTP response to client
+    def from_aws_lambda_event(
+        self, events: Tuple[AwsLambdaEvent]
+    ) -> Iterable[InferenceTask]:
+        """
+        Handles AWS lambda events, convert it into InferenceTasks
+        """
+        raise NotImplementedError()
 
-        :param request: Flask request object
-        :param func: user API function
+    def from_cli(self, cli_args: Tuple[str]) -> Iterable[InferenceTask]:
+        """
+        Handles CLI command, generate InferenceTasks
         """
-        raise NotImplementedError
+        raise NotImplementedError()
 
-    def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func
-    ) -> Iterable[SimpleResponse]:
-        """Handles an HTTP request, convert it into corresponding data
-        format that user API function is expecting, and return API
-        function result as the HTTP response to client
+    def validate_task(self, _: InferenceTask):
+        return True
 
-        :param requests: List of flask request object
-        :param func: user API function
+    def extract_user_func_args(self, tasks: Iterable[InferenceTask]) -> ApiFuncArgs:
+        """
+        Extract args that user API function is expecting from InferenceTasks
         """
-        raise NotImplementedError
+        raise NotImplementedError()
 
-    def handle_cli(self, args, func):
-        """Handles an CLI command call, convert CLI arguments into
-        corresponding data format that user API function is expecting, and
-        prints the API function result to console output
 
-        :param args: CLI arguments
-        :param func: user API function
-        """
-        raise NotImplementedError
+def parse_cli_input(cli_args: Iterable[str]) -> Iterator[bytes]:
+    parser = argparse.ArgumentParser()
+    input_g = parser.add_mutually_exclusive_group(required=True)
+    input_g.add_argument('--input', nargs="+", type=str)
+    input_g.add_argument('--input-file', nargs="+")
 
-    def handle_aws_lambda_event(self, event, func):
-        """Handles a Lambda event, convert event dict into corresponding
-        data format that user API function is expecting, and use API
-        function result as response
+    parsed_args, _ = parser.parse_known_args(list(cli_args))
 
-        :param event: AWS lambda event data of the python `dict` type
-        :param func: user API function
-        """
-        raise NotImplementedError
+    inputs = tuple(
+        parsed_args.input if parsed_args.input_file is None else parsed_args.input_file
+    )
+    is_file = parsed_args.input_file is not None
+    if is_file:
+        for input_ in inputs:
+            with open(input_, "rb") as f:
+                yield f.read()
 
-    @property
-    def request_schema(self):
-        """
-        :return: OpenAPI json schema for the HTTP API endpoint created with this input
-                 adapter
-        """
-        return {"application/json": {"schema": {"type": "object"}}}
+    else:
+        for input_ in inputs:
+            yield input_.encode()
 
-    @property
-    def pip_dependencies(self):
-        """
-        :return: List of PyPI package names required by this InputAdapter
-        """
-        return []
+    return _
diff --git a/bentoml/adapters/base_output.py b/bentoml/adapters/base_output.py
index 0fb93210e18..9b4018fb95e 100644
--- a/bentoml/adapters/base_output.py
+++ b/bentoml/adapters/base_output.py
@@ -12,15 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Iterable
+from typing import Iterable, Tuple, Generic
 
-import flask
+from bentoml.types import (
+    AwsLambdaEvent,
+    HTTPResponse,
+    ApiFuncReturnValue,
+    InferenceResult,
+    InferenceContext,
+)
 
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
 
-
-class BaseOutputAdapter:
-    """OutputAdapter is an layer between result of user defined API callback function
+class BaseOutputAdapter(Generic[ApiFuncReturnValue]):
+    """
+    OutputAdapter is an layer between result of user defined API callback function
     and final output in a variety of different forms,
     such as HTTP response, command line stdout or AWS Lambda event object.
     """
@@ -32,46 +37,39 @@ def __init__(self, cors='*'):
     def config(self):
         return dict(cors=self.cors,)
 
-    def to_response(self, result, request: flask.Request) -> flask.Response:
-        """Converts corresponding data into an HTTP response
-
-        :param result: result of user API function
-        :param request: request object
+    @property
+    def pip_dependencies(self):
         """
-        simple_req = SimpleRequest.from_flask_request(request)
-        simple_resp = self.to_batch_response((result,), requests=(simple_req,))[0]
-        return simple_resp.to_flask_response()
-
-    def to_batch_response(
-        self, result_conc, slices=None, fallbacks=None, requests=None,
-    ) -> Iterable[SimpleResponse]:
-        """Converts corresponding data merged by batching service into HTTP responses
+        :return: List of PyPI package names required by this OutputAdapter
+        """
+        return []
 
-        :param result_conc: result of user API function
-        :param slices: auto-batching slices
-        :param requests: request objects
+    def pack_user_func_return_value(
+        self, return_result: ApiFuncReturnValue, contexts: Tuple[InferenceContext],
+    ) -> Tuple[InferenceResult]:
+        """
+        Pack the return value of user defined API function into InferenceResults
         """
         raise NotImplementedError()
 
-    def to_cli(self, result, args):
-        """Converts corresponding data into an CLI output.
-
-        :param result: result of user API function
-        :param args: CLI args
+    def to_http_response(
+        self, results: Iterable[InferenceResult]
+    ) -> Iterable[HTTPResponse]:
+        """
+        Converts InferenceResults into HTTP responses.
         """
         raise NotImplementedError()
 
-    def to_aws_lambda_event(self, result, event):
-        """Converts corresponding data into a Lambda event.
-
-        :param result: result of user API function
-        :param event: input event
+    def to_cli(self, results: Iterable[InferenceResult]) -> int:
         """
-        raise NotImplementedError
+        Converts InferenceResults into CLI output.
+        """
+        raise NotImplementedError()
 
-    @property
-    def pip_dependencies(self):
+    def to_aws_lambda_event(
+        self, results: Iterable[InferenceResult]
+    ) -> Iterable[AwsLambdaEvent]:
         """
-        :return: List of PyPI package names required by this OutputAdapter
+        Converts InferenceResults into AWS lambda events.
         """
-        return []
+        raise NotImplementedError()
diff --git a/bentoml/adapters/clipper_input.py b/bentoml/adapters/clipper_input.py
index b3d6610afcf..038c84dfe72 100644
--- a/bentoml/adapters/clipper_input.py
+++ b/bentoml/adapters/clipper_input.py
@@ -39,7 +39,7 @@ def pip_dependencies(self):
         # to a clipper cluster, not required by the API Server itself
         return []
 
-    def handle_request(self, request, func):
+    def handle_request(self, request):
         raise NotImplementedError(
             "ClipperInput does not support handling REST API prediction request"
         )
diff --git a/bentoml/adapters/dataframe_input.py b/bentoml/adapters/dataframe_input.py
index cab6700222d..1a37d393ae5 100644
--- a/bentoml/adapters/dataframe_input.py
+++ b/bentoml/adapters/dataframe_input.py
@@ -30,7 +30,7 @@
     check_dataframe_column_contains,
     PANDAS_DATAFRAME_TO_JSON_ORIENT_OPTIONS,
 )
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml.utils.s3 import is_s3_url
 from bentoml.exceptions import BadInput, MissingDependencyException
 
@@ -66,7 +66,7 @@ def __init__(
         **base_kwargs,
     ):
         if not is_batch_input:
-            raise ValueError('DataframeInput can not accpept none batch inputs')
+            raise ValueError('DataframeInput can not accept none batch inputs')
         super(DataframeInput, self).__init__(
             is_batch_input=is_batch_input, **base_kwargs
         )
@@ -144,7 +144,7 @@ def request_schema(self):
 
         return default
 
-    def handle_request(self, request: flask.Request, func):
+    def handle_request(self, request: flask.Request):
         if request.content_type == "text/csv":
             csv_string = StringIO(request.get_data(as_text=True))
             df = pd.read_csv(csv_string)
@@ -167,10 +167,10 @@ def handle_request(self, request: flask.Request, func):
         return self.output_adapter.to_response(result, request)
 
     def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func
-    ) -> Iterable[SimpleResponse]:
+        self, requests: Iterable[HTTPRequest], func
+    ) -> Iterable[HTTPResponse]:
 
-        datas = [r.data for r in requests]
+        datas = [r.body for r in requests]
         content_types = [
             r.parsed_headers.get('content-type', 'application/json') for r in requests
         ]
diff --git a/bentoml/adapters/dataframe_output.py b/bentoml/adapters/dataframe_output.py
index 0cf66d49999..d0fdb361102 100644
--- a/bentoml/adapters/dataframe_output.py
+++ b/bentoml/adapters/dataframe_output.py
@@ -17,8 +17,8 @@
 import argparse
 
 from bentoml.exceptions import BentoMLException
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml.utils.dataframe_util import PANDAS_DATAFRAME_TO_JSON_ORIENT_OPTIONS
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
 from bentoml.adapters.base_output import BaseOutputAdapter
 
 
@@ -27,7 +27,7 @@ def df_to_json(result, pandas_dataframe_orient="records"):
 
     assert (
         pandas_dataframe_orient in PANDAS_DATAFRAME_TO_JSON_ORIENT_OPTIONS
-    ), f"unkown pandas dataframe orient '{pandas_dataframe_orient}'"
+    ), f"unknown pandas dataframe orient '{pandas_dataframe_orient}'"
 
     if isinstance(result, pd.DataFrame):
         return result.to_json(orient=pandas_dataframe_orient)
@@ -68,8 +68,8 @@ def to_batch_response(
         result_conc,
         slices=None,
         fallbacks=None,
-        requests: Iterable[SimpleRequest] = None,
-    ) -> Iterable[SimpleResponse]:
+        requests: Iterable[HTTPRequest] = None,
+    ) -> Iterable[HTTPResponse]:
         # TODO(bojiang): header content_type
 
         if slices is None:
@@ -87,13 +87,13 @@ def to_batch_response(
                 json_output = df_to_json(
                     result, pandas_dataframe_orient=self.output_orient
                 )
-                responses[i] = SimpleResponse(
+                responses[i] = HTTPResponse(
                     200, (("Content-Type", "application/json"),), json_output
                 )
             except AssertionError as e:
-                responses[i] = SimpleResponse(400, None, str(e))
+                responses[i] = HTTPResponse(400, body=str(e))
             except Exception as e:  # pylint: disable=broad-except
-                responses[i] = SimpleResponse(500, None, str(e))
+                responses[i] = HTTPResponse(500, body=str(e))
         return responses
 
     def to_cli(self, result, args):
diff --git a/bentoml/adapters/default_output.py b/bentoml/adapters/default_output.py
index 4695882829b..7875316b6fb 100644
--- a/bentoml/adapters/default_output.py
+++ b/bentoml/adapters/default_output.py
@@ -12,13 +12,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Iterable
-
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
+from typing import Iterable, Tuple, Type, Union
+
+from bentoml.types import (
+    ApiFuncReturnValue,
+    HTTPResponse,
+    InferenceResult,
+    InferenceContext,
+)
 from .base_output import BaseOutputAdapter
 
 
-def detect_suitable_adapter(result, slices=None):
+def detect_suitable_adapter(result) -> Type[BaseOutputAdapter]:
     try:
         import pandas as pd
 
@@ -39,12 +44,8 @@ def detect_suitable_adapter(result, slices=None):
     except ImportError:
         pass
 
-    if slices is not None:
-        for s in slices:
-            if s:
-                return detect_suitable_adapter(result[s])
-    if isinstance(result, (list, tuple)):
-        return detect_suitable_adapter(result[0])
+    # if isinstance(result, (list, tuple)):
+    # return detect_suitable_adapter(result[0])
 
     from .json_output import JsonSerializableOutput
 
@@ -66,41 +67,23 @@ def __init__(self, **kwargs):
         super(DefaultOutput, self).__init__(**kwargs)
         self.actual_adapter = None
 
-    def to_batch_response(
-        self,
-        result_conc,
-        slices=None,
-        fallbacks=None,
-        requests: Iterable[SimpleRequest] = None,
-    ) -> Iterable[SimpleResponse]:
-        """Converts corresponding data merged by batching service into HTTP responses
-
-        :param result_conc: result of user API function
-        :param slices: auto-batching slices
-        :param requests: request objects
-        """
+    def pack_user_func_return_value(
+        self, return_result: ApiFuncReturnValue, contexts: Tuple[InferenceContext],
+    ) -> Tuple[InferenceResult]:
         if self.actual_adapter is None:
-            self.actual_adapter = detect_suitable_adapter(result_conc, slices)()
-        return self.actual_adapter.to_batch_response(
-            result_conc, slices, fallbacks, requests
-        )
+            self.actual_adapter = detect_suitable_adapter(return_result)()
+        if self.actual_adapter:
+            return self.actual_adapter.pack_user_func_return_value(
+                return_result, contexts=contexts
+            )
 
-    def to_cli(self, result, args):
-        """Converts corresponding data into an CLI output.
+        raise NotImplementedError()
 
-        :param result: result of user API function
-        :param args: CLI args
-        """
-        if self.actual_adapter is None:
-            self.actual_adapter = detect_suitable_adapter(result)()
-        return self.actual_adapter.to_cli(result, args)
+    def to_http_response(self, results) -> Iterable[HTTPResponse]:
+        return self.actual_adapter.to_http_response(results)
 
-    def to_aws_lambda_event(self, result, event):
-        """Converts corresponding data into a Lambda event.
+    def to_cli(self, results):
+        return self.actual_adapter.to_cli(results)
 
-        :param result: result of user API function
-        :param event: input event
-        """
-        if self.actual_adapter is None:
-            self.actual_adapter = detect_suitable_adapter(result)()
-        return self.actual_adapter.to_aws_lambda_event(result, event)
+    def to_aws_lambda_event(self, results):
+        return self.actual_adapter.to_aws_lambda_event(results)
diff --git a/bentoml/adapters/fastai_image_input.py b/bentoml/adapters/fastai_image_input.py
index 11b699c4df2..7b5ecfc6f5a 100644
--- a/bentoml/adapters/fastai_image_input.py
+++ b/bentoml/adapters/fastai_image_input.py
@@ -55,7 +55,7 @@ def api_func_result_to_json(result, pandas_dataframe_orient="records"):
 
     assert (
         pandas_dataframe_orient in PANDAS_DATAFRAME_TO_JSON_ORIENT_OPTIONS
-    ), f"unkown pandas dataframe orient '{pandas_dataframe_orient}'"
+    ), f"unknown pandas dataframe orient '{pandas_dataframe_orient}'"
 
     if pd and isinstance(result, pd.DataFrame):
         return result.to_json(orient=pandas_dataframe_orient)
@@ -156,7 +156,7 @@ def pip_dependencies(self):
     def handle_batch_request(self, requests, func):
         raise NotImplementedError
 
-    def handle_request(self, request, func):
+    def handle_request(self, request):
         input_streams = []
         for filename in self.input_names:
             file = request.files.get(filename)
diff --git a/bentoml/adapters/file_input.py b/bentoml/adapters/file_input.py
index e10168fe4ff..1dbb82e5ddf 100644
--- a/bentoml/adapters/file_input.py
+++ b/bentoml/adapters/file_input.py
@@ -25,9 +25,9 @@
 
 from werkzeug.wrappers import Request
 
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml.exceptions import BadInput
 from bentoml.adapters.base_input import BaseInputAdapter
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
 
 
 class FileInput(BaseInputAdapter):
@@ -79,7 +79,7 @@ def predict(self, file_streams):
     BATCH_MODE_SUPPORTED = True
 
     def __init__(
-        self, **base_kwargs,
+            self, **base_kwargs,
     ):
         super(FileInput, self).__init__(**base_kwargs)
 
@@ -117,8 +117,8 @@ def _load_file(self, request: Request):
         return input_stream
 
     def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func: callable
-    ) -> Iterable[SimpleResponse]:
+            self, requests: Iterable[HTTPRequest], func: callable
+    ) -> Iterable[HTTPResponse]:
         """
         Batch version of handle_request
         """
@@ -126,12 +126,12 @@ def handle_batch_request(
         ids = []
 
         for i, req in enumerate(requests):
-            if not req.data:
+            if not req.body:
                 ids.append(None)
                 continue
             request = Request.from_values(
-                input_stream=BytesIO(req.data),
-                content_length=len(req.data),
+                input_stream=BytesIO(req.body),
+                content_length=len(req.body),
                 headers=req.headers,
             )
             try:
@@ -146,7 +146,7 @@ def handle_batch_request(
         results = func(input_datas) if input_datas else []
         return self.output_adapter.to_batch_response(results, ids, requests)
 
-    def handle_request(self, request, func):
+    def handle_request(self, request):
         """Handle http request that has one file. It will convert file into a
         BytesIO object for the function to consume.
 
@@ -172,7 +172,7 @@ def handle_cli(self, args, func):
         )
 
         for i in range(0, len(file_paths), batch_size):
-            step_file_paths = file_paths[i : i + batch_size]
+            step_file_paths = file_paths[i: i + batch_size]
             input_list = []
             with contextlib.ExitStack() as stack:
                 for file_path in step_file_paths:
diff --git a/bentoml/adapters/image_input.py b/bentoml/adapters/image_input.py
index 909ca85efa0..abfd06cfd26 100644
--- a/bentoml/adapters/image_input.py
+++ b/bentoml/adapters/image_input.py
@@ -21,9 +21,9 @@
 from werkzeug.utils import secure_filename
 from werkzeug.wrappers import Request
 
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml import config
 from bentoml.utils.lazy_loader import LazyLoader
-from bentoml.marshal.utils import SimpleRequest, SimpleResponse
 from bentoml.exceptions import BadInput
 from bentoml.adapters.base_input import BaseInputAdapter
 
@@ -50,8 +50,8 @@ def get_default_accept_image_formats():
     return [
         extension.strip()
         for extension in config("apiserver")
-        .get("default_image_input_accept_file_extensions")
-        .split(",")
+            .get("default_image_input_accept_file_extensions")
+            .split(",")
     ]
 
 
@@ -87,7 +87,7 @@ class ImageInput(BaseInputAdapter):
         >>>
         >>> CLASS_NAMES = ['cat', 'dog']
         >>>
-        >>> @artifacts([TensorflowArtifact('classifer')])
+        >>> @artifacts([TensorflowArtifact('classifier')])
         >>> class PetClassification(BentoService):
         >>>     @api(input=ImageInput())
         >>>     def predict(self, image_ndarrays):
@@ -99,16 +99,16 @@ class ImageInput(BaseInputAdapter):
     BATCH_MODE_SUPPORTED = True
 
     def __init__(
-        self,
-        accept_image_formats=None,
-        pilmode="RGB",
-        is_batch_input=False,
-        **base_kwargs,
+            self,
+            accept_image_formats=None,
+            pilmode="RGB",
+            is_batch_input=False,
+            **base_kwargs,
     ):
         assert imageio, "`imageio` dependency can be imported"
 
         if is_batch_input:
-            raise ValueError('ImageInput can not accpept batch inputs')
+            raise ValueError('ImageInput can not accept batch inputs')
         super(ImageInput, self).__init__(is_batch_input=is_batch_input, **base_kwargs)
         if 'input_names' in base_kwargs:
             raise TypeError(
@@ -119,7 +119,7 @@ def __init__(
 
         self.pilmode = pilmode
         self.accept_image_formats = (
-            accept_image_formats or get_default_accept_image_formats()
+                accept_image_formats or get_default_accept_image_formats()
         )
 
     @property
@@ -173,20 +173,20 @@ def _load_image_data(self, request: Request):
         return input_data
 
     def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func: callable
-    ) -> Iterable[SimpleResponse]:
+            self, requests: Iterable[HTTPRequest], func: callable
+    ) -> Iterable[HTTPResponse]:
         """
         Batch version of handle_request
         """
         input_datas = []
         ids = []
         for i, req in enumerate(requests):
-            if not req.data:
+            if not req.body:
                 ids.append(None)
                 continue
             request = Request.from_values(
-                input_stream=BytesIO(req.data),
-                content_length=len(req.data),
+                input_stream=BytesIO(req.body),
+                content_length=len(req.body),
                 headers=req.headers,
             )
             try:
@@ -201,20 +201,14 @@ def handle_batch_request(
         results = func(input_datas) if input_datas else []
         return self.output_adapter.to_batch_response(results, ids, requests)
 
-    def handle_request(self, request, func):
+    def handle_request(self, headers, body):
         """Handle http request that has one image file. It will convert image into a
         ndarray for the function to consume.
 
-        Args:
-            request: incoming request object.
-            func: function that will take ndarray as its arg.
-            options: configuration for handling request object.
-        Return:
-            response object
+        :param headers: the request's HTTP HEADERS
+        :param body: the request body
         """
-        input_data = self._load_image_data(request)
-        result = func((input_data,))[0]
-        return self.output_adapter.to_response(result, request)
+        return self._load_image_data(Request.from_values(headers=headers, data=body))
 
     def handle_cli(self, args, func):
         parser = argparse.ArgumentParser()
@@ -228,7 +222,7 @@ def handle_cli(self, args, func):
         )
 
         for i in range(0, len(file_paths), batch_size):
-            step_file_paths = file_paths[i : i + batch_size]
+            step_file_paths = file_paths[i: i + batch_size]
             image_arrays = []
             for file_path in step_file_paths:
                 verify_image_format_or_raise(file_path, self.accept_image_formats)
diff --git a/bentoml/adapters/json_input.py b/bentoml/adapters/json_input.py
index ec3b83bc24d..251fb97fd05 100644
--- a/bentoml/adapters/json_input.py
+++ b/bentoml/adapters/json_input.py
@@ -12,21 +12,28 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
 import json
-import argparse
-from typing import Iterable
-from json import JSONDecodeError
+import traceback
+from typing import Iterable, Tuple, Iterator
 
-import flask
 
-from bentoml.exceptions import BadInput
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
-from bentoml.adapters.base_input import BaseInputAdapter
-from bentoml.adapters.utils import concat_list
+from bentoml.types import (
+    HTTPRequest,
+    JsonSerializable,
+    AwsLambdaEvent,
+    InferenceTask,
+    InferenceContext,
+    JSON_CHARSET,
+)
+from bentoml.adapters.base_input import BaseInputAdapter, parse_cli_input
 
 
-class JsonInput(BaseInputAdapter):
+ApiFuncArgs = Tuple[
+    Tuple[JsonSerializable],
+]
+
+
+class JsonInput(BaseInputAdapter[ApiFuncArgs]):
     """JsonInput parses REST API request or CLI command into parsed_jsons(a list of
     json serializable object in python) and pass down to user defined API function
 
@@ -60,70 +67,47 @@ def predict(self, parsed_jsons):
 
     BATCH_MODE_SUPPORTED = True
 
-    def __init__(self, is_batch_input=False, **base_kwargs):
-        super(JsonInput, self).__init__(is_batch_input=is_batch_input, **base_kwargs)
+    def from_http_request(
+        self, reqs: Iterable[HTTPRequest]
+    ) -> Tuple[InferenceTask[bytes]]:
+        return tuple(
+            InferenceTask(
+                context=InferenceContext(http_headers=r.parsed_headers), data=r.body,
+            )
+            for r in reqs
+        )
 
-    def handle_request(self, request: flask.Request, func):
-        if request.content_type != "application/json":
-            raise BadInput(
-                "Request content-type must be 'application/json' for this "
-                "BentoService API"
+    def from_aws_lambda_event(
+        self, events: Iterable[AwsLambdaEvent]
+    ) -> Tuple[InferenceTask[bytes]]:
+        return tuple(
+            InferenceTask(
+                context=InferenceContext(aws_lambda_event=e),
+                data=e['body'].encode(JSON_CHARSET),
             )
-        resps = self.handle_batch_request(
-            [SimpleRequest.from_flask_request(request)], func
+            for e in events
         )
-        return resps[0].to_flask_response()
-
-    def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func
-    ) -> Iterable[SimpleResponse]:
-        bad_resp = SimpleResponse(400, None, "Bad Input")
-        instances_list = [None] * len(requests)
-        fallbacks = [bad_resp] * len(requests)
-        batch_flags = [None] * len(requests)
-
-        for i, request in enumerate(requests):
-            batch_flags[i] = self.is_batch_request(request)
+
+    def from_cli(self, cli_args: Tuple[str]) -> Iterator[InferenceTask[bytes]]:
+        for i in parse_cli_input(cli_args):
+            yield InferenceTask(context=InferenceContext(cli_args=cli_args), data=i)
+
+    def extract_user_func_args(
+        self, tasks: Iterable[InferenceTask[bytes]]
+    ) -> ApiFuncArgs:
+        json_inputs = []
+        for task in tasks:
             try:
-                raw_str = request.data
-                parsed_json = json.loads(raw_str)
-                instances_list[i] = parsed_json
-            except (json.JSONDecodeError, UnicodeDecodeError):
-                fallbacks[i] = SimpleResponse(400, None, "Not a valid json")
+                json_str = task.data.decode(JSON_CHARSET)
+                parsed_json = json.loads(json_str)
+                json_inputs.append(parsed_json)
+            except UnicodeDecodeError:
+                task.discard(
+                    http_status=400, err_msg=f"JSON must be encoded in {JSON_CHARSET}"
+                )
+            except json.JSONDecodeError:
+                task.discard(http_status=400, err_msg="Not a valid JSON format")
             except Exception:  # pylint: disable=broad-except
-                import traceback
-
                 err = traceback.format_exc()
-                fallbacks[i] = SimpleResponse(
-                    500, None, f"Internal Server Error: {err}"
-                )
-
-        merged_instances, slices = concat_list(instances_list, batch_flags=batch_flags)
-        merged_result = func(merged_instances)
-        return self.output_adapter.to_batch_response(
-            merged_result, slices, fallbacks, requests
-        )
-
-    def handle_cli(self, args, func):
-        parser = argparse.ArgumentParser()
-        parser.add_argument("--input", required=True)
-        parsed_args, unknown_args = parser.parse_known_args(args)
-
-        if os.path.isfile(parsed_args.input):
-            with open(parsed_args.input, "r") as content_file:
-                content = content_file.read()
-        else:
-            content = parsed_args.input
-
-        input_json = json.loads(content)
-        result = func([input_json])[0]
-        return self.output_adapter.to_cli(result, unknown_args)
-
-    def handle_aws_lambda_event(self, event, func):
-        try:
-            parsed_json = json.loads(event["body"])
-        except JSONDecodeError:
-            raise BadInput("Request body must contain valid json")
-
-        result = func([parsed_json])[0]
-        return self.output_adapter.to_aws_lambda_event(result, event)
+                task.discard(http_status=500, err_msg=f"Internal Server Error: {err}")
+        return json_inputs,
diff --git a/bentoml/adapters/json_output.py b/bentoml/adapters/json_output.py
index b1a08ffec1e..d5f0399e10a 100644
--- a/bentoml/adapters/json_output.py
+++ b/bentoml/adapters/json_output.py
@@ -12,17 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Iterable
 import json
-
 import argparse
-
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
+from typing import Iterable, Tuple, Iterator
+
+from bentoml.types import (
+    HTTPResponse,
+    InferenceContext,
+    DefaultErrorContext,
+    InferenceResult,
+    JsonSerializable,
+)
 from bentoml.adapters.utils import NumpyJsonEncoder
 from bentoml.adapters.base_output import BaseOutputAdapter
 
 
-class JsonSerializableOutput(BaseOutputAdapter):
+ApiFuncReturnValue = Tuple[JsonSerializable]
+
+
+class JsonSerializableOutput(BaseOutputAdapter[ApiFuncReturnValue]):
     """
     Converts result of user defined API function into specific output.
 
@@ -32,66 +40,82 @@ class JsonSerializableOutput(BaseOutputAdapter):
             the header will not be set.
     """
 
-    def to_batch_response(
-        self,
-        result_conc,
-        slices=None,
-        fallbacks=None,
-        requests: Iterable[SimpleRequest] = None,
-    ) -> Iterable[SimpleResponse]:
-        # TODO(bojiang): header content_type
-
-        if slices is None:
-            slices = [i for i, _ in enumerate(result_conc)]
-        if fallbacks is None:
-            fallbacks = [None] * len(slices)
-
-        responses = [None] * len(slices)
-
-        for i, (s, f) in enumerate(zip(slices, fallbacks)):
-            if s is None:
-                responses[i] = f
-                continue
-
-            result = result_conc[s]
+    def pack_user_func_return_value(
+        self, return_result: ApiFuncReturnValue, contexts: Tuple[InferenceContext],
+    ) -> Tuple[InferenceResult[str]]:
+        results = []
+        for json_obj, context in zip(return_result, contexts):
+            args = context.cli_args
+            if args:
+                parser = argparse.ArgumentParser()
+                parser.add_argument(
+                    "-o", "--output", default="str", choices=["str", "json"]
+                )
+                parsed_args, _ = parser.parse_known_args(args)
+                output = parsed_args.output
+            else:
+                output = "json"
             try:
-                json_output = json.dumps(result, cls=NumpyJsonEncoder)
-                responses[i] = SimpleResponse(
-                    200, (("Content-Type", "application/json"),), json_output
+                if output == "json":
+                    json_str = json.dumps(json_obj, cls=NumpyJsonEncoder)
+                else:
+                    json_str = str(json_obj)
+                results.append(
+                    InferenceResult(
+                        data=json_str,
+                        context=InferenceContext(
+                            http_status=200,
+                            http_headers=(("Content-Type", "application/json"),),
+                        ),
+                    )
                 )
             except AssertionError as e:
-                responses[i] = SimpleResponse(400, None, str(e))
+                results.append(
+                    InferenceResult(
+                        context=DefaultErrorContext(err_msg=str(e), http_status=400,),
+                    )
+                )
             except Exception as e:  # pylint: disable=broad-except
-                responses[i] = SimpleResponse(500, None, str(e))
-        return responses
+                results.append(
+                    InferenceResult(
+                        context=DefaultErrorContext(err_msg=str(e), http_status=500,),
+                    )
+                )
+        return tuple(results)
 
-    def to_cli(self, result, args):
+    def to_http_response(
+        self, results: Iterable[InferenceResult],
+    ) -> Iterator[HTTPResponse]:
+        return (
+            HTTPResponse(r.context.http_status, r.context.http_headers, r.data)
+            for r in results
+        )
+
+    def to_cli(self, results: Iterable[InferenceResult]) -> int:
         """Handles an CLI command call, convert CLI arguments into
         corresponding data format that user API function is expecting, and
         prints the API function result to console output
 
         :param args: CLI arguments
         """
-        parser = argparse.ArgumentParser()
-        parser.add_argument("-o", "--output", default="str", choices=["str", "json"])
-        parsed_args = parser.parse_args(args)
-
-        if parsed_args.output == 'json':
-            result = json.dumps(result, cls=NumpyJsonEncoder)
-        else:
-            result = str(result)
-        print(result)
-
-    def to_aws_lambda_event(self, result, event):
+        flag = 0
+        for result in results:
+            if result.context.err_msg:
+                print(result.context.err_msg)
+                flag = 1
+            else:
+                print(result.data)
+        return flag
 
-        result = json.dumps(result, cls=NumpyJsonEncoder)
+    def to_aws_lambda_event(self, results: Iterable[InferenceResult]):
 
         # Allow disabling CORS by setting it to None
-        if self.cors:
-            return {
-                "statusCode": 200,
-                "body": result,
-                "headers": {"Access-Control-Allow-Origin": self.cors},
-            }
-
-        return {"statusCode": 200, "body": result}
+        for result in results:
+            if self.cors:
+                yield {
+                    "statusCode": result.context.http_status,
+                    "body": result.data,
+                    "headers": {"Access-Control-Allow-Origin": self.cors},
+                }
+
+            yield {"statusCode": result.context.http_status, "body": result.data}
diff --git a/bentoml/adapters/legacy_image_input.py b/bentoml/adapters/legacy_image_input.py
index 5f2abca6712..888d244f1e1 100644
--- a/bentoml/adapters/legacy_image_input.py
+++ b/bentoml/adapters/legacy_image_input.py
@@ -138,7 +138,7 @@ def pip_dependencies(self):
     def handle_batch_request(self, requests, func):
         raise NotImplementedError
 
-    def handle_request(self, request, func):
+    def handle_request(self, request):
         """Handle http request that has image file/s. It will convert image into a
         ndarray for the function to consume.
 
diff --git a/bentoml/adapters/legacy_json_input.py b/bentoml/adapters/legacy_json_input.py
index 897807a6772..a69a5a7b4b0 100644
--- a/bentoml/adapters/legacy_json_input.py
+++ b/bentoml/adapters/legacy_json_input.py
@@ -20,7 +20,7 @@
 import flask
 
 from bentoml.exceptions import BadInput
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml.adapters.base_input import BaseInputAdapter
 
 
@@ -37,7 +37,7 @@ def __init__(self, is_batch_input=False, **base_kwargs):
             is_batch_input=is_batch_input, **base_kwargs
         )
 
-    def handle_request(self, request: flask.Request, func):
+    def handle_request(self, request: flask.Request):
         if request.content_type.lower() == "application/json":
             parsed_json = json.loads(request.get_data(as_text=True))
         else:
@@ -50,8 +50,8 @@ def handle_request(self, request: flask.Request, func):
         return self.output_adapter.to_response(result, request)
 
     def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func
-    ) -> Iterable[SimpleResponse]:
+        self, requests: Iterable[HTTPRequest], func
+    ) -> Iterable[HTTPResponse]:
         raise NotImplementedError("Use JsonInput instead to enable micro batching")
 
     def handle_cli(self, args, func):
diff --git a/bentoml/adapters/multi_image_input.py b/bentoml/adapters/multi_image_input.py
index 2171a2cb2fc..d8064b355b3 100644
--- a/bentoml/adapters/multi_image_input.py
+++ b/bentoml/adapters/multi_image_input.py
@@ -26,7 +26,7 @@
     verify_image_format_or_raise,
 )
 from bentoml.exceptions import BadInput
-from bentoml.marshal.utils import SimpleRequest, SimpleResponse
+from bentoml.types import HTTPRequest, HTTPResponse
 
 
 # BentoML optional dependencies, using lazy load to avoid ImportError
@@ -98,7 +98,7 @@ def __init__(
             accepted_image_formats or get_default_accept_image_formats()
         )
 
-    def handle_request(self, request: Request, func):
+    def handle_request(self, request: Request):
         files = {
             name: self.read_file(file.filename, file.stream)
             for (name, file) in request.files.items()
@@ -112,8 +112,8 @@ def read_file(self, name: str, file: BytesIO):
         return imageio.imread(file, pilmode=self.pilmode)
 
     def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func
-    ) -> Iterable[SimpleResponse]:
+        self, requests: Iterable[HTTPRequest], func
+    ) -> Iterable[HTTPResponse]:
         inputs = []
         slices = []
         for i, req in enumerate(requests):
@@ -126,7 +126,7 @@ def handle_batch_request(
             else:
                 files = {}
                 request = Request.from_values(
-                    data=req.data, content_type=content_type, headers=req.headers,
+                    data=req.body, content_type=content_type, headers=req.headers,
                 )
                 for name in request.files:
                     file = request.files[name]
diff --git a/bentoml/adapters/pytorch_tensor_input.py b/bentoml/adapters/pytorch_tensor_input.py
index 61b1a7cb1d6..f37f0b0d00b 100644
--- a/bentoml/adapters/pytorch_tensor_input.py
+++ b/bentoml/adapters/pytorch_tensor_input.py
@@ -20,7 +20,7 @@ class PytorchTensorInput(BaseInputAdapter):
     Tensor input adapter for Pytorch models
     """
 
-    def handle_request(self, request, func):
+    def handle_request(self, request):
         raise NotImplementedError
 
     def handle_cli(self, args, func):
diff --git a/bentoml/adapters/tensorflow_tensor_input.py b/bentoml/adapters/tensorflow_tensor_input.py
index e9d3c5da1e5..c63430bb760 100644
--- a/bentoml/adapters/tensorflow_tensor_input.py
+++ b/bentoml/adapters/tensorflow_tensor_input.py
@@ -18,10 +18,10 @@
 
 import argparse
 
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml.adapters.utils import concat_list, TF_B64_KEY
 from bentoml.adapters.base_input import BaseInputAdapter
 from bentoml.exceptions import BentoMLException
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
 
 
 def b64_hook(o):
@@ -96,22 +96,22 @@ def _handle_raw_str(self, raw_str, func):
         return func(parsed_tensor)
 
     def handle_batch_request(
-        self, requests: Iterable[SimpleRequest], func
-    ) -> Iterable[SimpleResponse]:
+        self, requests: Iterable[HTTPRequest], func
+    ) -> Iterable[HTTPResponse]:
         """
         TODO(bojiang):
         1. specify batch dim
         """
         import tensorflow as tf
 
-        bad_resp = SimpleResponse(400, None, "input format error")
+        bad_resp = HTTPResponse(400, body="input format error")
         instances_list = [None] * len(requests)
         responses = [bad_resp] * len(requests)
         batch_flags = [None] * len(requests)
 
         for i, request in enumerate(requests):
             try:
-                raw_str = request.data
+                raw_str = request.body
                 batch_flags[i] = self.is_batch_request(request)
                 parsed_json = json.loads(raw_str, object_hook=b64_hook)
                 if parsed_json.get("instances") is not None:
@@ -121,8 +121,8 @@ def handle_batch_request(
                     instances_list[i] = instances
 
                 elif parsed_json.get("inputs"):
-                    responses[i] = SimpleResponse(
-                        501, None, "Column format 'inputs' not implemented"
+                    responses[i] = HTTPResponse(
+                        501, body="Column format 'inputs' not implemented"
                     )
 
             except (json.JSONDecodeError, UnicodeDecodeError):
@@ -131,17 +131,15 @@ def handle_batch_request(
                 import traceback
 
                 err = traceback.format_exc()
-                responses[i] = SimpleResponse(
-                    500, None, f"Internal Server Error: {err}"
-                )
+                responses[i] = HTTPResponse(500, body=f"Internal Server Error: {err}")
         merged_instances, slices = concat_list(instances_list, batch_flags=batch_flags)
         parsed_tensor = tf.constant(merged_instances)
         merged_result = func(parsed_tensor)
-        return self.output_adapter.to_batch_response(
+        return self.output_adapter.to_http_response(
             merged_result, slices=slices, fallbacks=responses, requests=requests
         )
 
-    def handle_request(self, request, func):
+    def handle_request(self, request):
         """Handle http request that has jsonlized tensorflow tensor. It will convert it
         into a tf tensor for the function to consume.
 
@@ -151,7 +149,7 @@ def handle_request(self, request, func):
         Return:
             response object
         """
-        req = SimpleRequest.from_flask_request(request)
+        req = HTTPRequest.from_flask_request(request)
         res = self.handle_batch_request([req], func)[0]
         return res.to_flask_response()
 
diff --git a/bentoml/adapters/tensorflow_tensor_output.py b/bentoml/adapters/tensorflow_tensor_output.py
index 6c171fa4210..a361e4ac391 100644
--- a/bentoml/adapters/tensorflow_tensor_output.py
+++ b/bentoml/adapters/tensorflow_tensor_output.py
@@ -18,8 +18,8 @@
 
 import argparse
 
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml.utils.lazy_loader import LazyLoader
-from bentoml.marshal.utils import SimpleResponse, SimpleRequest
 from bentoml.adapters.utils import TfTensorJsonEncoder
 from bentoml.adapters.base_output import BaseOutputAdapter
 
@@ -61,8 +61,8 @@ def to_batch_response(
         result_conc,
         slices=None,
         fallbacks=None,
-        requests: Iterable[SimpleRequest] = None,
-    ) -> Iterable[SimpleResponse]:
+        requests: Iterable[HTTPRequest] = None,
+    ) -> Iterable[HTTPResponse]:
         # TODO(bojiang): header content_type
         results = tf_to_numpy(result_conc)
         assert isinstance(results, np.ndarray)
@@ -78,7 +78,7 @@ def to_batch_response(
                 continue
             result = results[s]
             result_str = json.dumps(result, cls=TfTensorJsonEncoder)
-            responses[i] = SimpleResponse(200, dict(), result_str)
+            responses[i] = HTTPResponse(status=200, body=result_str)
         return responses
 
     def to_cli(self, result, args):
diff --git a/bentoml/artifact/artifact.py b/bentoml/artifact/artifact.py
index d45ec20c8b2..1fa32a2782b 100644
--- a/bentoml/artifact/artifact.py
+++ b/bentoml/artifact/artifact.py
@@ -30,7 +30,7 @@ class BentoServiceArtifact:
     serialization and deserialization process.
     """
 
-    def __init__(self, name):
+    def __init__(self, name: str):
         if not name.isidentifier():
             raise ValueError(
                 "Artifact name must be a valid python identifier, a string \
@@ -207,7 +207,7 @@ def __getattr__(self, item: str):
         artifact in its native form. E.g. for a BentoService class with an artifact
         `SklearnModelArtifact('model')`, when user code accesses `self.artifacts.model`
         in the BentoService API callback function, instead of returning an instance of
-        BentoServcieArtifact, it returns a Sklearn model object
+        BentoServiceArtifact, it returns a Sklearn model object
         """
         return self[item].get()
 
diff --git a/bentoml/artifact/fasttext_model_artifact.py b/bentoml/artifact/fasttext_model_artifact.py
index eaf7e7b883f..1594476ea41 100644
--- a/bentoml/artifact/fasttext_model_artifact.py
+++ b/bentoml/artifact/fasttext_model_artifact.py
@@ -33,7 +33,7 @@ class FasttextModelArtifact(BentoServiceArtifact):
 
     >>> import fasttext
     >>> # prepare training data and store to file
-    >>> training_data_file = 'trainging-data-file.train'
+    >>> training_data_file = 'training-data-file.train'
     >>> model = fasttext.train_supervised(input=training_data_file)
     >>>
     >>> import bentoml
diff --git a/bentoml/artifact/json_artifact.py b/bentoml/artifact/json_artifact.py
index 027c7806bf9..7c0f5a6c880 100644
--- a/bentoml/artifact/json_artifact.py
+++ b/bentoml/artifact/json_artifact.py
@@ -24,7 +24,7 @@ class JSONArtifact(BentoServiceArtifact):
 
     Args:
         name (str): Name of the artifact
-        file_extension (:obj:`str`, optional): The file extention used for the saved
+        file_extension (:obj:`str`, optional): The file extension used for the saved
             text file. Defaults to ".txt"
         encoding (:obj:`str`, optional): The encoding will be used for saving/loading
             text. Defaults to "utf8"
diff --git a/bentoml/artifact/keras_model_artifact.py b/bentoml/artifact/keras_model_artifact.py
index 222bfffebd5..f516930b734 100644
--- a/bentoml/artifact/keras_model_artifact.py
+++ b/bentoml/artifact/keras_model_artifact.py
@@ -51,7 +51,7 @@ class KerasModelArtifact(BentoServiceArtifact):
     >>> from tensorflow.keras.preprocessing import sequence, text
     >>>
     >>> model_to_save = Sequential()
-    >>> # traing model
+    >>> # training model
     >>> model_to_save.compile(...)
     >>> model_to_save.fit(...)
     >>>
diff --git a/bentoml/artifact/onnx_model_artifact.py b/bentoml/artifact/onnx_model_artifact.py
index 11459d63a0e..74e77ba8d67 100644
--- a/bentoml/artifact/onnx_model_artifact.py
+++ b/bentoml/artifact/onnx_model_artifact.py
@@ -140,7 +140,7 @@ def pack(self, path_or_model_proto):  # pylint:disable=arguments-differ
                 )
 
         assert self._onnx_model_path or self._model_proto, (
-            "Either self._onnx_model_path or self._model_proto has to be initilaized "
+            "Either self._onnx_model_path or self._model_proto has to be initialized "
             "after initializing _OnnxModelArtifactWrapper"
         )
 
@@ -159,7 +159,7 @@ def _get_onnx_inference_session(self):
                 import onnxruntime
             except ImportError:
                 raise MissingDependencyException(
-                    '"onnxruntime" package is required for inferencing with onnx '
+                    '"onnxruntime" package is required for inference with onnx '
                     'runtime as backend'
                 )
 
diff --git a/bentoml/artifact/text_file_artifact.py b/bentoml/artifact/text_file_artifact.py
index 9e7ae882e21..d8161af1e48 100644
--- a/bentoml/artifact/text_file_artifact.py
+++ b/bentoml/artifact/text_file_artifact.py
@@ -23,7 +23,7 @@ class TextFileArtifact(BentoServiceArtifact):
 
     Args:
         name (str): Name of the artifact
-        file_extension (:obj:`str`, optional): The file extention used for the saved
+        file_extension (:obj:`str`, optional): The file extension used for the saved
             text file. Defaults to ".txt"
         encoding (str): The encoding will be used for saving/loading text. Defaults
             to "utf8"
diff --git a/bentoml/cli/bento_management.py b/bentoml/cli/bento_management.py
index 6a45ede2628..b3637e76ce7 100644
--- a/bentoml/cli/bento_management.py
+++ b/bentoml/cli/bento_management.py
@@ -20,7 +20,7 @@
     _echo,
     parse_bento_tag_list_callback,
 )
-from bentoml.cli.utils import humanfriendly_age_from_datetime, get_default_yatai_client
+from bentoml.cli.utils import human_friendly_age_from_datetime, get_default_yatai_client
 from bentoml.utils import pb_to_yaml, status_pb_to_error_code_and_message
 from bentoml.saved_bundle import safe_retrieve
 from bentoml.exceptions import CLIException
@@ -59,7 +59,7 @@ def _print_bento_table(bentos, wide=False):
                 "%Y-%m-%d %H:%M"
             )
         else:
-            created_at = humanfriendly_age_from_datetime(
+            created_at = human_friendly_age_from_datetime(
                 bento.bento_service_metadata.created_at.ToDatetime()
             )
         row = [
diff --git a/bentoml/cli/bento_service.py b/bentoml/cli/bento_service.py
index 740c188511a..da373c007fb 100644
--- a/bentoml/cli/bento_service.py
+++ b/bentoml/cli/bento_service.py
@@ -1,4 +1,5 @@
 import click
+import sys
 import os
 import json
 import re
@@ -173,7 +174,8 @@ def run(api_name, run_args, bento=None):
         saved_bundle_path = resolve_bundle_path(bento, pip_installed_bundle_path)
 
         api = load_bento_service_api(saved_bundle_path, api_name)
-        api.handle_cli(run_args)
+        exit_code = api.handle_cli(run_args)
+        sys.exit(exit_code)
 
     # Example Usage: bentoml info {BUNDLE_PATH}
     @bentoml_cli.command(
@@ -214,9 +216,9 @@ def open_api_spec(bento=None):
     @click.option(
         "--port",
         type=click.INT,
-        default=BentoAPIServer._DEFAULT_PORT,
+        default=BentoAPIServer.DEFAULT_PORT,
         help=f"The port to listen on for the REST api server, "
-        f"default is {BentoAPIServer._DEFAULT_PORT}",
+        f"default is {BentoAPIServer.DEFAULT_PORT}",
         envvar='BENTOML_PORT',
     )
     @click.option(
@@ -248,9 +250,9 @@ def serve(port, bento=None, enable_microbatch=False, run_with_ngrok=False):
         "-p",
         "--port",
         type=click.INT,
-        default=BentoAPIServer._DEFAULT_PORT,
+        default=BentoAPIServer.DEFAULT_PORT,
         help=f"The port to listen on for the REST api server, "
-        f"default is {BentoAPIServer._DEFAULT_PORT}",
+        f"default is {BentoAPIServer.DEFAULT_PORT}",
         envvar='BENTOML_PORT',
     )
     @click.option(
@@ -285,7 +287,7 @@ def serve_gunicorn(
     ):
         if not psutil.POSIX:
             _echo(
-                "The `bentoml serve-gunicon` command is only supported on POSIX. "
+                "The `bentoml serve-gunicorn` command is only supported on POSIX. "
                 "On windows platform, use `bentoml serve` for local API testing and "
                 "docker for running production API endpoint: "
                 "https://docs.docker.com/docker-for-windows/ "
diff --git a/bentoml/cli/click_utils.py b/bentoml/cli/click_utils.py
index 0706662a516..9716036e7f4 100644
--- a/bentoml/cli/click_utils.py
+++ b/bentoml/cli/click_utils.py
@@ -27,7 +27,6 @@
 from bentoml.utils.usage_stats import track
 from bentoml.configuration import set_debug_mode
 
-
 # Available CLI colors for _echo:
 #
 # _ansi_colors = {
@@ -54,7 +53,6 @@
 CLI_COLOR_ERROR = "red"
 CLI_COLOR_WARNING = "yellow"
 
-
 logger = logging.getLogger(__name__)
 TRACK_CLI_EVENT_NAME = 'bentoml-cli'
 
@@ -155,8 +153,8 @@ def wrapper(func):
 
             # move common parameters to end of the parameters list
             func.__click_params__ = (
-                func.__click_params__[-self.NUMBER_OF_COMMON_PARAMS :]
-                + func.__click_params__[: -self.NUMBER_OF_COMMON_PARAMS]
+                    func.__click_params__[-self.NUMBER_OF_COMMON_PARAMS:]
+                    + func.__click_params__[: -self.NUMBER_OF_COMMON_PARAMS]
             )
             return super(BentoMLCommandGroup, self).command(*args, **kwargs)(func)
 
@@ -221,7 +219,7 @@ def parse_labels_callback(ctx, param, value):  # pylint: disable=unused-argument
 
 
 def validate_labels_query_callback(
-    ctx, param, value
+        ctx, param, value
 ):  # pylint: disable=unused-argument
     if not value:
         return value
diff --git a/bentoml/cli/deployment.py b/bentoml/cli/deployment.py
index 23c624a07a0..d1d42b12bc8 100644
--- a/bentoml/cli/deployment.py
+++ b/bentoml/cli/deployment.py
@@ -48,7 +48,7 @@ def get_deployment_sub_command():
     # pylint: disable=unused-variable
 
     @click.group(
-        help='Commands for manageing and operating BentoService deployments',
+        help='Commands for managing and operating BentoService deployments',
         cls=BentoMLCommandGroup,
     )
     def deployment():
diff --git a/bentoml/cli/utils.py b/bentoml/cli/utils.py
index c38ebb8c5a9..2fac93daa2f 100644
--- a/bentoml/cli/utils.py
+++ b/bentoml/cli/utils.py
@@ -106,7 +106,7 @@ def echo_docker_api_result(docker_generator):
                 humanfriendly.format_size(sum(cur)),
                 humanfriendly.format_size(sum(total)),
             )
-            yield (f"Pushed {cur} / {total}")
+            yield f"Pushed {cur} / {total}"
         if "errorDetail" in line:
             error = line["errorDetail"]
             raise BentoMLException(error["message"])
@@ -145,10 +145,10 @@ def _format_deployment_age_for_print(deployment_pb):
         # we will not show the age for those deployments
         return None
     else:
-        return humanfriendly_age_from_datetime(deployment_pb.created_at.ToDatetime())
+        return human_friendly_age_from_datetime(deployment_pb.created_at.ToDatetime())
 
 
-def humanfriendly_age_from_datetime(dt, detailed=False, max_unit=2):
+def human_friendly_age_from_datetime(dt, detailed=False, max_unit=2):
     return humanfriendly.format_timespan(datetime.utcnow() - dt, detailed, max_unit)
 
 
diff --git a/bentoml/clipper/__init__.py b/bentoml/clipper/__init__.py
index e4d1c8bdff5..b2202cb1208 100644
--- a/bentoml/clipper/__init__.py
+++ b/bentoml/clipper/__init__.py
@@ -71,7 +71,7 @@ def predict_strings(self, inputs):
 
 
 if __name__ == "__main__":
-    print("Starting BentoService Clipper Containter")
+    print("Starting BentoService Clipper Container")
     rpc_service = rpc.RPCService()
 
     try:
@@ -110,8 +110,8 @@ def predict_strings(self, inputs):
 """  # noqa: E501
 
 
-def get_clipper_compatiable_string(item):
-    """Generate clipper compatiable string. It must be a valid DNS-1123.
+def get_clipper_compatible_string(item):
+    """Generate clipper compatible string. It must be a valid DNS-1123.
     It must consist of lower case alphanumeric characters, '-' or '.',
     and must start and end with an alphanumeric character
 
@@ -150,7 +150,7 @@ def deploy_bentoml(
     if not clipper_conn.connected:
         raise BentoMLException(
             "No connection to Clipper cluster. CallClipperConnection.connect to "
-            "connect to an existing cluster or ClipperConnnection.start_clipper to "
+            "connect to an existing cluster or ClipperConnection.start_clipper to "
             "create a new one"
         )
 
@@ -173,10 +173,10 @@ def deploy_bentoml(
         )
 
     input_type = ADAPTER_TYPE_TO_INPUT_TYPE[api_metadata.input_type]
-    model_name = model_name or get_clipper_compatiable_string(
+    model_name = model_name or get_clipper_compatible_string(
         bento_service_metadata.name + "-" + api_metadata.name
     )
-    model_version = get_clipper_compatiable_string(bento_service_metadata.version)
+    model_version = get_clipper_compatible_string(bento_service_metadata.version)
 
     with TempDirectory() as tempdir:
         entry_py_content = CLIPPER_ENTRY.format(api_name=api_name)
diff --git a/bentoml/configuration/__init__.py b/bentoml/configuration/__init__.py
index 7251c206fc1..92fbedb112d 100644
--- a/bentoml/configuration/__init__.py
+++ b/bentoml/configuration/__init__.py
@@ -175,7 +175,7 @@ def get_bentoml_deploy_version():
         if _is_pip_installed_bentoml():
             logger.warning(
                 "Using BentoML not from official PyPI release. In order to find the "
-                "same version of BentoML when deplying your BentoService, you must "
+                "same version of BentoML when deploying your BentoService, you must "
                 "set the 'core/bentoml_deploy_version' config to a http/git location "
                 "of your BentoML fork, e.g.: 'bentoml_deploy_version = "
                 "git+https://github.com/{username}/bentoml.git@{branch}'"
diff --git a/bentoml/configuration/configparser.py b/bentoml/configuration/configparser.py
index d522e945c20..fe700439172 100644
--- a/bentoml/configuration/configparser.py
+++ b/bentoml/configuration/configparser.py
@@ -40,7 +40,7 @@ def _env_var_name(section, key):
         return "BENTOML__{}__{}".format(section.upper(), key.upper())
 
     def get(self, section, key=None, **kwargs):  # pylint:disable=arguments-differ
-        """ A simple hierachical config access, priority order:
+        """ A simple hierarchical config access, priority order:
             1. environment var
             2. user config file
             3. bentoml default config file
diff --git a/bentoml/exceptions.py b/bentoml/exceptions.py
index 47b45efbd8b..e0f51cf0260 100644
--- a/bentoml/exceptions.py
+++ b/bentoml/exceptions.py
@@ -54,7 +54,7 @@ def status_code(self):
 
 class RemoteException(BentoMLException):
     """
-    Raise when known exceptions happend in remote server(a model server normally)
+    Raise when known exceptions happened in remote server(a model server normally)
     """
 
     def __init__(self, *args, payload, **kwargs):
@@ -142,7 +142,7 @@ class YataiDeploymentException(YataiServiceException):
 
 
 class YataiRepositoryException(YataiServiceException):
-    """Raise when YataiService encounters an issue managing BentoService repoistory"""
+    """Raise when YataiService encounters an issue managing BentoService repository"""
 
 
 class AWSServiceError(YataiDeploymentException):
diff --git a/bentoml/marshal/dispatcher.py b/bentoml/marshal/dispatcher.py
index 32009594998..5232aac80ce 100644
--- a/bentoml/marshal/dispatcher.py
+++ b/bentoml/marshal/dispatcher.py
@@ -18,7 +18,7 @@ class NonBlockSema:
     def __init__(self, count):
         self.sema = count
 
-    def aquire(self):
+    def acquire(self):
         if self.sema < 1:
             return False
         self.sema -= 1
@@ -176,7 +176,7 @@ async def controller(self):
 
                 n_call_out = min(self.max_batch_size, n,)
                 # call
-                self._sema.aquire()
+                self._sema.acquire()
                 inputs_info = tuple(self._queue.pop() for _ in range(n_call_out))
                 self._loop.create_task(self.outbound_call(inputs_info))
             except asyncio.CancelledError:
diff --git a/bentoml/marshal/marshal.py b/bentoml/marshal/marshal.py
index 041f68e3d09..0412dff218a 100644
--- a/bentoml/marshal/marshal.py
+++ b/bentoml/marshal/marshal.py
@@ -24,11 +24,11 @@
 
 from bentoml import config
 from bentoml.exceptions import RemoteException
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml.server.trace import async_trace, make_http_headers
-from bentoml.marshal.utils import DataLoader, SimpleRequest
+from bentoml.marshal.utils import DataLoader
 from bentoml.saved_bundle import load_bento_service_metadata
 from bentoml.marshal.dispatcher import CorkDispatcher, NonBlockSema
-from bentoml.marshal.utils import SimpleResponse
 
 logger = logging.getLogger(__name__)
 ZIPKIN_API_URL = config("tracing").get("zipkin_api_url")
@@ -59,19 +59,19 @@ def __init__(self, *args, **kwargs):
             )
             self.metrics_request_in_progress = Gauge(
                 name=service_name + "_mb_request_in_progress",
-                documentation='Totoal number of HTTP requests in progress now',
+                documentation='Total number of HTTP requests in progress now',
                 namespace=namespace,
                 labelnames=['endpoint', 'http_method'],
             )
             self.metrics_request_exception = Counter(
                 name=service_name + "_mb_request_exception",
-                documentation='Totoal number of service exceptions',
+                documentation='Total number of service exceptions',
                 namespace=namespace,
                 labelnames=['endpoint', 'exception_class'],
             )
             self.metrics_request_total = Counter(
                 name=service_name + "_mb_request_total",
-                documentation='Totoal number of service exceptions',
+                documentation='Total number of service exceptions',
                 namespace=namespace,
                 labelnames=['endpoint', 'http_response_code'],
             )
@@ -203,7 +203,7 @@ async def request_dispatcher(self, request):
         ):
             api_name = request.match_info.get("name")
             if api_name in self.batch_handlers:
-                req = SimpleRequest(request.raw_headers, await request.read())
+                req = HTTPRequest(request.raw_headers, await request.read())
                 try:
                     resp = await self.batch_handlers[api_name](req)
                 except RemoteException as e:
@@ -269,16 +269,16 @@ async def _batch_handler_template(self, requests, api_name):
                         raw = await resp.read()
             except aiohttp.client_exceptions.ClientConnectionError as e:
                 raise RemoteException(
-                    e, payload=SimpleResponse(503, None, b"Service Unavailable")
+                    e, payload=HTTPResponse(status=503, body=b"Service Unavailable")
                 )
             if resp.status != 200:
                 raise RemoteException(
                     f"Bad response status from model server:\n{resp.status}\n{raw}",
-                    payload=SimpleResponse(resp.status, resp.headers, raw),
+                    payload=HTTPResponse(resp.status, resp.headers, raw),
                 )
             merged = DataLoader.split_responses(raw)
             return tuple(
-                aiohttp.web.Response(body=i.data, headers=i.headers, status=i.status)
+                aiohttp.web.Response(body=i.body, headers=i.headers, status=i.status)
                 for i in merged
             )
 
diff --git a/bentoml/marshal/utils.py b/bentoml/marshal/utils.py
index 49d2f7aaa38..4e53222b700 100644
--- a/bentoml/marshal/utils.py
+++ b/bentoml/marshal/utils.py
@@ -4,51 +4,12 @@
 
 from multidict import CIMultiDict
 
+from bentoml.types import HTTPRequest, HTTPResponse
 from bentoml import config as bentoml_config
 
 BATCH_REQUEST_HEADER = bentoml_config("apiserver").get("batch_request_header")
 
 
-class SimpleRequest(NamedTuple):
-    '''
-    headers: tuple of key value pairs in bytes
-    data: str
-    '''
-
-    headers: tuple
-    data: str
-
-    @property
-    @lru_cache()
-    def parsed_headers(self):
-        return CIMultiDict(
-            (hk.decode("latin1").lower(), hv.decode("latin1"))
-            for hk, hv in self.headers or tuple()
-        )
-
-    @classmethod
-    def from_flask_request(cls, request):
-        # For non latin1 characters: https://tools.ietf.org/html/rfc8187
-        # Also https://github.com/benoitc/gunicorn/issues/1778
-        return cls(
-            tuple((k.encode("latin1"), v.encode("latin1")) for k, v in request.headers),
-            request.get_data(),
-        )
-
-
-class SimpleResponse(NamedTuple):
-    status: int
-    headers: tuple
-    data: str
-
-    def to_flask_response(self):
-        import flask
-
-        return flask.Response(
-            headers=self.headers, response=self.data, status=self.status
-        )
-
-
 class PlasmaDataLoader:
     """
     Transfer datas with plasma store, in development now
@@ -100,19 +61,19 @@ def merge_responses(cls, resps) -> bytes:
 
 class PickleDataLoader:
     @classmethod
-    def merge_requests(cls, reqs: Iterable[SimpleRequest]) -> bytes:
+    def merge_requests(cls, reqs: Iterable[HTTPRequest]) -> bytes:
         return pickle.dumps(reqs)
 
     @classmethod
-    def split_requests(cls, raw: bytes) -> Iterable[SimpleRequest]:
+    def split_requests(cls, raw: bytes) -> Iterable[HTTPRequest]:
         return pickle.loads(raw)
 
     @classmethod
-    def merge_responses(cls, resps: Iterable[SimpleResponse]) -> bytes:
-        return pickle.dumps(resps)
+    def merge_responses(cls, resps: Iterable[HTTPResponse]) -> bytes:
+        return pickle.dumps(list(resps))
 
     @classmethod
-    def split_responses(cls, raw: bytes) -> Iterable[SimpleResponse]:
+    def split_responses(cls, raw: bytes) -> Iterable[HTTPResponse]:
         try:
             return pickle.loads(raw)
         except pickle.UnpicklingError:
diff --git a/bentoml/saved_bundle/config.py b/bentoml/saved_bundle/config.py
index 51c62a30b2d..00d87a2e789 100644
--- a/bentoml/saved_bundle/config.py
+++ b/bentoml/saved_bundle/config.py
@@ -26,7 +26,7 @@
 from bentoml.utils import dump_to_yaml_str
 from bentoml.exceptions import BentoMLConfigException
 
-BENTOML_CONFIG_YAML_TEPMLATE = """\
+BENTOML_CONFIG_YAML_TEMPLATE = """\
 version: {bentoml_version}
 kind: {kind}
 metadata:
@@ -45,13 +45,13 @@ def _get_apis_list(bento_service):
         api_obj = {
             "name": api.name,
             "docs": api.doc,
-            "input_type": api.handler.__class__.__name__,
-            "output_type": api.handler.output_adapter.__class__.__name__,
+            "input_type": api.input_adapter.__class__.__name__,
+            "output_type": api.output_adapter.__class__.__name__,
             "mb_max_batch_size": api.mb_max_batch_size,
             "mb_max_latency": api.mb_max_latency,
         }
-        if api.handler.config:
-            api_obj["input_config"] = api.handler.config
+        if api.input_adapter.config:
+            api_obj["input_config"] = api.input_adapter.config
         if api.output_adapter.config:
             api_obj["output_config"] = api.output_adapter.config
         result.append(api_obj)
@@ -73,7 +73,7 @@ def __init__(self, bento_service=None, kind="BentoService"):
         self._yaml = YAML()
         self._yaml.default_flow_style = False
         self.config = self._yaml.load(
-            BENTOML_CONFIG_YAML_TEPMLATE.format(
+            BENTOML_CONFIG_YAML_TEMPLATE.format(
                 kind=self.kind,
                 bentoml_version=get_bentoml_deploy_version(),
                 created_at=str(datetime.utcnow()),
diff --git a/bentoml/server/api_server.py b/bentoml/server/api_server.py
index 988862499f9..b6168dbeeb7 100644
--- a/bentoml/server/api_server.py
+++ b/bentoml/server/api_server.py
@@ -23,13 +23,13 @@
 from werkzeug.utils import secure_filename
 from werkzeug.exceptions import BadRequest, NotFound
 
-from bentoml import config
+from bentoml import config, BentoService
 from bentoml.configuration import get_debug_mode
 from bentoml.exceptions import BentoMLException
 from bentoml.server.instruments import InstrumentMiddleware
 from bentoml.server.open_api import get_open_api_spec_json
 from bentoml.server import trace
-
+from bentoml.service import InferenceAPI
 
 CONTENT_TYPE_LATEST = str("text/plain; version=0.0.4; charset=utf-8")
 
@@ -65,6 +65,18 @@ def _request_to_json(req):
     return {}
 
 
+def log_exception(exc_info):
+    """
+    Logs an exception.  This is called by :meth:`handle_exception`
+    if debugging is disabled and right before the handler is called.
+    The default implementation logs the exception as error on the
+    :attr:`logger`.
+    """
+    logger.error(
+        "Exception on %s [%s]", request.path, request.method, exc_info=exc_info
+    )
+
+
 class BentoAPIServer:
     """
     BentoAPIServer creates a REST API server based on APIs defined with a BentoService
@@ -74,10 +86,10 @@ class BentoAPIServer:
     request data into a Service API function
     """
 
-    _DEFAULT_PORT = config("apiserver").getint("default_port")
+    DEFAULT_PORT = config("apiserver").getint("default_port")
     _MARSHAL_FLAG = config("marshal_server").get("marshal_request_header_flag")
 
-    def __init__(self, bento_service, port=_DEFAULT_PORT, app_name=None):
+    def __init__(self, bento_service: BentoService, port=DEFAULT_PORT, app_name=None):
         app_name = bento_service.name if app_name is None else app_name
 
         self.port = port
@@ -155,6 +167,7 @@ def healthz_view_func():
         return Response(response="\n", status=200, mimetype="text/plain")
 
     def metrics_view_func(self):
+        # noinspection PyProtectedMember
         from prometheus_client import generate_latest
 
         return generate_latest()
@@ -249,47 +262,10 @@ def setup_bento_service_api_routes(self):
                 rule="/{}".format(api.name),
                 endpoint=api.name,
                 view_func=route_function,
-                methods=api.handler.HTTP_METHODS,
-            )
-
-    @staticmethod
-    def log_image(req, request_id):
-        if not config('logging').getboolean('log_request_image_files'):
-            return []
-
-        img_prefix = 'image/'
-        log_folder = config('logging').get('base_log_dir')
-
-        all_paths = []
-
-        if req.content_type and req.content_type.startswith(img_prefix):
-            filename = '{timestamp}-{request_id}.{ext}'.format(
-                timestamp=int(time.time()),
-                request_id=request_id,
-                ext=req.content_type[len(img_prefix) :],
+                methods=api.input_adapter.HTTP_METHODS,
             )
-            path = os.path.join(log_folder, filename)
-            all_paths.append(path)
-            with open(path, 'wb') as f:
-                f.write(req.get_data())
-
-        for name in req.files:
-            file = req.files[name]
-            if file and file.filename:
-                orig_filename = secure_filename(file.filename)
-                filename = '{timestamp}-{request_id}-{orig_filename}'.format(
-                    timestamp=int(time.time()),
-                    request_id=request_id,
-                    orig_filename=orig_filename,
-                )
-                path = os.path.join(log_folder, filename)
-                all_paths.append(path)
-                file.save(path)
-                file.stream.seek(0)
-
-        return all_paths
 
-    def bento_service_api_func_wrapper(self, api):
+    def bento_service_api_func_wrapper(self, api: InferenceAPI):
         """
         Create api function for flask route, it wraps around user defined API
         callback and adapter class, and adds request logging and instrument metrics
@@ -299,13 +275,6 @@ def bento_service_api_func_wrapper(self, api):
         service_version = self.bento_service.version
 
         def api_func():
-            # Log image files in request if there is any
-            image_paths = self.log_image(request, request_id)
-
-            # _request_to_json parses request as JSON; in case errors, it raises
-            # a 400 exception. (consider 4xx before 5xx.)
-            request_for_log = _request_to_json(request)
-
             # handle_request may raise 4xx or 5xx exception.
             try:
                 if request.headers.get(self._MARSHAL_FLAG):
@@ -314,7 +283,7 @@ def api_func():
                 else:
                     response = api.handle_request(request)
             except BentoMLException as e:
-                self.log_exception(sys.exc_info())
+                log_exception(sys.exc_info())
 
                 if 400 <= e.status_code < 500 and e.status_code not in (401, 403):
                     response = make_response(
@@ -329,7 +298,7 @@ def api_func():
             except Exception:  # pylint: disable=broad-except
                 # For all unexpected error, return 500 by default. For example,
                 # if users' model raises an error of division by zero.
-                self.log_exception(sys.exc_info())
+                log_exception(sys.exc_info())
 
                 response = make_response(
                     'An error has occurred in BentoML user code when handling this '
@@ -337,23 +306,6 @@ def api_func():
                     500,
                 )
 
-            request_log = {
-                "request_id": request_id,
-                "service_name": service_name,
-                "service_version": service_version,
-                "api": api.name,
-                "request": request_for_log,
-                "response_code": response.status_code,
-            }
-
-            if len(image_paths) > 0:
-                request_log['image_paths'] = image_paths
-
-            if 200 <= response.status_code < 300:
-                request_log['response'] = response.response
-
-            prediction_logger.info(request_log)
-
             response.headers["request_id"] = request_id
 
             return response
@@ -365,13 +317,3 @@ def api_func_with_tracing():
                 return api_func()
 
         return api_func_with_tracing
-
-    def log_exception(self, exc_info):
-        """Logs an exception.  This is called by :meth:`handle_exception`
-        if debugging is disabled and right before the handler is called.
-        The default implementation logs the exception as error on the
-        :attr:`logger`.
-        """
-        logger.error(
-            "Exception on %s [%s]", request.path, request.method, exc_info=exc_info
-        )
diff --git a/bentoml/server/instruments.py b/bentoml/server/instruments.py
index 0f271127a79..d6c6f167818 100644
--- a/bentoml/server/instruments.py
+++ b/bentoml/server/instruments.py
@@ -31,14 +31,14 @@ def __init__(self, app, bento_service):
         )
         self.metrics_request_total = Counter(
             name=service_name + "_request_total",
-            documentation='Totoal number of HTTP requests',
+            documentation='Total number of HTTP requests',
             namespace=namespace,
             labelnames=['endpoint', 'service_version', 'http_response_code'],
             registry=self.collector_registry,
         )
         self.metrics_request_in_progress = Gauge(
             name=service_name + "_request_in_progress",
-            documentation='Totoal number of HTTP requests in progress now',
+            documentation='Total number of HTTP requests in progress now',
             namespace=namespace,
             labelnames=['endpoint', 'service_version'],
             registry=self.collector_registry,
diff --git a/bentoml/service.py b/bentoml/service.py
index b24ce27d998..e45114b2e8a 100644
--- a/bentoml/service.py
+++ b/bentoml/service.py
@@ -6,34 +6,42 @@
 
 # http://www.apache.org/licenses/LICENSE-2.0
 
+
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
-import re
+import argparse
+import functools
 import inspect
+import itertools
 import logging
+import os
+import re
+import sys
 import uuid
 from datetime import datetime
-from typing import List
+from typing import Tuple, List, Iterable
 
 import flask
 from werkzeug.utils import cached_property
 
 from bentoml import config
+from bentoml.adapters import BaseInputAdapter, BaseOutputAdapter, DefaultOutput
+from bentoml.artifact import BentoServiceArtifact, ArtifactCollection
+from bentoml.exceptions import NotFound, InvalidArgument, BentoMLException
 from bentoml.saved_bundle import save_to_dir
 from bentoml.saved_bundle.config import SavedBundleConfig
+from bentoml.server import trace
 from bentoml.service_env import BentoServiceEnv
-from bentoml.utils import isidentifier
+from bentoml.types import (
+    HTTPRequest,
+    InferenceTask,
+    InferenceResult,
+)
 from bentoml.utils.hybridmethod import hybridmethod
-from bentoml.exceptions import NotFound, InvalidArgument, BentoMLException
-from bentoml.server import trace
-from bentoml.artifact import BentoServiceArtifact, ArtifactCollection
-from bentoml.adapters import BaseInputAdapter, BaseOutputAdapter
-
 
 ARTIFACTS_DIR_NAME = "artifacts"
 DEFAULT_MAX_LATENCY = config("marshal_server").getint("default_max_latency")
@@ -57,17 +65,29 @@ class InferenceAPI(object):
     """
 
     def __init__(
-        self, service, name, doc, handler, func, mb_max_latency, mb_max_batch_size
+        self,
+        service,
+        name,
+        doc,
+        input_adapter: BaseInputAdapter,
+        user_func: callable,
+        output_adapter: BaseOutputAdapter,
+        mb_max_latency=10000,
+        mb_max_batch_size=1000,
     ):
         """
         :param service: ref to service containing this API
         :param name: API name
         :param doc: the user facing document of this inference API, default to the
             docstring of the inference API function
-        :param handler: A InputAdapter that transforms HTTP Request and/or
+        :param input_adapter: A InputAdapter that transforms HTTP Request and/or
             CLI options into parameters for API func
-        :param func: the user-defined API callback function, this is
+        :param user_func: the user-defined API callback function, this is
             typically the 'predict' method on a model
+        :param output_adapter: A OutputAdapter is an layer between result of user
+            defined API callback function
+            and final output in a variety of different forms,
+            such as HTTP response, command line stdout or AWS Lambda event object.
         :param mb_max_latency: The latency goal of this inference API in milliseconds.
             Default: 10000.
         :param mb_max_batch_size: The maximum size of requests batch accepted by this
@@ -78,9 +98,9 @@ def __init__(
         """
         self._service = service
         self._name = name
-        self._handler = handler
-        self._func = func
-        self._wrapped_func = None
+        self._input_adapter = input_adapter
+        self._user_func = user_func
+        self._output_adapter = output_adapter
         self.mb_max_latency = mb_max_latency
         self.mb_max_batch_size = mb_max_batch_size
 
@@ -88,8 +108,8 @@ def __init__(
             # generate a default doc string for this inference API
             doc = (
                 f"BentoService inference API '{self.name}', input: "
-                f"'{type(handler).__name__}', output: "
-                f"'{type(handler.output_adapter).__name__}'"
+                f"'{type(input_adapter).__name__}', output: "
+                f"'{type(output_adapter).__name__}'"
             )
         self._doc = doc
 
@@ -115,73 +135,139 @@ def doc(self):
         return self._doc
 
     @property
-    def handler(self):
+    def input_adapter(self) -> BaseInputAdapter:
         """
-        handler is a deprecated concept, we are moving it to the new input/output
-        adapter interface. Currently this `handler` property returns the input adapter
-        of this inference API
+        :return: the input adapter of this inference API
         """
-        return self._handler
+        return self._input_adapter
 
     @property
-    def output_adapter(self):
+    def output_adapter(self) -> BaseOutputAdapter:
         """
         :return: the output adapter of this inference API
         """
-        return self.handler.output_adapter
+        return self._output_adapter
 
     @cached_property
-    def func(self):
+    def user_func(self):
         """
         :return: user-defined inference API callback function
         """
 
-        def func_with_tracing(*args, **kwargs):
+        # allow user to define handlers without 'contexts' kwargs
+        _sig = inspect.signature(self._user_func)
+        try:
+            _sig.bind(contexts=None)
+            append_contexts = False
+        except TypeError:
+            append_contexts = True
+
+        @functools.wraps(self._user_func)
+        def wrapped_func(*args, **kwargs):
             with trace(
                 service_name=self.__class__.__name__,
                 span_name="user defined inference api callback function",
             ):
-                return self._func(*args, **kwargs)
+                if append_contexts:
+                    kwargs.pop('contexts')
+                return self._user_func(*args, **kwargs)
 
-        return func_with_tracing
+        return wrapped_func
 
     @property
     def request_schema(self):
         """
         :return: the HTTP API request schema in OpenAPI/Swagger format
         """
-        schema = self.handler.request_schema
+        schema = self.input_adapter.request_schema
         if schema.get('application/json'):
-            schema.get('application/json')['example'] = self.handler._http_input_example
+            schema.get('application/json')[
+                'example'
+            ] = self.input_adapter._http_input_example
         return schema
 
+    def infer(self, inf_tasks: Iterable[InferenceTask]) -> Tuple[InferenceResult]:
+        # task validation
+        def _validation(inf_tasks):
+            for task in inf_tasks:
+                if task.is_discarded:
+                    continue
+                if self.input_adapter.validate_task(task):
+                    yield task
+                else:
+                    task.discard(http_status=400, err_msg="Input validation failed")
+
+        # extract args
+        user_args = self.input_adapter.extract_user_func_args(_validation(inf_tasks))
+        contexts = tuple(t.context for t in inf_tasks if not t.is_discarded)
+
+        # call user function
+        user_return = self.user_func(*user_args, contexts=contexts)
+
+        if (
+            isinstance(user_return, (tuple, list))
+            and user_return
+            and isinstance(user_return[0], InferenceResult)
+        ):
+            inf_results = user_return
+        else:
+            # pack return value
+            inf_results = self.output_adapter.pack_user_func_return_value(
+                user_return, contexts=contexts
+            )
+        full_results = InferenceResult.complete_discarded(inf_tasks, inf_results)
+        return full_results
+
     def handle_request(self, request: flask.Request):
-        return self.handler.handle_request(request, self.func)
+        reqs = [HTTPRequest.from_flask_request(request)]
+        inf_tasks = self.input_adapter.from_http_request(reqs)
+        results = self.infer(inf_tasks)
+        resps = self.output_adapter.to_http_response(results)
+        response = next(iter(resps))
+        return response.to_flask_response()
 
     def handle_batch_request(self, request: flask.Request):
         from bentoml.marshal.utils import DataLoader
 
-        requests = DataLoader.split_requests(request.get_data())
+        reqs = DataLoader.split_requests(request.get_data())
 
         with trace(
             service_name=self.__class__.__name__,
-            span_name=f"call `{self.handler.__class__.__name__}`",
+            span_name=f"call `{self.input_adapter.__class__.__name__}`",
         ):
-            responses = self.handler.handle_batch_request(requests, self.func)
+            inf_tasks = self.input_adapter.from_http_request(reqs)
+            results = self.infer(inf_tasks)
+            responses = self.output_adapter.to_http_response(results)
 
         return DataLoader.merge_responses(responses)
 
-    def handle_cli(self, args):
-        return self.handler.handle_cli(args, self.func)
+    def handle_cli(self, cli_args: Tuple[str]) -> int:
+        parser = argparse.ArgumentParser()
+        parser.add_argument("--max-batch-size", default=sys.maxsize, type=int)
+        parsed_args, _ = parser.parse_known_args(cli_args)
+
+        exit_code = 0
+
+        tasks_iter = self.input_adapter.from_cli(cli_args)
+        while True:
+            tasks = tuple(itertools.islice(tasks_iter, parsed_args.max_batch_size))
+            if not len(tasks):
+                break
+            results = self.infer(tasks)
+            exit_code = exit_code or self.output_adapter.to_cli(results)
+
+        return exit_code
 
     def handle_aws_lambda_event(self, event):
-        return self.handler.handle_aws_lambda_event(event, self.func)
+        inf_tasks = self.input_adapter.from_aws_lambda_event([event])
+        results = self.infer(inf_tasks)
+        return next(iter(self.output_adapter.to_aws_lambda_event(results)))
 
 
-def validate_inference_api_name(api_name):
-    if not isidentifier(api_name):
+def validate_inference_api_name(api_name: str):
+    if not api_name.isidentifier():
         raise InvalidArgument(
-            "Invalid API name: '{}', a valid identifier must contains only letters,"
+            "Invalid API name: '{}', a valid identifier may only contain letters,"
             " numbers, underscores and not starting with a number.".format(api_name)
         )
 
@@ -195,10 +281,10 @@ def api_decorator(
     *args,
     input: BaseInputAdapter = None,
     output: BaseOutputAdapter = None,
-    api_name=None,
-    api_doc=None,
-    mb_max_batch_size=DEFAULT_MAX_BATCH_SIZE,
-    mb_max_latency=DEFAULT_MAX_LATENCY,
+    api_name: str = None,
+    api_doc: str = None,
+    mb_max_batch_size: int = DEFAULT_MAX_BATCH_SIZE,
+    mb_max_latency: int = DEFAULT_MAX_LATENCY,
     **kwargs,
 ):  # pylint: disable=redefined-builtin
     """
@@ -233,23 +319,6 @@ def api_decorator(
     >>>     @api(input=DataframeInput(input_json_orient='records'))
     >>>     def identity(self, df):
     >>>         # user-defined callback function that process inference requests
-
-
-    Deprecated syntax before version 0.8.0:
-
-    >>> from bentoml import BentoService, api
-    >>> from bentoml.handlers import JsonHandler, DataframeHandler  # deprecated
-    >>>
-    >>> class FraudDetectionAndIdentityService(BentoService):
-    >>>
-    >>>     @api(JsonHandler)  # deprecated
-    >>>     def fraud_detect(self, parsed_json_list):
-    >>>         # do something
-    >>>
-    >>>     @api(DataframeHandler, input_json_orient='records')  # deprecated
-    >>>     def identity(self, df):
-    >>>         # do something
-
     """
 
     def decorator(func):
@@ -257,26 +326,31 @@ def decorator(func):
         validate_inference_api_name(_api_name)
 
         if input is None:
-            # support bentoml<=0.7
+            # Support passing the desired adapter without instantiation
             if not args or not (
                 inspect.isclass(args[0]) and issubclass(args[0], BaseInputAdapter)
             ):
                 raise InvalidArgument(
                     "BentoService @api decorator first parameter must "
-                    "be class derived from bentoml.adapters.BaseInputAdapter"
+                    "be an instance of a class derived from "
+                    "bentoml.adapters.BaseInputAdapter "
                 )
 
-            handler = args[0](*args[1:], output_adapter=output, **kwargs)
+            # noinspection PyPep8Naming
+            InputAdapter = args[0]
+            input_adapter = InputAdapter(*args[1:], **kwargs)
+            output_adapter = DefaultOutput()
         else:
             assert isinstance(input, BaseInputAdapter), (
-                "API input parameter must be an instance of any classes inherited from "
+                "API input parameter must be an instance of a class derived from "
                 "bentoml.adapters.BaseInputAdapter"
             )
-            handler = input
-            handler._output_adapter = output
+            input_adapter = input
+            output_adapter = output or DefaultOutput()
 
         setattr(func, "_is_api", True)
-        setattr(func, "_handler", handler)
+        setattr(func, "_input_adapter", input_adapter)
+        setattr(func, "_output_adapter", output_adapter)
         setattr(func, "_api_name", _api_name)
         setattr(func, "_api_doc", api_doc)
         setattr(func, "_mb_max_batch_size", mb_max_batch_size)
@@ -400,6 +474,9 @@ def ver_decorator(major, minor):
     'Patch' is provided(or auto generated) when calling BentoService#save,
     while 'Major' and 'Minor' can be defined with '@ver' decorator
 
+    >>>  from bentoml import ver, artifacts
+    >>>  from bentoml.artifact import PickleArtifact
+    >>>
     >>>  @ver(major=1, minor=4)
     >>>  @artifacts([PickleArtifact('model')])
     >>>  class MyMLService(BentoService):
@@ -421,7 +498,7 @@ def decorator(bento_service_cls):
     return decorator
 
 
-def _validate_version_str(version_str):
+def validate_version_str(version_str):
     """
     Validate that version str format is either a simple version string that:
         * Consist of only ALPHA / DIGIT / "-" / "." / "_"
@@ -437,7 +514,7 @@ def _validate_version_str(version_str):
         raise InvalidArgument(
             'Invalid BentoService version: "{}", it can only consist'
             ' ALPHA / DIGIT / "-" / "." / "_", and must be less than'
-            "128 characthers".format(version_str)
+            "128 characters".format(version_str)
         )
 
     if version_str.lower() == "latest":
@@ -500,15 +577,15 @@ class BentoService:
     """
 
     # List of inference APIs that this BentoService provides
-    _inference_apis = []
+    _inference_apis: InferenceAPI = []
 
     # Name of this BentoService. It is default the class name of this BentoService class
-    _bento_service_name = None
+    _bento_service_name: str = None
 
     # For BentoService loaded from saved bundle, this will be set to the path of bundle.
     # When user install BentoService bundle as a PyPI package, this will be set to the
     # installed site-package location of current python environment
-    _bento_service_bundle_path = None
+    _bento_service_bundle_path: str = None
 
     # List of artifacts required by this BentoService class, declared via the `@env`
     # decorator. This list is used for initializing an empty ArtifactCollection when
@@ -546,7 +623,9 @@ def _config_environments(self):
         self._env = self.__class__._env or BentoServiceEnv(self.name)
 
         for api in self._inference_apis:
-            self._env.add_pip_dependencies_if_missing(api.handler.pip_dependencies)
+            self._env.add_pip_dependencies_if_missing(
+                api.input_adapter.pip_dependencies
+            )
             self._env.add_pip_dependencies_if_missing(
                 api.output_adapter.pip_dependencies
             )
@@ -564,20 +643,22 @@ def _config_inference_apis(self):
             if hasattr(function, "_is_api"):
                 api_name = getattr(function, "_api_name")
                 api_doc = getattr(function, "_api_doc")
-                handler = getattr(function, "_handler")
+                input_adapter = getattr(function, "_input_adapter")
+                output_adapter = getattr(function, "_output_adapter")
                 mb_max_latency = getattr(function, "_mb_max_latency")
                 mb_max_batch_size = getattr(function, "_mb_max_batch_size")
 
                 # Bind api method call with self(BentoService instance)
-                func = function.__get__(self)
+                user_func = function.__get__(self)
 
                 self._inference_apis.append(
                     InferenceAPI(
                         self,
                         api_name,
                         api_doc,
-                        handler=handler,
-                        func=func,
+                        input_adapter=input_adapter,
+                        user_func=user_func,
+                        output_adapter=output_adapter,
                         mb_max_latency=mb_max_latency,
                         mb_max_batch_size=mb_max_batch_size,
                     )
@@ -680,7 +761,7 @@ def name(cls):  # pylint: disable=no-self-argument,invalid-overridden-method
         :return: BentoService name
         """
         if cls._bento_service_name is not None:
-            if not isidentifier(cls._bento_service_name):
+            if not cls._bento_service_name.isidentifier():
                 raise InvalidArgument(
                     'BentoService#_bento_service_name must be valid python identifier'
                     'matching regex `(letter|"_")(letter|digit|"_")*`'
@@ -709,7 +790,7 @@ def set_version(self, version_str=None):
                 [str(self._version_major), str(self._version_minor), version_str]
             )
 
-        _validate_version_str(version_str)
+        validate_version_str(version_str)
 
         if self.__class__._bento_service_bundle_version is not None:
             logger.warning(
diff --git a/bentoml/types.py b/bentoml/types.py
new file mode 100644
index 00000000000..b69c43e428f
--- /dev/null
+++ b/bentoml/types.py
@@ -0,0 +1,153 @@
+# Copyright 2020 Atalaya Tech, Inc.
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import io
+import functools
+from typing import (
+    NamedTuple,
+    Tuple,
+    Union,
+    Dict,
+    List,
+    Generic,
+    TypeVar,
+    Optional,
+    Iterable,
+    Iterator,
+)
+
+from multidict import CIMultiDict
+
+
+# For non latin1 characters: https://tools.ietf.org/html/rfc8187
+# Also https://github.com/benoitc/gunicorn/issues/1778
+HEADER_CHARSET = 'latin1'
+
+JSON_CHARSET = 'utf-8'
+
+
+class HTTPRequest(NamedTuple):
+    '''
+    headers: tuple of key value pairs in bytes
+    data: str
+    '''
+
+    headers: tuple = tuple()
+    body: bytes = b""
+
+    @property
+    @functools.lru_cache()
+    def parsed_headers(self):
+        return CIMultiDict(
+            (k.decode(HEADER_CHARSET), v.decode(HEADER_CHARSET))
+            for k, v in self.headers or tuple()
+        )
+
+    @classmethod
+    def from_flask_request(cls, request):
+        return cls(
+            tuple(
+                (k.encode(HEADER_CHARSET), v.encode(HEADER_CHARSET))
+                for k, v in request.headers
+            ),
+            request.get_data(),
+        )
+
+    def to_flask_request(self):
+        from werkzeug.wrappers import Request
+
+        return Request.from_values(
+            input_stream=io.BytesIO(self.body),
+            content_length=len(self.body),
+            headers=self.headers,
+        )
+
+
+class HTTPResponse(NamedTuple):
+    status: int = 200
+    headers: tuple = tuple()
+    body: str = ""  # TODO(bojiang): bytes
+
+    def to_flask_response(self):
+        import flask
+
+        return flask.Response(
+            status=self.status, headers=self.headers, response=self.body
+        )
+
+
+# https://tools.ietf.org/html/rfc7159#section-3
+JsonSerializable = Union[bool, None, Dict, List, int, float, str]
+
+# https://docs.aws.amazon.com/lambda/latest/dg/python-handler.html
+AwsLambdaEvent = Union[Dict, List, str, int, float, None]
+
+
+Input = TypeVar("Input")
+Output = TypeVar("Output")
+
+
+ApiFuncArgs = TypeVar("ApiFuncArgs")
+ApiFuncReturnValue = TypeVar("ApiFuncReturnValue")
+
+
+class InferenceContext(NamedTuple):
+    task_id: Optional[str] = None
+    err_msg: str = ''
+
+    http_method: Optional[str] = None
+    http_status: Optional[int] = None
+    http_headers: Optional[CIMultiDict] = None
+
+    aws_lambda_event: Optional[dict] = None
+
+    cli_status: Optional[int] = 0
+    cli_args: Optional[Tuple[str]] = None
+
+
+class DefaultErrorContext(InferenceContext):
+    http_status: int = 500
+    cli_status: int = 1
+
+
+class InferenceTask(Generic[Input]):
+    def __init__(self, data: Input, context: InferenceContext = None):
+        self.data = data
+        self.context = context or InferenceContext()
+        self.is_discarded = False
+
+    def discard(self, err_msg="", **context):
+        self.is_discarded = True
+        self.context = DefaultErrorContext(err_msg=err_msg, **context)
+
+
+class InferenceResult(Generic[Output]):
+    def __init__(self, data: Output = None, context: InferenceContext = None):
+        self.data = data
+        self.context = context or InferenceContext()
+
+    @classmethod
+    def complete_discarded(
+        cls, tasks: Iterable[InferenceTask], results: Iterable['InferenceResult'],
+    ) -> Iterator['InferenceResult']:
+        iresults = iter(results)
+        try:
+            for task in tasks:
+                if task.is_discarded:
+                    yield cls(None, context=DefaultErrorContext(*task.context))
+                else:
+                    yield next(iresults)
+        except StopIteration:
+            raise StopIteration(
+                'The results does not match the number of tasks'
+            ) from None
diff --git a/bentoml/utils/__init__.py b/bentoml/utils/__init__.py
index 93f582f95da..c5a28dc2f69 100644
--- a/bentoml/utils/__init__.py
+++ b/bentoml/utils/__init__.py
@@ -12,13 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import re
-from io import StringIO
 import socket
 from contextlib import contextmanager
 from functools import wraps
+from io import StringIO
 from urllib.parse import urlparse, uses_netloc, uses_params, uses_relative
 
+from google.protobuf.message import Message
+
 _VALID_URLS = set(uses_relative + uses_netloc + uses_params)
 _VALID_URLS.discard("")
 
@@ -35,27 +36,14 @@ def reserve_free_port(host='localhost'):
     sock.close()
 
 
-def is_url(url):
+def is_url(url: str) -> bool:
     try:
         return urlparse(url).scheme in _VALID_URLS
     except Exception:  # pylint:disable=broad-except
         return False
 
 
-def isidentifier(s):
-    """
-    Return true if string is in a valid python identifier format:
-
-    https://docs.python.org/2/reference/lexical_analysis.html#identifiers
-    """
-    try:
-        return s.isidentifier()
-    except AttributeError:
-        # str#isidentifier is only available in python 3
-        return re.match(r"[A-Za-z_][A-Za-z_0-9]*\Z", s) is not None
-
-
-def dump_to_yaml_str(yaml_dict):
+def dump_to_yaml_str(yaml_dict: object) -> str:
     from ruamel.yaml import YAML
 
     yaml = YAML()
@@ -64,14 +52,14 @@ def dump_to_yaml_str(yaml_dict):
     return string_io.getvalue()
 
 
-def pb_to_yaml(message):
+def pb_to_yaml(message: Message) -> str:
     from google.protobuf.json_format import MessageToDict
 
     message_dict = MessageToDict(message)
     return dump_to_yaml_str(message_dict)
 
 
-def ProtoMessageToDict(protobuf_msg, **kwargs):
+def ProtoMessageToDict(protobuf_msg: Message, **kwargs) -> object:
     from google.protobuf.json_format import MessageToDict
 
     if 'preserving_proto_field_name' not in kwargs:
@@ -81,7 +69,7 @@ def ProtoMessageToDict(protobuf_msg, **kwargs):
 
 
 # This function assume the status is not status.OK
-def status_pb_to_error_code_and_message(pb_status):
+def status_pb_to_error_code_and_message(pb_status) -> (int, str):
     from bentoml.yatai.proto import status_pb2
 
     assert pb_status.status_code != status_pb2.Status.OK
diff --git a/bentoml/utils/alg.py b/bentoml/utils/alg.py
index 8f456cf1b9f..0353e918c04 100644
--- a/bentoml/utils/alg.py
+++ b/bentoml/utils/alg.py
@@ -33,7 +33,7 @@ def __len__(self):
     def __getitem__(self, sl):
         if not self._flag_full:
             return self._data[: self._cur][sl]
-        return (self._data[self._cur :] + self._data[: self._cur])[sl]
+        return (self._data[self._cur:] + self._data[: self._cur])[sl]
 
 
 class TokenBucket:
diff --git a/bentoml/utils/benchmark.py b/bentoml/utils/benchmark.py
index 8a9c40724f6..0a8f2b0bec9 100644
--- a/bentoml/utils/benchmark.py
+++ b/bentoml/utils/benchmark.py
@@ -50,9 +50,9 @@ class DynamicBucketMerge:
     def __init__(self, sample_range=1, bucket_num=10):
         self.bucket_num = bucket_num
         self.sample_range = sample_range
-        self.bucket = [0] * (bucket_num)
-        self.bucket_sample = [0] * (bucket_num)
-        self.bucket_ver = [0] * (bucket_num)
+        self.bucket = [0] * bucket_num
+        self.bucket_sample = [0] * bucket_num
+        self.bucket_ver = [0] * bucket_num
 
     def put(self, timestamp, num):
         timestamp = timestamp / self.sample_range
@@ -198,7 +198,7 @@ def print_sumup(self):
         if health < 90:
             print(
                 """
-                *** WARNNING ***
+                *** WARNING ***
                 The client health rate is low. The benchmark result is not reliable.
                 Possible solutions:
                 * check the failure_rate and avoid request failures
@@ -229,14 +229,14 @@ class BenchmarkClient:
     """
     A locust-like benchmark tool with asyncio.
     Features:
-    * Very effcient, low CPU cost
-    * Could be embeded into other asyncio APPs, like jupyter notebook
+    * Very efficient, low CPU cost
+    * Could be embedded into other asyncio apps, like jupyter notebook
 
     Paras:
     * request_producer: The test case producer, a function with return value
         (url: str, method: str, headers: dict, data: str)
     * request_interval: intervals in seconds between each requests of the same user,
-      lazy value supoorted.
+      lazy value supported.
         for eg:
             - 1  # for constant 1 sec
             - lambda: random.random()  # for random wait time between 0 and 1
@@ -298,6 +298,7 @@ async def _start(self):
                 req_url = self.url_override or url
                 err = ''
                 group = ''
+                # noinspection PyUnresolvedReferences
                 try:
                     async with sess.request(
                         method,
@@ -310,7 +311,7 @@ async def _start(self):
                         if not self.verify_response(r.status, msg):
                             group = f"{r.status}"
                             err = f"<status: {r.status}>\n{msg}"
-                except asyncio.CancelledError:
+                except asyncio.CancelledError:  # pylint: disable=try-except-raise
                     raise
                 except (
                     aiohttp.client_exceptions.ServerDisconnectedError,
@@ -417,7 +418,7 @@ def start_session(self, session_time, total_user, spawn_speed=None):
         Paras:
         * session_time: session time in sec
         * total_user: user count need to be spawned
-        * spawn_speed: user spawnning speed, in user/sec
+        * spawn_speed: user spawning speed, in user/sec
         """
         if spawn_speed is None:
             spawn_speed = total_user
diff --git a/bentoml/utils/dataframe_util.py b/bentoml/utils/dataframe_util.py
index 0a55874c25f..00730a383d5 100644
--- a/bentoml/utils/dataframe_util.py
+++ b/bentoml/utils/dataframe_util.py
@@ -121,7 +121,7 @@ def _iter_line(line):
             for i, c in enumerate(line):
                 if c == '"':
                     quoted = not quoted
-                if not quoted and string[i : i + dlen] == delimiter:
+                if not quoted and string[i: i + dlen] == delimiter:
                     yield line[last_cur:i]
                     last_cur = i + dlen
                     split += 1
@@ -202,8 +202,8 @@ def _guess_orient(table):
             if any(isinstance(k, str) and not k.isnumeric() for k in table.keys()):
                 return 'columns'
             if any(
-                isinstance(k, str) and not k.isnumeric()
-                for k in next(iter(table.values())).keys()
+                    isinstance(k, str) and not k.isnumeric()
+                    for k in next(iter(table.values())).keys()
             ):
                 return 'index'
             return {'columns', 'index'}
@@ -221,14 +221,13 @@ def _guess_orient(table):
     # 'table': _from_json_table,
 }
 
-
 PANDAS_DATAFRAME_TO_JSON_ORIENT_OPTIONS = {k for k in _ORIENT_MAP}
 
 
 def _dataframe_csv_from_input(tables, content_types, orients):
     state = DataFrameState()
     for table_id, (table, content_type, orient) in enumerate(
-        zip(tables, content_types, orients)
+            zip(tables, content_types, orients)
     ):
         content_type = content_type or "application/json"
         if content_type.lower() == "application/json":
@@ -298,10 +297,11 @@ def _gen_slice(ids):
 
 
 def read_dataframes_from_json_n_csv(
-    datas: Iterable["pd.DataFrame"], content_types: Iterable[str], orient: str = None,
+        datas: Iterable["pd.DataFrame"], content_types: Iterable[str],
+        orient: str = None,
 ) -> ("pd.DataFrame", Iterable[slice]):
     '''
-    load detaframes from multiple raw datas in json or csv fromat, efficiently
+    load dataframes from multiple raw datas in json or csv format, efficiently
 
     Background: Each calling of pandas.read_csv or pandas.read_json cost about 100ms,
     no matter how many lines it contains. Concat jsons/csvs before read_json/read_csv
diff --git a/bentoml/utils/flask_ngrok.py b/bentoml/utils/flask_ngrok.py
index 5a2a14566b0..cfb9026d9d5 100644
--- a/bentoml/utils/flask_ngrok.py
+++ b/bentoml/utils/flask_ngrok.py
@@ -71,7 +71,7 @@ def _download_ngrok(ngrok_path):
 
 def _download_file(url):
     """
-    Download ngrok bibary file to local
+    Download ngrok binary file to local
     :param url:
     :return:
     """
diff --git a/bentoml/utils/lazy_loader.py b/bentoml/utils/lazy_loader.py
index 1b7735851ac..4b7d9922334 100644
--- a/bentoml/utils/lazy_loader.py
+++ b/bentoml/utils/lazy_loader.py
@@ -6,7 +6,7 @@
 
 
 class LazyLoader(types.ModuleType):
-    """LazyLoder module borrowed from Tensorflow
+    """LazyLoader module borrowed from Tensorflow
   https://github.com/tensorflow/tensorflow/blob/v2.2.0/tensorflow/python/util/lazy_loader.py
 
   Lazily import a module, mainly to avoid pulling in large dependencies.
diff --git a/bentoml/utils/tempdir.py b/bentoml/utils/tempdir.py
index ca944a536dc..93e1d6cef05 100644
--- a/bentoml/utils/tempdir.py
+++ b/bentoml/utils/tempdir.py
@@ -58,7 +58,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     def create(self):
         if self.path is not None:
-            logger.debug("Skipped temp direcotry creation: %s", self.path)
+            logger.debug("Skipped temp directory creation: %s", self.path)
             return self.path
 
         tempdir = tempfile.mkdtemp(prefix="bentoml-{}-".format(self._prefix))
diff --git a/bentoml/yatai/deployment/azure_functions/operator.py b/bentoml/yatai/deployment/azure_functions/operator.py
index fac95952904..06dbebe2b5b 100644
--- a/bentoml/yatai/deployment/azure_functions/operator.py
+++ b/bentoml/yatai/deployment/azure_functions/operator.py
@@ -113,10 +113,10 @@ def _init_azure_functions_project(
 
 
 def _generate_azure_resource_names(namespace, deployment_name):
-    # Genrate resource names base on
+    # Generate resource names base on
     # https://docs.microsoft.com/en-us/azure/azure-resource-manager/management/resource-name-rules
 
-    # 1-90, alphannumeric(A-Za-z0-9) underscores, parenthese, hyphens, periods
+    # 1-90, alphanumeric(A-Za-z0-9) underscores, parentheses, hyphens, periods
     # scope subscription
     resource_group_name = f'{namespace}-{deployment_name}'
     if len(resource_group_name) > MAX_RESOURCE_GROUP_NAME_LENGTH:
diff --git a/bentoml/yatai/deployment/sagemaker/model_server.py b/bentoml/yatai/deployment/sagemaker/model_server.py
index 5f70ebb546b..68c420b0359 100644
--- a/bentoml/yatai/deployment/sagemaker/model_server.py
+++ b/bentoml/yatai/deployment/sagemaker/model_server.py
@@ -42,7 +42,7 @@ def setup_routes(app, bento_service, api_name):
     setup_bento_service_api_route(app, api)
 
 
-# AWS Sagemaker requires custom inference docker conatiner to implement a web server
+# AWS Sagemaker requires custom inference docker container to implement a web server
 # that responds to /invocations and /ping on port 8080.
 AWS_SAGEMAKER_SERVE_PORT = 8080
 
diff --git a/bentoml/yatai/deployment/sagemaker/operator.py b/bentoml/yatai/deployment/sagemaker/operator.py
index aea45d75192..5b1291c8aea 100644
--- a/bentoml/yatai/deployment/sagemaker/operator.py
+++ b/bentoml/yatai/deployment/sagemaker/operator.py
@@ -183,7 +183,7 @@ def _aws_client_error_to_bentoml_exception(e, message_prefix=None):
     """parse botocore.exceptions.ClientError into Bento StatusProto
 
     We handle two most common errors when deploying to Sagemaker.
-        1. Authenication issue/invalid access(InvalidSignatureException)
+        1. Authentication issue/invalid access(InvalidSignatureException)
         2. resources not found (ValidationException)
     It will return correlated StatusProto(NOT_FOUND, UNAUTHENTICATED)
 
diff --git a/bentoml/yatai/deployment/sagemaker/serve b/bentoml/yatai/deployment/sagemaker/serve
index d354092f21e..d240df52906 100644
--- a/bentoml/yatai/deployment/sagemaker/serve
+++ b/bentoml/yatai/deployment/sagemaker/serve
@@ -56,7 +56,7 @@ def _serve():
         signal.SIGTERM, lambda a, b: sigterm_handler(nginx.pid, gunicorn_app.pid)
     )
 
-    pids = set([nginx.pid, gunicorn_app.pid])
+    pids = {nginx.pid, gunicorn_app.pid}
     while True:
         pid, _ = os.wait()
         if pid in pids:
diff --git a/bentoml/yatai/deployment/utils.py b/bentoml/yatai/deployment/utils.py
index 284b11768d9..32e429fd4ac 100644
--- a/bentoml/yatai/deployment/utils.py
+++ b/bentoml/yatai/deployment/utils.py
@@ -90,7 +90,7 @@ def generate_aws_compatible_string(*items, max_length=63):
     length in a tuple, e.g.:
 
     >> generate_aws_compatible_string("abc", "def")
-    >> 'abc-def'  # concatenate mupltiple parts
+    >> 'abc-def'  # concatenate multiple parts
 
     >> generate_aws_compatible_string("abc_def")
     >> 'abc-def'  # replace invalid chars to '-'
@@ -98,16 +98,16 @@ def generate_aws_compatible_string(*items, max_length=63):
     >> generate_aws_compatible_string(("ab", 1), ("bcd", 2), max_length=4)
     >> 'a-bc'  # trim based on max_length of each part
     """
-    trimed_items = [
+    trimmed_items = [
         item[0][: item[1]] if type(item) == tuple else item for item in items
     ]
     items = [item[0] if type(item) == tuple else item for item in items]
 
-    for i in range(len(trimed_items)):
+    for i in range(len(trimmed_items)):
         if len('-'.join(items)) <= max_length:
             break
         else:
-            items[i] = trimed_items[i]
+            items[i] = trimmed_items[i]
 
     name = '-'.join(items)
     if len(name) > max_length:
diff --git a/bentoml/yatai/web/client/src/components/BentoServiceTable.tsx b/bentoml/yatai/web/client/src/components/BentoServiceTable.tsx
index d5d2baa7e91..3827e9c4a2d 100644
--- a/bentoml/yatai/web/client/src/components/BentoServiceTable.tsx
+++ b/bentoml/yatai/web/client/src/components/BentoServiceTable.tsx
@@ -1,7 +1,7 @@
 import * as React from "react";
 import { Link } from "react-router-dom";
 
-import { displayTimeInFromNowFormat } from "../utils/index";
+import { displayTimeInFromNowFormat } from "../utils";
 import Table from "../ui/Table";
 
 const apisToDisplayFormat = (apis) => {
diff --git a/bentoml/yatai/web/client/src/components/DeploymentsTable.tsx b/bentoml/yatai/web/client/src/components/DeploymentsTable.tsx
index 3c4d51a26f6..1b179f6f762 100644
--- a/bentoml/yatai/web/client/src/components/DeploymentsTable.tsx
+++ b/bentoml/yatai/web/client/src/components/DeploymentsTable.tsx
@@ -1,8 +1,7 @@
 import * as React from "react";
-import * as moment from "moment";
 import { Link } from "react-router-dom";
 
-import { displayTimeInFromNowFormat } from "../utils/index";
+import { displayTimeInFromNowFormat } from "../utils";
 import Table from "../ui/Table";
 import DeploymentStatusTag from "./DeploymentDetail/DeploymentStatusTag";
 
diff --git a/bentoml/yatai/web/client/src/components/NavigationBar.tsx b/bentoml/yatai/web/client/src/components/NavigationBar.tsx
index cd3de899d37..e3e1139ef77 100644
--- a/bentoml/yatai/web/client/src/components/NavigationBar.tsx
+++ b/bentoml/yatai/web/client/src/components/NavigationBar.tsx
@@ -13,12 +13,12 @@ import {
 import logo from "../assets/bentoml-logo.png";
 
 const NavigationBar = () => {
-  const [highlight, setHighlignt] = React.useState("");
+  const [highlight, setHighlight] = React.useState("");
   const location = useLocation();
 
   React.useEffect(() => {
     const currentLocation = location.pathname.split("/")[1];
-    setHighlignt(currentLocation);
+    setHighlight(currentLocation);
   }, [location]);
 
   return (
@@ -26,7 +26,7 @@ const NavigationBar = () => {
       <NavbarGroup align={Alignment.LEFT}>
         <NavbarHeading>
           <Link to="/">
-            <img src={`/${logo}`} width={150} />{" "}
+            <img src={`/${logo}`} width={150}  alt={""}/>{" "}
           </Link>
         </NavbarHeading>
         <NavbarDivider />
diff --git a/bentoml/yatai/web/src/yatai_client.ts b/bentoml/yatai/web/src/yatai_client.ts
index 9d2ac4dc0ee..7e59f2e6498 100644
--- a/bentoml/yatai/web/src/yatai_client.ts
+++ b/bentoml/yatai/web/src/yatai_client.ts
@@ -1,5 +1,5 @@
-import { Client, credentials, ServerCredentials } from "@grpc/grpc-js";
-import { bentoml } from "./generated/bentoml_grpc";
+import {Client, credentials} from "@grpc/grpc-js";
+import {bentoml} from "./generated/bentoml_grpc";
 
 export const createYataiClient = (grpcAddress: string) => {
   const client = new Client(grpcAddress, credentials.createInsecure());
@@ -19,6 +19,5 @@ export const createYataiClient = (grpcAddress: string) => {
     );
   };
 
-  const yataiClient = bentoml.Yatai.create(rpcImpl, false, false);
-  return yataiClient;
+  return bentoml.Yatai.create(rpcImpl, false, false);
 };
diff --git a/bentoml/yatai/yatai_service.py b/bentoml/yatai/yatai_service.py
index c2f8eb87a11..81e24f199c9 100644
--- a/bentoml/yatai/yatai_service.py
+++ b/bentoml/yatai/yatai_service.py
@@ -99,9 +99,9 @@ def start_yatai_service_grpc_server(
         )
 
         ensure_node_available_or_raise()
-        yatai_grpc_server_addess = f'localhost:{grpc_port}'
+        yatai_grpc_server_address = f'localhost:{grpc_port}'
         async_start_yatai_service_web_ui(
-            yatai_grpc_server_addess, ui_port, web_ui_log_path, debug_mode
+            yatai_grpc_server_address, ui_port, web_ui_log_path, debug_mode
         )
 
     # We don't import _echo function from click_utils because of circular dep
diff --git a/tests/adapters/test_annotated_image_handler.py b/tests/adapters/test_annotated_image_handler.py
index 0a9375ec006..a2a42cd42f3 100644
--- a/tests/adapters/test_annotated_image_handler.py
+++ b/tests/adapters/test_annotated_image_handler.py
@@ -8,9 +8,9 @@
 from werkzeug import Request
 from urllib3.filepost import encode_multipart_formdata
 
+from bentoml.types import HTTPRequest
 from bentoml.exceptions import BadInput
 from bentoml.adapters import AnnotatedImageInput
-from bentoml.marshal.utils import SimpleRequest
 
 
 def generate_multipart_body(image_file, json_file=None):
@@ -310,7 +310,7 @@ def test_anno_image_input_batch_request(img_file, json_file):
     adapter = AnnotatedImageInput(is_batch_input=True)
 
     multipart_data, headers = generate_multipart_body(img_file, json_file)
-    request = SimpleRequest.from_flask_request(
+    request = HTTPRequest.from_flask_request(
         Request.from_values(
             data=multipart_data,
             content_type=headers['Content-Type'],
@@ -347,9 +347,9 @@ def test_anno_image_input_batch_request_skip_bad(img_file, json_file):
 
     multipart_data, headers = generate_multipart_body(img_file, json_file)
 
-    empty_request = SimpleRequest(headers=headers, data=None)
+    empty_request = HTTPRequest(headers=headers, data=None)
 
-    request = SimpleRequest.from_flask_request(
+    request = HTTPRequest.from_flask_request(
         Request.from_values(
             data=multipart_data,
             content_type=headers['Content-Type'],
@@ -362,7 +362,7 @@ def test_anno_image_input_batch_request_skip_bad(img_file, json_file):
     files = {"image.invalid": image, "annotations.invalid": json}
     bad_data, content_type = encode_multipart_formdata(files)
 
-    bad_request = SimpleRequest.from_flask_request(
+    bad_request = HTTPRequest.from_flask_request(
         Request.from_values(
             data=bad_data, content_type=content_type, content_length=len(bad_data),
         )
@@ -388,7 +388,7 @@ def test_anno_image_input_batch_request_image_only(img_file):
     adapter = AnnotatedImageInput(is_batch_input=True)
 
     multipart_data, headers = generate_multipart_body(img_file)
-    request = SimpleRequest.from_flask_request(
+    request = HTTPRequest.from_flask_request(
         Request.from_values(
             data=multipart_data,
             content_type=headers['Content-Type'],
diff --git a/tests/adapters/test_dataframe_handler.py b/tests/adapters/test_dataframe_handler.py
index 3735c1619b1..14ff0854d40 100644
--- a/tests/adapters/test_dataframe_handler.py
+++ b/tests/adapters/test_dataframe_handler.py
@@ -122,7 +122,7 @@ def test_function(df):
     request.content_type = 'text/csv'
     request.get_data.return_value = csv_data
 
-    result = input_adapter.handle_request(request, test_function)
+    result = input_adapter.handle_request(request)
     assert result.get_data().decode('utf-8') == '"john"'
 
 
diff --git a/tests/adapters/test_file_handler.py b/tests/adapters/test_file_handler.py
index ef7ea979391..9b81ceb39a3 100644
--- a/tests/adapters/test_file_handler.py
+++ b/tests/adapters/test_file_handler.py
@@ -7,8 +7,8 @@
 import flask
 
 from bentoml.exceptions import BadInput
+from bentoml.types import HTTPRequest
 from bentoml.adapters import FileInput
-from bentoml.marshal.utils import SimpleRequest
 
 
 def predict(files):
@@ -62,16 +62,16 @@ def test_file_input_http_request_post_binary(bin_file):
     request.headers = {}
     request.get_data.return_value = open(str(bin_file), 'rb').read()
 
-    response = test_file_input.handle_request(request, predict)
+    response = test_file_input.handle_request(request)
 
     assert response.status_code == 200
     assert b'{"b64": "gTCJOQ=="}' in response.data
 
-    simple_request = SimpleRequest.from_flask_request(request)
+    simple_request = HTTPRequest.from_flask_request(request)
     responses = test_file_input.handle_batch_request([simple_request], predict)
 
     assert responses[0].status == 200
-    assert '{"b64": "gTCJOQ=="}' == responses[0].data
+    assert '{"b64": "gTCJOQ=="}' == responses[0].body
 
 
 def test_file_input_http_request_multipart_form(bin_file):
@@ -90,7 +90,7 @@ def test_file_input_http_request_multipart_form(bin_file):
     request.headers = {}
     request.get_data.return_value = None
 
-    response = test_file_input.handle_request(request, predict)
+    response = test_file_input.handle_request(request)
 
     assert response.status_code == 200
     assert b'{"b64": "gTCJOQ=="}' in response.data
@@ -112,7 +112,7 @@ def test_file_input_http_request_single_file_different_name(bin_file):
     request.headers = {}
     request.get_data.return_value = None
 
-    response = test_file_input.handle_request(request, predict)
+    response = test_file_input.handle_request(request)
 
     assert response.status_code == 200
     assert b'{"b64": "gTCJOQ=="}' in response.data
@@ -128,7 +128,7 @@ def test_file_input_http_request_malformatted_input_missing_file_file():
     request.get_data.return_value = None
 
     with pytest.raises(BadInput) as e:
-        test_file_input.handle_request(request, predict)
+        test_file_input.handle_request(request)
 
     assert "unexpected HTTP request format" in str(e.value)
 
@@ -143,6 +143,6 @@ def test_file_input_http_request_malformatted_input_wrong_input_name():
     request.get_data.return_value = None
 
     with pytest.raises(BadInput) as e:
-        test_file_input.handle_request(request, predict)
+        test_file_input.handle_request(request)
 
     assert "unexpected HTTP request format" in str(e.value)
diff --git a/tests/adapters/test_image_handler.py b/tests/adapters/test_image_handler.py
index e7e6d88ea63..d76349be06c 100644
--- a/tests/adapters/test_image_handler.py
+++ b/tests/adapters/test_image_handler.py
@@ -7,8 +7,8 @@
 import pytest
 
 from bentoml.exceptions import BadInput
+from bentoml.types import HTTPRequest
 from bentoml.adapters import ImageInput
-from bentoml.marshal.utils import SimpleRequest
 
 
 def predict(images):
@@ -62,16 +62,16 @@ def test_image_input_http_request_post_binary(img_file):
     request.headers = {}
     request.get_data.return_value = open(str(img_file), 'rb').read()
 
-    response = test_image_input.handle_request(request, predict)
+    response = test_image_input.handle_request(request)
 
     assert response.status_code == 200
     assert "[10, 10, 3]" in str(response.response)
 
-    simple_request = SimpleRequest.from_flask_request(request)
+    simple_request = HTTPRequest.from_flask_request(request)
     responses = test_image_input.handle_batch_request([simple_request], predict)
 
     assert responses[0].status == 200
-    assert "[10, 10, 3]" in str(responses[0].data)
+    assert "[10, 10, 3]" in str(responses[0].body)
 
 
 def test_image_input_http_request_multipart_form(img_file):
@@ -90,7 +90,7 @@ def test_image_input_http_request_multipart_form(img_file):
     request.headers = {}
     request.get_data.return_value = None
 
-    response = test_image_input.handle_request(request, predict)
+    response = test_image_input.handle_request(request)
 
     assert response.status_code == 200
     assert "[10, 10, 3]" in str(response.response)
@@ -112,7 +112,7 @@ def test_image_input_http_request_single_image_different_name(img_file):
     request.headers = {}
     request.get_data.return_value = None
 
-    response = test_image_input.handle_request(request, predict)
+    response = test_image_input.handle_request(request)
 
     assert response.status_code == 200
     assert "[10, 10, 3]" in str(response.response)
@@ -128,7 +128,7 @@ def test_image_input_http_request_malformatted_input_missing_image_file():
     request.get_data.return_value = None
 
     with pytest.raises(BadInput) as e:
-        test_image_input.handle_request(request, predict)
+        test_image_input.handle_request(request)
 
     assert "unexpected HTTP request format" in str(e.value)
 
@@ -143,6 +143,6 @@ def test_image_input_http_request_malformatted_input_wrong_input_name():
     request.get_data.return_value = None
 
     with pytest.raises(BadInput) as e:
-        test_image_input.handle_request(request, predict)
+        test_image_input.handle_request(request)
 
     assert "unexpected HTTP request format" in str(e.value)
diff --git a/tests/adapters/test_json_handler.py b/tests/adapters/test_json_handler.py
deleted file mode 100644
index 1216df02475..00000000000
--- a/tests/adapters/test_json_handler.py
+++ /dev/null
@@ -1,58 +0,0 @@
-import pytest
-
-from bentoml.adapters import JsonInput
-from bentoml.exceptions import BadInput
-
-
-def test_json_handle_cli(capsys, tmpdir):
-    def test_func(objs):
-        return [obj["name"] for obj in objs]
-
-    input_adapter = JsonInput()
-
-    json_file = tmpdir.join("test.json")
-    with open(str(json_file), "w") as f:
-        f.write('{"name": "john","game": "mario","city": "sf"}')
-
-    test_args = ["--input={}".format(json_file)]
-    input_adapter.handle_cli(test_args, test_func)
-    out, _ = capsys.readouterr()
-    assert out.strip().endswith("john")
-
-
-def test_json_handle_aws_lambda_event():
-    test_content = '{"name": "john","game": "mario","city": "sf"}'
-
-    def test_func(objs):
-        return [obj["name"] for obj in objs]
-
-    input_adapter = JsonInput()
-    success_event_obj = {
-        "headers": {"Content-Type": "application/json"},
-        "body": test_content,
-    }
-    success_response = input_adapter.handle_aws_lambda_event(
-        success_event_obj, test_func
-    )
-
-    assert success_response["statusCode"] == 200
-    assert success_response["body"] == '"john"'
-
-    success_event_obj = {
-        "headers": {"Content-Type": "this_will_also_work"},
-        "body": test_content,
-    }
-    success_response = input_adapter.handle_aws_lambda_event(
-        success_event_obj, test_func
-    )
-    assert success_response["statusCode"] == 200
-    assert success_response["body"] == '"john"'
-
-    error_event_obj = {
-        "headers": {"Content-Type": "application/json"},
-        "body": "not a valid, json {}",
-    }
-    with pytest.raises(BadInput) as e:
-        input_adapter.handle_aws_lambda_event(error_event_obj, test_func)
-
-    assert "Request body must contain valid json" in str(e.value)
diff --git a/tests/adapters/test_json_input.py b/tests/adapters/test_json_input.py
new file mode 100644
index 00000000000..0dcec25d6c0
--- /dev/null
+++ b/tests/adapters/test_json_input.py
@@ -0,0 +1,72 @@
+# pylint: disable=redefined-outer-name
+import pytest
+import json
+import glob
+import contextlib
+from typing import List
+
+from bentoml.adapters import JsonInput
+
+
+@pytest.fixture()
+def input_adapter():
+    return JsonInput()
+
+
+@pytest.fixture()
+def raw_jsons(json_files_dir) -> List[bytes]:
+    with contextlib.ExitStack() as stack:
+        file_names = glob.glob(json_files_dir)
+        raws = [stack.enter_context(open(f, "rb")).read() for f in file_names]
+    return raws
+
+
+def test_json_from_cli(input_adapter, raw_jsons):
+    tasks = input_adapter.from_cli(raw_jsons)
+    for t, b in zip(tasks, raw_jsons):
+        assert t.data == b
+
+
+def test_json_from_aws_lambda_event(input_adapter, raw_jsons):
+    events = [
+        {"headers": {"Content-Type": "application/json"}, "body": r.decode(),}
+        for r in raw_jsons
+    ]
+    tasks = input_adapter.from_aws_lambda_event(events)
+    for t, r in zip(tasks, raw_jsons):
+        assert t.data == r
+
+    events = [
+        {"headers": {"Content-Type": "this_will_also_work"}, "body": r.decode(),}
+        for r in raw_jsons
+    ]
+    tasks = input_adapter.from_aws_lambda_event(events)
+    for t, r in zip(tasks, raw_jsons):
+        assert t.data == r
+
+    raw_jsons = [b"not a valid json {}"]
+    events = [
+        {"headers": {"Content-Type": "application/json"}, "body": r.decode(),}
+        for r in raw_jsons
+    ]
+    tasks = input_adapter.from_aws_lambda_event(events)
+    for t, r in zip(tasks, raw_jsons):
+        assert t.data == r
+
+
+def test_json_extract(input_adapter, raw_jsons, non_utf8_bytes):
+    tasks = input_adapter.from_cli(raw_jsons, [])
+    args = input_adapter.extract_user_func_args(tasks)
+    json_obj_list = args[0]
+    for o, r in zip(json_obj_list, raw_jsons):
+        assert o == json.loads(r.decode())
+
+    tasks = input_adapter.from_cli([non_utf8_bytes, b"not a valid json {}"], [])
+    args = input_adapter.extract_user_func_args(tasks)
+    json_obj_list = args[0]
+    assert json_obj_list == []
+    for task in tasks:
+        assert task.is_discarded
+        assert task.context.err_msg
+        assert task.context.http_status != 200
+        assert task.context.cli_status != 0
diff --git a/tests/adapters/test_legacy_image_handler.py b/tests/adapters/test_legacy_image_handler.py
index c2699b7da6e..81888572cff 100644
--- a/tests/adapters/test_legacy_image_handler.py
+++ b/tests/adapters/test_legacy_image_handler.py
@@ -48,7 +48,7 @@ def test_image_input_http_request_post_binary(img_file):
     request.headers = {}
     request.get_data.return_value = open(str(img_file), 'rb')
 
-    response = test_image_input.handle_request(request, predict)
+    response = test_image_input.handle_request(request)
 
     assert response.status_code == 200
     assert "[10, 10, 3]" in str(response.response)
@@ -68,7 +68,7 @@ def test_image_input_http_request_multipart_form(img_file):
     request.headers = {}
     request.get_data.return_value = None
 
-    response = test_image_input.handle_request(request, predict)
+    response = test_image_input.handle_request(request)
 
     assert response.status_code == 200
     assert "[10, 10, 3]" in str(response.response)
@@ -88,7 +88,7 @@ def test_image_input_http_request_single_image_different_name(img_file):
     request.headers = {}
     request.get_data.return_value = None
 
-    response = test_image_input.handle_request(request, predict)
+    response = test_image_input.handle_request(request)
 
     assert response.status_code == 200
     assert "[10, 10, 3]" in str(response.response)
@@ -104,7 +104,7 @@ def test_image_input_http_request_malformatted_input_missing_image_file():
     request.get_data.return_value = None
 
     with pytest.raises(BadInput) as e:
-        test_image_input.handle_request(request, predict)
+        test_image_input.handle_request(request)
 
     assert "unexpected HTTP request format" in str(e.value)
 
@@ -119,6 +119,6 @@ def test_image_input_http_request_malformatted_input_wrong_input_name():
     request.get_data.return_value = None
 
     with pytest.raises(BadInput) as e:
-        test_image_input.handle_request(request, predict)
+        test_image_input.handle_request(request)
 
     assert "unexpected HTTP request format" in str(e.value)
diff --git a/tests/adapters/test_multi_image_input_adapter.py b/tests/adapters/test_multi_image_input_adapter.py
index 52ab0f93c3c..74bd86fbb0d 100644
--- a/tests/adapters/test_multi_image_input_adapter.py
+++ b/tests/adapters/test_multi_image_input_adapter.py
@@ -1,8 +1,8 @@
 from werkzeug import Request
 
 from bentoml.exceptions import BadInput
+from bentoml.types import HTTPRequest
 from bentoml.adapters.multi_image_input import MultiImageInput
-from bentoml.marshal.utils import SimpleRequest
 from urllib3.filepost import encode_multipart_formdata
 
 import pytest
@@ -43,7 +43,7 @@ def test_multi_image_input_request(img_file):
         content_length=headers['Content-Length'],
     )
 
-    response = adapter.handle_request(request, predict)
+    response = adapter.handle_request(request)
     assert response.status_code == 200
     assert response.data == b'[[10, 10, 3], [10, 10, 3]]'
 
@@ -52,7 +52,7 @@ def test_multi_image_batch_input(img_file):
     adapter = MultiImageInput(("imageX", "imageY"), is_batch_input=True)
 
     multipart_data, headers = generate_multipart_body(img_file)
-    request = SimpleRequest.from_flask_request(
+    request = HTTPRequest.from_flask_request(
         Request.from_values(
             data=multipart_data,
             content_type=headers['Content-Type'],
@@ -63,14 +63,14 @@ def test_multi_image_batch_input(img_file):
     responses = adapter.handle_batch_request([request] * 5, predict)
     for response in responses:
         assert response.status == 200
-        assert response.data == '[[10, 10, 3], [10, 10, 3]]'
+        assert response.body == '[[10, 10, 3], [10, 10, 3]]'
 
 
 def test_bad_multi_image_batch_input(img_file):
     adapter = MultiImageInput(("imageX", "imageY"), is_batch_input=True)
 
     multipart_data, headers = generate_multipart_body(img_file)
-    request = SimpleRequest.from_flask_request(
+    request = HTTPRequest.from_flask_request(
         Request.from_values(
             data=multipart_data,
             content_type=headers['Content-Type'],
@@ -81,7 +81,7 @@ def test_bad_multi_image_batch_input(img_file):
     responses = adapter.handle_batch_request(
         [request] * 5
         + [
-            SimpleRequest.from_flask_request(
+            HTTPRequest.from_flask_request(
                 Request.from_values(
                     data=multipart_data,
                     content_type='application/octet-stream',
@@ -92,7 +92,7 @@ def test_bad_multi_image_batch_input(img_file):
         predict,
     )
     print(responses[-1])
-    assert isinstance(responses[-1], SimpleRequest)
+    assert isinstance(responses[-1], HTTPRequest)
 
 
 def test_multi_image_aws_event(img_file):
diff --git a/tests/adapters/test_tf_tensor_handler.py b/tests/adapters/test_tf_tensor_handler.py
index c130cb0d736..5b86fc923bc 100644
--- a/tests/adapters/test_tf_tensor_handler.py
+++ b/tests/adapters/test_tf_tensor_handler.py
@@ -113,7 +113,7 @@ def test_tf_tensor_handle_request(test_cases):
     input_data, headers, except_result = test_cases
     request.get_data.return_value = json.dumps(input_data).encode('utf-8')
     request.headers = headers
-    response = input_adapter.handle_request(request, lambda i: i)
+    response = input_adapter.handle_request(request)
 
     prediction = json.loads(response.get_data())
     assert_eq_or_both_nan(except_result, prediction)
@@ -124,7 +124,7 @@ def test_tf_tensor_handle_batch_request(test_cases):
     ref: https://www.tensorflow.org/tfx/serving/api_rest#request_format_2
     '''
     from bentoml.adapters import TfTensorInput
-    from bentoml.marshal.utils import SimpleRequest
+    from bentoml.types import HTTPRequest
 
     input_adapter = TfTensorInput()
     request = MagicMock(spec=flask.Request)
@@ -133,9 +133,9 @@ def test_tf_tensor_handle_batch_request(test_cases):
     request.get_data.return_value = json.dumps(input_data).encode('utf-8')
     request.headers = headers
     responses = input_adapter.handle_batch_request(
-        [SimpleRequest.from_flask_request(request)] * 3, lambda i: i
+        [HTTPRequest.from_flask_request(request)] * 3, lambda i: i
     )
 
     for response in responses:
-        prediction = json.loads(response.data)
+        prediction = json.loads(response.body)
         assert_eq_or_both_nan(except_result, prediction)
diff --git a/tests/conftest.py b/tests/conftest.py
index 8685f165fcb..cbcaab221a8 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -88,6 +88,11 @@ def json_file(tmpdir):
     return str(json_file_)
 
 
+@pytest.fixture()
+def non_utf8_bytes():
+    return "â".encode('gb18030')
+
+
 @pytest.fixture()
 def bin_file(tmpdir):
     bin_file_ = tmpdir.join("bin_file")
@@ -108,11 +113,20 @@ def bin_files(tmpdir):
 @pytest.fixture()
 def img_files(tmpdir):
     for i in range(10):
-        img_file_ = tmpdir.join(f"test_img_{i}.jpg")
+        img_file_ = tmpdir.join(f"img_{i}.jpg")
         imageio.imwrite(str(img_file_), np.zeros((10, 10)))
     return str(tmpdir.join("*.jpg"))
 
 
+@pytest.fixture()
+def json_files_dir(tmpdir):
+    for i in range(10):
+        file_ = tmpdir.join(f"json_{i}.json")
+        with open(file_, "w") as of:
+            of.write('{"i": %d, "name": "kaith", "game": "morrowind"}' % i)
+    return str(tmpdir.join("*.json"))
+
+
 class TestModel(object):
     def predict_dataframe(self, df):
         return df["col1"].sum()
diff --git a/tests/test_auto_pip_dependency.py b/tests/test_auto_pip_dependency.py
index 8bb2d69348d..ce4c40c582f 100644
--- a/tests/test_auto_pip_dependency.py
+++ b/tests/test_auto_pip_dependency.py
@@ -17,7 +17,7 @@ def test_auto_adapter_dependencies(bento_bundle_path):
     assert 'imageio' in dependencies
     assert 'bentoml' in dependencies
 
-    # Test that dependencies also wrote to BentoServiceMetadat config file
+    # Test that dependencies also wrote to BentoServiceMetadata config file
 
 
 def test_auto_artifact_dependencies():
diff --git a/tests/test_save_and_load.py b/tests/test_save_and_load.py
index f7e78d3f9e1..c94f3e16886 100644
--- a/tests/test_save_and_load.py
+++ b/tests/test_save_and_load.py
@@ -40,7 +40,7 @@ def test_save_and_load_model(tmpdir, example_bento_service_class):
     assert api.name == "predict"
     assert api.mb_max_latency == 1000
     assert api.mb_max_batch_size == 2000
-    assert isinstance(api.handler, DataframeInput)
+    assert isinstance(api.input_adapter, DataframeInput)
     assert api.func(1) == 2
 
     # Check api methods are available
@@ -82,7 +82,7 @@ def test_pack_on_bento_service_instance(tmpdir, example_bento_service_class):
 
     api = model_service.get_inference_api('predict')
     assert api.name == "predict"
-    assert isinstance(api.handler, DataframeInput)
+    assert isinstance(api.input_adapter, DataframeInput)
     assert api.func(1) == 2
     # Check api methods are available
     assert model_service.predict(1) == 2
diff --git a/tests/test_service.py b/tests/test_service.py
index 4afe45a0ade..d3201683a31 100644
--- a/tests/test_service.py
+++ b/tests/test_service.py
@@ -3,7 +3,7 @@
 import bentoml
 from bentoml.artifact import PickleArtifact
 from bentoml.adapters import DataframeInput, FastaiImageInput, ImageInput
-from bentoml.service import _validate_version_str
+from bentoml.service import validate_version_str
 from bentoml.exceptions import InvalidArgument
 
 
@@ -40,40 +40,46 @@ def test(self, image):
     assert 'fastai' in service._env._pip_dependencies
 
 
+# noinspection PyUnusedLocal
 def test_invalid_artifact_type():
     with pytest.raises(InvalidArgument) as e:
 
-        @bentoml.artifacts(["Not A Artifact"])  # pylint: disable=unused-variable
-        class ExampleBentoService(bentoml.BentoService):
+        @bentoml.artifacts(["Not A Artifact"])
+        class ExampleBentoService(  # pylint: disable=unused-variable
+            bentoml.BentoService
+        ):
             pass
 
     assert "only accept list of BentoServiceArtifact" in str(e.value)
 
 
+# noinspection PyUnusedLocal
 def test_duplicated_artifact_name():
     with pytest.raises(InvalidArgument) as e:
 
-        @bentoml.artifacts(  # pylint: disable=unused-variable
-            [PickleArtifact("model"), PickleArtifact("model")]
-        )
-        class ExampleBentoService(bentoml.BentoService):
+        @bentoml.artifacts([PickleArtifact("model"), PickleArtifact("model")])
+        class ExampleBentoService(  # pylint: disable=unused-variable
+            bentoml.BentoService
+        ):
             pass
 
     assert "Duplicated artifact name `model` detected" in str(e.value)
 
 
+# noinspection PyUnusedLocal
 def test_invalid_api_input():
     with pytest.raises(InvalidArgument) as e:
 
-        class ExampleBentoService(
+        class ExampleBentoService(  # pylint: disable=unused-variable
             bentoml.BentoService
-        ):  # pylint: disable=unused-variable
+        ):
             @bentoml.api("Not A InputAdapter")
             def test(self):
                 pass
 
-    assert "must be class derived from bentoml.adapters.BaseInputAdapter" in str(
-        e.value
+    assert (
+        "must be an instance of a class derived from "
+        "bentoml.adapters.BaseInputAdapter" in str(e.value)
     )
 
 
@@ -89,20 +95,20 @@ def test(self, images):
 
 def test_validate_version_str_fails():
     with pytest.raises(InvalidArgument):
-        _validate_version_str("44&")
+        validate_version_str("44&")
 
     with pytest.raises(InvalidArgument):
-        _validate_version_str("44 123")
+        validate_version_str("44 123")
 
     with pytest.raises(InvalidArgument):
-        _validate_version_str("")
+        validate_version_str("")
 
 
 def test_validate_version_str_pass():
-    _validate_version_str("abc_123")
-    _validate_version_str("1")
-    _validate_version_str("a_valid_version")
-    _validate_version_str("AValidVersion")
-    _validate_version_str("_AValidVersion")
-    _validate_version_str("1.3.4")
-    _validate_version_str("1.3.4-g375a71b")
+    validate_version_str("abc_123")
+    validate_version_str("1")
+    validate_version_str("a_valid_version")
+    validate_version_str("AValidVersion")
+    validate_version_str("_AValidVersion")
+    validate_version_str("1.3.4")
+    validate_version_str("1.3.4-g375a71b")


Code review comments:
diff_hunk: @@ -91,7 +91,7 @@ def parse_cli_input(cli_args: Iterable[str]) -> Iterator[bytes]:
     input_g.add_argument('--input', nargs="+", type=str)
     input_g.add_argument('--input-file', nargs="+")
 
-    parsed_args, _ = parser.parse_known_args(cli_args)
+    parsed_args, _ = parser.parse_known_args(list(cli_args))
path: bentoml/adapters/base_input.py
text: @user1:
@author why list here?

@author:
The function signature takes an `Optional[Sequence[str]]`, but the variable is an `Iterable[str]` Maybe the issue is with the function signature.