Repository: SeldonIO/MLServer (language: python)

Issue:
Simplify interface to write custom runtimes
When writing custom runtimes, MLServer requires the user to know a bit about the V2 request and response structure, as well as codecs. This can add some friction when it comes to writing custom code. 

When working in [Tempo](https://github.com/SeldonIO/tempo), one of the main aims was to simplify how “custom models” were implemented. Therefore, it would be interesting to revisit some of these ideas to help simplify MLServer runtimes.

With this goal in mind, it would be interesting to let the user write runtimes like the one below:

```python
class MyCustomRuntime(MLModel):
    # Use signatures to declare the expected request and response content types
    async def predict(self, payload: np.array) -> np.array:
        # Do something with the payload
        pred = payload.sum(keepdims=1)

        # Return result as is
        return pred
```

Under the hood, MLServer could look at the Python type hints to determine the right content types that should be used, and add them to the model metadata. This would allow MLServer to, on-the-fly, find the right set of codecs and encode / decode the request and response. 

## Multiple Inputs / Outputs

The V2 protocol lets the user send and return multiple input and output “heads”. To account for this, the new simplified interface could use the names of the function arguments to match them with the incoming request:

```python
class MyCustomRuntime(MLModel):
    # Input `foo` would get passed as kwarg `foo`, and input `bar` as kwarg `bar`
    async def predict(self, foo: np.array, bar: np.array) -> np.array:
            ...
```

When encoding back the response though, the signature itself doesn’t show any info about the expected names. Therefore, MLServer will need to:

- Look at the `outputs` field of the model metadata, and match the returned values based on their order.
- In the absence of the above, make up some default names (e.g. `output-0`, `output-1`, etc.).

Note that this doesn’t include function signatures with a single “multi-input/-output” return type, like a Pandas Dataframe. In these cases, MLServer will look at the column names to infer the right output names. 

## Variable Content Types & Advanced Use Cases

In some cases, the approach suggested above may show some limitations though. For example, if we think of runtimes which accept a variable set of content types, like the MLflow or SKLearn runtimes (which accept either dataframes or numpy arrays), or other advanced use cases. 

However, these use cases should still be supported by the current approach which would remain valid. That is, when using the “low-level” `InferenceRequest` and `InferenceResponse` types (which correspond to the respective V2 payloads), the encoding / decoding will be the user’s responsibility (who will need to call something like `self.decode`). 


Discussion / hints:

Fix patch:
diff --git a/Makefile b/Makefile
index 54f6c6b5f..4e14bad07 100644
--- a/Makefile
+++ b/Makefile
@@ -60,8 +60,8 @@ lint: generate
 	mypy ./mlserver
 	for _runtime in ./runtimes/*; \
 	do \
-		mypy $$_runtime; \
-	done || exit 1 # Return error if mypy failed
+		mypy $$_runtime || exit 1; \
+	done
 	mypy ./benchmarking
 	mypy ./docs/examples
 	# Check if something has changed after generation
diff --git a/docs/examples/custom/README.ipynb b/docs/examples/custom/README.ipynb
index 7b94ebceb..dd72c685c 100644
--- a/docs/examples/custom/README.ipynb
+++ b/docs/examples/custom/README.ipynb
@@ -151,14 +151,14 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "%%writefile models.py\n",
+    "# %load models.py\n",
     "import json\n",
     "import numpyro\n",
     "import numpy as np\n",
     "\n",
-    "from typing import Dict\n",
     "from jax import random\n",
-    "from mlserver import MLModel, types\n",
+    "from mlserver import MLModel\n",
+    "from mlserver.codecs import decode_args\n",
     "from mlserver.utils import get_model_uri\n",
     "from numpyro.infer import Predictive\n",
     "from numpyro import distributions as dist\n",
@@ -179,33 +179,21 @@
     "        self.ready = True\n",
     "        return self.ready\n",
     "\n",
-    "    async def predict(self, payload: types.InferenceRequest) -> types.InferenceResponse:\n",
-    "        inputs = self._extract_inputs(payload)\n",
-    "        predictions = self._predictive(rng_key=random.PRNGKey(0), **inputs)\n",
+    "    @decode_args\n",
+    "    async def predict(\n",
+    "        self,\n",
+    "        marriage: np.ndarray = None,\n",
+    "        age: np.ndarray = None,\n",
+    "        divorce: np.ndarray = None,\n",
+    "    ) -> np.ndarray:\n",
+    "        predictions = self._predictive(\n",
+    "            rng_key=random.PRNGKey(0), marriage=marriage, age=age, divorce=divorce\n",
+    "        )\n",
     "\n",
     "        obs = predictions[\"obs\"]\n",
     "        obs_mean = obs.mean()\n",
     "\n",
-    "        return types.InferenceResponse(\n",
-    "            id=payload.id,\n",
-    "            model_name=self.name,\n",
-    "            model_version=self.version,\n",
-    "            outputs=[\n",
-    "                types.ResponseOutput(\n",
-    "                    name=\"obs_mean\",\n",
-    "                    shape=obs_mean.shape,\n",
-    "                    datatype=\"FP32\",\n",
-    "                    data=np.asarray(obs_mean).tolist(),\n",
-    "                )\n",
-    "            ],\n",
-    "        )\n",
-    "\n",
-    "    def _extract_inputs(self, payload: types.InferenceRequest) -> Dict[str, np.ndarray]:\n",
-    "        inputs = {}\n",
-    "        for inp in payload.inputs:\n",
-    "            inputs[inp.name] = np.array(inp.data)\n",
-    "\n",
-    "        return inputs\n",
+    "        return np.asarray(obs_mean)\n",
     "\n",
     "    def _model(self, marriage=None, age=None, divorce=None):\n",
     "        a = numpyro.sample(\"a\", dist.Normal(0.0, 0.2))\n",
@@ -218,7 +206,7 @@
     "            A = bA * age\n",
     "        sigma = numpyro.sample(\"sigma\", dist.Exponential(1.0))\n",
     "        mu = a + M + A\n",
-    "        numpyro.sample(\"obs\", dist.Normal(mu, sigma), obs=divorce)"
+    "        numpyro.sample(\"obs\", dist.Normal(mu, sigma), obs=divorce)\n"
    ]
   },
   {
@@ -246,10 +234,10 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "%%writefile settings.json\n",
+    "# %load settings.json\n",
     "{\n",
     "    \"debug\": \"true\"\n",
-    "}"
+    "}\n"
    ]
   },
   {
@@ -265,14 +253,14 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "%%writefile model-settings.json\n",
+    "# %load model-settings.json\n",
     "{\n",
     "    \"name\": \"numpyro-divorce\",\n",
     "    \"implementation\": \"models.NumpyroModel\",\n",
     "    \"parameters\": {\n",
     "        \"uri\": \"./numpyro-divorce.json\"\n",
     "    }\n",
-    "}"
+    "}\n"
    ]
   },
   {
@@ -310,14 +298,20 @@
    "outputs": [],
    "source": [
     "import requests\n",
+    "import numpy as np\n",
+    "\n",
+    "from mlserver.types import InferenceRequest\n",
+    "from mlserver.codecs import NumpyCodec\n",
     "\n",
-    "x_0 = [28.0]\n",
-    "inference_request = {\n",
-    "    \"inputs\": [{\"name\": \"marriage\", \"shape\": [1], \"datatype\": \"FP32\", \"data\": x_0}]\n",
-    "}\n",
+    "x_0 = np.array([28.0])\n",
+    "inference_request = InferenceRequest(\n",
+    "    inputs=[\n",
+    "        NumpyCodec.encode_input(name=\"marriage\", payload=x_0)\n",
+    "    ]\n",
+    ")\n",
     "\n",
     "endpoint = \"http://localhost:8080/v2/models/numpyro-divorce/infer\"\n",
-    "response = requests.post(endpoint, json=inference_request)\n",
+    "response = requests.post(endpoint, json=inference_request.dict())\n",
     "\n",
     "response.json()"
    ]
@@ -346,10 +340,11 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "%%writefile requirements.txt\n",
-    "numpy==1.22.0\n",
+    "# %load requirements.txt\n",
+    "numpy==1.22.4\n",
     "numpyro==0.8.0\n",
-    "jax==0.2.24"
+    "jax==0.2.24\n",
+    "jaxlib==0.3.7\n"
    ]
   },
   {
@@ -403,18 +398,22 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "import requests\n",
+    "import numpy as np\n",
     "\n",
-    "x_0 = [28.0]\n",
-    "inference_request = {\n",
-    "    \"inputs\": [{\"name\": \"marriage\", \"shape\": [1], \"datatype\": \"FP32\", \"data\": x_0}]\n",
-    "}\n",
+    "from mlserver.types import InferenceRequest\n",
+    "from mlserver.codecs import NumpyCodec\n",
+    "\n",
+    "x_0 = np.array([28.0])\n",
+    "inference_request = InferenceRequest(\n",
+    "    inputs=[\n",
+    "        NumpyCodec.encode_input(name=\"marriage\", payload=x_0)\n",
+    "    ]\n",
+    ")\n",
     "\n",
     "endpoint = \"http://localhost:8080/v2/models/numpyro-divorce/infer\"\n",
-    "response = requests.post(endpoint, json=inference_request)\n",
+    "response = requests.post(endpoint, json=inference_request.dict())\n",
     "\n",
-    "print(response)\n",
-    "print(response.text)"
+    "response.json()"
    ]
   },
   {
@@ -481,7 +480,7 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "Python 3.8.10 64-bit",
+   "display_name": "Python 3 (ipykernel)",
    "language": "python",
    "name": "python3"
   },
@@ -495,7 +494,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.8.10"
+   "version": "3.9.8"
   },
   "vscode": {
    "interpreter": {
diff --git a/docs/examples/custom/README.md b/docs/examples/custom/README.md
index 5cd8cfed8..11671dd5e 100644
--- a/docs/examples/custom/README.md
+++ b/docs/examples/custom/README.md
@@ -33,27 +33,29 @@ from numpyro import distributions as dist
 from jax import random
 from numpyro.infer import MCMC, NUTS
 
-DATASET_URL = 'https://raw.githubusercontent.com/rmcelreath/rethinking/master/data/WaffleDivorce.csv'
-dset = pd.read_csv(DATASET_URL, sep=';')
+DATASET_URL = "https://raw.githubusercontent.com/rmcelreath/rethinking/master/data/WaffleDivorce.csv"
+dset = pd.read_csv(DATASET_URL, sep=";")
 
 standardize = lambda x: (x - x.mean()) / x.std()
 
-dset['AgeScaled'] = dset.MedianAgeMarriage.pipe(standardize)
-dset['MarriageScaled'] = dset.Marriage.pipe(standardize)
-dset['DivorceScaled'] = dset.Divorce.pipe(standardize)
+dset["AgeScaled"] = dset.MedianAgeMarriage.pipe(standardize)
+dset["MarriageScaled"] = dset.Marriage.pipe(standardize)
+dset["DivorceScaled"] = dset.Divorce.pipe(standardize)
+
 
 def model(marriage=None, age=None, divorce=None):
-    a = numpyro.sample('a', dist.Normal(0., 0.2))
-    M, A = 0., 0.
+    a = numpyro.sample("a", dist.Normal(0.0, 0.2))
+    M, A = 0.0, 0.0
     if marriage is not None:
-        bM = numpyro.sample('bM', dist.Normal(0., 0.5))
+        bM = numpyro.sample("bM", dist.Normal(0.0, 0.5))
         M = bM * marriage
     if age is not None:
-        bA = numpyro.sample('bA', dist.Normal(0., 0.5))
+        bA = numpyro.sample("bA", dist.Normal(0.0, 0.5))
         A = bA * age
-    sigma = numpyro.sample('sigma', dist.Exponential(1.))
+    sigma = numpyro.sample("sigma", dist.Exponential(1.0))
     mu = a + M + A
-    numpyro.sample('obs', dist.Normal(mu, sigma), obs=divorce)
+    numpyro.sample("obs", dist.Normal(mu, sigma), obs=divorce)
+
 
 # Start from this source of randomness. We will split keys for subsequent operations.
 rng_key = random.PRNGKey(0)
@@ -64,7 +66,9 @@ num_warmup, num_samples = 1000, 2000
 # Run NUTS.
 kernel = NUTS(model)
 mcmc = MCMC(kernel, num_warmup=num_warmup, num_samples=num_samples)
-mcmc.run(rng_key_, marriage=dset.MarriageScaled.values, divorce=dset.DivorceScaled.values)
+mcmc.run(
+    rng_key_, marriage=dset.MarriageScaled.values, divorce=dset.DivorceScaled.values
+)
 mcmc.print_summary()
 ```
 
@@ -83,9 +87,9 @@ samples = mcmc.get_samples()
 serialisable = {}
 for k, v in samples.items():
     serialisable[k] = np.asarray(v).tolist()
-    
+
 model_file_name = "numpyro-divorce.json"
-with open(model_file_name, 'w') as model_file:
+with open(model_file_name, "w") as model_file:
     json.dump(serialisable, model_file)
 ```
 
@@ -104,14 +108,14 @@ Our custom inference wrapper should be responsible of:
 
 
 ```python
-%%writefile models.py
+# %load models.py
 import json
 import numpyro
 import numpy as np
 
-from typing import Dict
 from jax import random
-from mlserver import MLModel, types
+from mlserver import MLModel
+from mlserver.codecs import decode_args
 from mlserver.utils import get_model_uri
 from numpyro.infer import Predictive
 from numpyro import distributions as dist
@@ -132,33 +136,21 @@ class NumpyroModel(MLModel):
         self.ready = True
         return self.ready
 
-    async def predict(self, payload: types.InferenceRequest) -> types.InferenceResponse:
-        inputs = self._extract_inputs(payload)
-        predictions = self._predictive(rng_key=random.PRNGKey(0), **inputs)
+    @decode_args
+    async def predict(
+        self,
+        marriage: np.ndarray = None,
+        age: np.ndarray = None,
+        divorce: np.ndarray = None,
+    ) -> np.ndarray:
+        predictions = self._predictive(
+            rng_key=random.PRNGKey(0), marriage=marriage, age=age, divorce=divorce
+        )
 
         obs = predictions["obs"]
         obs_mean = obs.mean()
 
-        return types.InferenceResponse(
-            id=payload.id,
-            model_name=self.name,
-            model_version=self.version,
-            outputs=[
-                types.ResponseOutput(
-                    name="obs_mean",
-                    shape=obs_mean.shape,
-                    datatype="FP32",
-                    data=np.asarray(obs_mean).tolist(),
-                )
-            ],
-        )
-
-    def _extract_inputs(self, payload: types.InferenceRequest) -> Dict[str, np.ndarray]:
-        inputs = {}
-        for inp in payload.inputs:
-            inputs[inp.name] = np.array(inp.data)
-
-        return inputs
+        return np.asarray(obs_mean)
 
     def _model(self, marriage=None, age=None, divorce=None):
         a = numpyro.sample("a", dist.Normal(0.0, 0.2))
@@ -172,6 +164,7 @@ class NumpyroModel(MLModel):
         sigma = numpyro.sample("sigma", dist.Exponential(1.0))
         mu = a + M + A
         numpyro.sample("obs", dist.Normal(mu, sigma), obs=divorce)
+
 ```
 
 ### Settings files
@@ -185,17 +178,18 @@ The next step will be to create 2 configuration files:
 
 
 ```python
-%%writefile settings.json
+# %load settings.json
 {
     "debug": "true"
 }
+
 ```
 
 #### `model-settings.json`
 
 
 ```python
-%%writefile model-settings.json
+# %load model-settings.json
 {
     "name": "numpyro-divorce",
     "implementation": "models.NumpyroModel",
@@ -203,6 +197,7 @@ The next step will be to create 2 configuration files:
         "uri": "./numpyro-divorce.json"
     }
 }
+
 ```
 
 ### Start serving our model
@@ -226,21 +221,20 @@ For that, we can use the Python types that `mlserver` provides out of box, or we
 
 ```python
 import requests
+import numpy as np
+
+from mlserver.types import InferenceRequest
+from mlserver.codecs import NumpyCodec
 
-x_0 = [28.0]
-inference_request = {
-    "inputs": [
-        {
-          "name": "marriage",
-          "shape": [1],
-          "datatype": "FP32",
-          "data": x_0
-        }
+x_0 = np.array([28.0])
+inference_request = InferenceRequest(
+    inputs=[
+        NumpyCodec.encode_input(name="marriage", payload=x_0)
     ]
-}
+)
 
 endpoint = "http://localhost:8080/v2/models/numpyro-divorce/infer"
-response = requests.post(endpoint, json=inference_request)
+response = requests.post(endpoint, json=inference_request.dict())
 
 response.json()
 ```
@@ -256,10 +250,12 @@ MLServer will automatically find your requirements.txt file and install necessar
 
 
 ```python
-%%writefile requirements.txt
-numpy==1.22.0
+# %load requirements.txt
+numpy==1.22.4
 numpyro==0.8.0
 jax==0.2.24
+jaxlib==0.3.7
+
 ```
 
 ### Building a custom image
@@ -288,26 +284,22 @@ docker run -it --rm -p 8080:8080 my-custom-numpyro-server:0.1.0
 
 
 ```python
-import requests
+import numpy as np
+
+from mlserver.types import InferenceRequest
+from mlserver.codecs import NumpyCodec
 
-x_0 = [28.0]
-inference_request = {
-    "inputs": [
-        {
-          "name": "marriage",
-          "shape": [1],
-          "datatype": "FP32",
-          "data": x_0
-        }
+x_0 = np.array([28.0])
+inference_request = InferenceRequest(
+    inputs=[
+        NumpyCodec.encode_input(name="marriage", payload=x_0)
     ]
-}
+)
 
 endpoint = "http://localhost:8080/v2/models/numpyro-divorce/infer"
-response = requests.post(endpoint, json=inference_request)
-
-print(response)
-print(response.text)
+response = requests.post(endpoint, json=inference_request.dict())
 
+response.json()
 ```
 
 As we should be able to see, the server running within our Docker image responds as expected.
@@ -323,7 +315,7 @@ There is a large number of tools out there to deploy images.
 However, for our example, we will focus on deploying it to a cluster running [Seldon Core](https://docs.seldon.io/projects/seldon-core/en/latest/).
 
 ```{note}
-Also consider that depending on your Kubernetes installation Seldon Core might excpect to get the container image from a public container registry like [Docker hub](https://hub.docker.com/) or [Google Container Registry](https://cloud.google.com/container-registry). For that you need to do an extra step of pushing the container to the registry using `docker tag <image name> <container registry>/<image name>` and `docker push <container registry>/<image name>` and also updating the `image` section of the yaml file to `<container registry>/<image name>`. 
+Also consider that depending on your Kubernetes installation Seldon Core might expect to get the container image from a public container registry like [Docker hub](https://hub.docker.com/) or [Google Container Registry](https://cloud.google.com/container-registry). For that you need to do an extra step of pushing the container to the registry using `docker tag <image name> <container registry>/<image name>` and `docker push <container registry>/<image name>` and also updating the `image` section of the yaml file to `<container registry>/<image name>`. 
 ```
 
 For that, we will need to create a `SeldonDeployment` resource which instructs Seldon Core to deploy a model embedded within our custom image and compliant with the [V2 Inference Protocol](https://github.com/kserve/kserve/tree/master/docs/predict-api/v2).
diff --git a/docs/examples/custom/models.py b/docs/examples/custom/models.py
index 2d9ae1fdb..e30ec3a37 100644
--- a/docs/examples/custom/models.py
+++ b/docs/examples/custom/models.py
@@ -2,9 +2,9 @@
 import numpyro
 import numpy as np
 
-from typing import Dict
 from jax import random
-from mlserver import MLModel, types
+from mlserver import MLModel
+from mlserver.codecs import decode_args
 from mlserver.utils import get_model_uri
 from numpyro.infer import Predictive
 from numpyro import distributions as dist
@@ -25,33 +25,21 @@ async def load(self) -> bool:
         self.ready = True
         return self.ready
 
-    async def predict(self, payload: types.InferenceRequest) -> types.InferenceResponse:
-        inputs = self._extract_inputs(payload)
-        predictions = self._predictive(rng_key=random.PRNGKey(0), **inputs)
+    @decode_args
+    async def predict(
+        self,
+        marriage: np.ndarray = None,
+        age: np.ndarray = None,
+        divorce: np.ndarray = None,
+    ) -> np.ndarray:
+        predictions = self._predictive(
+            rng_key=random.PRNGKey(0), marriage=marriage, age=age, divorce=divorce
+        )
 
         obs = predictions["obs"]
         obs_mean = obs.mean()
 
-        return types.InferenceResponse(
-            id=payload.id,
-            model_name=self.name,
-            model_version=self.version,
-            outputs=[
-                types.ResponseOutput(
-                    name="obs_mean",
-                    shape=obs_mean.shape,
-                    datatype="FP32",
-                    data=np.asarray(obs_mean).tolist(),
-                )
-            ],
-        )
-
-    def _extract_inputs(self, payload: types.InferenceRequest) -> Dict[str, np.ndarray]:
-        inputs = {}
-        for inp in payload.inputs:
-            inputs[inp.name] = np.array(inp.data)
-
-        return inputs
+        return np.asarray(obs_mean)
 
     def _model(self, marriage=None, age=None, divorce=None):
         a = numpyro.sample("a", dist.Normal(0.0, 0.2))
diff --git a/docs/examples/custom/requirements.txt b/docs/examples/custom/requirements.txt
index 64ff4a513..fe4a9486f 100644
--- a/docs/examples/custom/requirements.txt
+++ b/docs/examples/custom/requirements.txt
@@ -1,3 +1,4 @@
-numpy==1.22.0
+numpy==1.22.4
 numpyro==0.8.0
 jax==0.2.24
+jaxlib==0.3.7
diff --git a/docs/user-guide/custom.md b/docs/user-guide/custom.md
index 56d12f9c7..eb0bc169f 100644
--- a/docs/user-guide/custom.md
+++ b/docs/user-guide/custom.md
@@ -47,6 +47,72 @@ class MyCustomRuntime(MLModel):
     return self._model.predict(payload)
 ```
 
+### Simplified interface
+
+MLServer exposes an alternative _"simplified" interface_ which can be used to
+write custom runtimes.
+This interface can be enabled by decorating your `predict()` method with the
+`mlserver.codecs.decode_args` decorator, and it lets you specify in the method
+signature both how you want your request payload to be decoded and how to
+encode the response back.
+
+Based on the information provided in the method signature, MLServer will
+automatically decode the request payload into the different inputs specified as
+keyword arguments.
+Under the hood, this is implemented through [MLServer's codecs and content types
+system](./content-type.md).
+
+```{note}
+MLServer's _"simplified" interface_ aims to cover use cases where encoding /
+decoding can be done through one of the codecs built-in into the MLServer
+package.
+However, there are instances where this may not be enough (e.g. variable number
+of inputs, variable content types, etc.).
+For these types of cases, please use MLServer's [_"advanced"
+interface_](#writing-a-custom-inference-runtime), where you will have full
+control over the full encoding / decoding process.
+```
+
+As an example of the above, let's assume a model which
+
+- Takes two lists of strings as inputs:
+  - `questions`, containing multiple questions to ask our model.
+  - `context`, containing multiple contexts for each of the
+    questions.
+- Returns a Numpy array with some predictions as the output.
+
+Leveraging MLServer's simplified notation, we can represent the above as the
+following custom runtime:
+
+```{code-block} python
+---
+emphasize-lines: 2, 12-13
+---
+from mlserver import MLModel
+from mlserver.codecs import decode_args
+
+class MyCustomRuntime(MLModel):
+
+  async def load(self) -> bool:
+    # TODO: Replace for custom logic to load a model artifact
+    self._model = load_my_custom_model()
+    self.ready = True
+    return self.ready
+
+  @decode_args
+  async def predict(self, questions: List[str], context: List[str]) -> np.ndarray:
+    # TODO: Replace for custom logic to run inference
+    return self._model.predict(questions, context)
+```
+
+Note that, the method signature of our `predict` method now specifies:
+
+- The input names that we should be looking for in the request
+  payload (i.e. `questions` and `context`).
+- The expected content type for each of the request inputs (i.e. `List[str]` on
+  both cases).
+- The expected content type of the response outputs (i.e. `np.ndarray`).
+
 ### Read and write headers
 
 ```{note}
diff --git a/mlserver/codecs/__init__.py b/mlserver/codecs/__init__.py
index f4d65a457..754bb5e85 100644
--- a/mlserver/codecs/__init__.py
+++ b/mlserver/codecs/__init__.py
@@ -4,11 +4,14 @@
 from .base64 import Base64Codec
 from .datetime import DatetimeCodec
 from .errors import CodecError
+from .decorator import decode_args
 from .base import (
     InputCodec,
     RequestCodec,
     register_input_codec,
     register_request_codec,
+    InputCodecLike,
+    RequestCodecLike,
 )
 from .utils import (
     DecodedParameterName,
@@ -19,8 +22,6 @@
     encode_response_output,
     decode_request_input,
     decode_inference_request,
-    InputCodecLike,
-    RequestCodecLike,
 )
 
 __all__ = [
@@ -45,4 +46,5 @@
     "encode_response_output",
     "decode_request_input",
     "decode_inference_request",
+    "decode_args",
 ]
diff --git a/mlserver/codecs/base.py b/mlserver/codecs/base.py
index a7bc750f5..7f2e291ae 100644
--- a/mlserver/codecs/base.py
+++ b/mlserver/codecs/base.py
@@ -5,6 +5,9 @@
 from ..types import InferenceRequest, InferenceResponse, RequestInput, ResponseOutput
 from ..logging import logger
 
+InputCodecLike = Union[Type["InputCodec"], "InputCodec"]
+RequestCodecLike = Union[Type["RequestCodec"], "RequestCodec"]
+
 
 def deprecated(reason: str):
     def _deprecated(func):
@@ -28,6 +31,7 @@ class InputCodec:
     """
 
     ContentType: ClassVar[str] = ""
+    TypeHint: ClassVar[Type] = type(None)
 
     @classmethod
     def can_encode(cls, payload: Any) -> bool:
@@ -69,7 +73,8 @@ class RequestCodec:
     interface instead.
     """
 
-    ContentType: Optional[str] = None
+    ContentType: ClassVar[str] = ""
+    TypeHint: ClassVar[Type] = type(None)
 
     @classmethod
     def can_encode(cls, payload: Any) -> bool:
@@ -123,7 +128,30 @@ def _find_codec_by_payload(
         logger.warning(
             f"Multiple codecs ({len(matching_codecs)} were found for a "
             f"payload with type {type(payload)}. "
-            f"The first one found will be used ({first_codec.ContentType})"
+            f"The first one found will be used ({first_codec.ContentType})."
+        )
+
+    return first_codec
+
+
+def _find_codec_by_type_hint(
+    type_hint: Type, codecs: Iterable[Union[RequestCodec, InputCodec]]
+) -> Optional[Union[RequestCodec, InputCodec]]:
+    matching_codecs = []
+    for codec in codecs:
+        if codec.TypeHint == type_hint:
+            matching_codecs.append(codec)
+
+    if len(matching_codecs) == 0:
+        logger.warning(f"No codec was found for type hint {type_hint}.")
+        return None
+
+    first_codec = matching_codecs[0]
+    if len(matching_codecs) > 1:
+        logger.warning(
+            f"Multiple codecs ({len(matching_codecs)} were found for a "
+            f"type hint {type_hint}. "
+            f"The first one found will be used ({first_codec.ContentType})."
         )
 
     return first_codec
@@ -138,56 +166,88 @@ class _CodecRegistry:
 
     def __init__(
         self,
-        input_codecs: Dict[str, Type[InputCodec]] = {},
-        request_codecs: Dict[str, Type[RequestCodec]] = {},
+        input_codecs: Dict[str, InputCodecLike] = {},
+        request_codecs: Dict[str, RequestCodecLike] = {},
     ):
         self._input_codecs = input_codecs
         self._request_codecs = request_codecs
 
-    def register_input_codec(self, content_type: str, codec: Type[InputCodec]):
+    def register_input_codec(self, content_type: str, codec: InputCodecLike):
         # TODO: Raise error if codec exists?
         self._input_codecs[content_type] = codec
 
-    def find_input_codec(self, content_type: str) -> Type[InputCodec]:
-        # TODO: Raise error if codec doesn't exist
-        return self._input_codecs[content_type]
+    def find_input_codec(
+        self, content_type: str = None, payload: Any = None, type_hint: Type = None
+    ) -> Optional[InputCodecLike]:
+        if content_type:
+            # TODO: Raise error if codec doesn't exist
+            return self._input_codecs[content_type]
+        elif payload:
+            return self.find_input_codec_by_payload(payload)
+        elif type_hint:
+            return self.find_input_codec_by_type_hint(type_hint)
+
+        return None
 
-    def find_input_codec_by_payload(self, payload: Any) -> Optional[Type[InputCodec]]:
+    def find_input_codec_by_payload(self, payload: Any) -> Optional[InputCodecLike]:
         return _find_codec_by_payload(
             payload, self._input_codecs.values()  # type: ignore
         )
 
-    def register_request_codec(self, content_type: str, codec: Type[RequestCodec]):
+    def find_input_codec_by_type_hint(
+        self, type_hint: Type
+    ) -> Optional[InputCodecLike]:
+        return _find_codec_by_type_hint(
+            type_hint, self._input_codecs.values()  # type: ignore
+        )
+
+    def register_request_codec(self, content_type: str, codec: RequestCodecLike):
         # TODO: Raise error if codec exists?
         self._request_codecs[content_type] = codec
 
-    def find_request_codec(self, content_type: str) -> Type[RequestCodec]:
-        # TODO: Raise error if codec doesn't exist
-        return self._request_codecs[content_type]
+    def find_request_codec(
+        self, content_type: str = None, payload: Any = None, type_hint: Type = None
+    ) -> Optional[RequestCodecLike]:
+        if content_type:
+            # TODO: Raise error if codec doesn't exist
+            return self._request_codecs[content_type]
+        elif payload:
+            return self.find_request_codec_by_payload(payload)
+        elif type_hint:
+            return self.find_request_codec_by_type_hint(type_hint)
+
+        return None
 
-    def find_request_codec_by_payload(
-        self, payload: Any
-    ) -> Optional[Type[RequestCodec]]:
+    def find_request_codec_by_payload(self, payload: Any) -> Optional[RequestCodecLike]:
         return _find_codec_by_payload(
             payload, self._request_codecs.values()  # type: ignore
         )
 
+    def find_request_codec_by_type_hint(
+        self, type_hint: Type
+    ) -> Optional[RequestCodecLike]:
+        return _find_codec_by_type_hint(
+            type_hint, self._request_codecs.values()  # type: ignore
+        )
+
 
 _codec_registry = _CodecRegistry()
 
 find_request_codec = _codec_registry.find_request_codec
 find_request_codec_by_payload = _codec_registry.find_request_codec_by_payload
+find_request_codec_by_type_hint = _codec_registry.find_request_codec_by_type_hint
 find_input_codec = _codec_registry.find_input_codec
 find_input_codec_by_payload = _codec_registry.find_input_codec_by_payload
+find_input_codec_by_type_hint = _codec_registry.find_input_codec_by_type_hint
 
 
-def register_request_codec(CodecKlass: Type[RequestCodec]):
+def register_request_codec(CodecKlass: RequestCodecLike):
     if CodecKlass.ContentType is not None:
         _codec_registry.register_request_codec(CodecKlass.ContentType, CodecKlass)
     return CodecKlass
 
 
-def register_input_codec(CodecKlass: Type[InputCodec]):
+def register_input_codec(CodecKlass: InputCodecLike):
     if CodecKlass.ContentType is not None:
         _codec_registry.register_input_codec(CodecKlass.ContentType, CodecKlass)
     return CodecKlass
diff --git a/mlserver/codecs/base64.py b/mlserver/codecs/base64.py
index c32284848..f4ec4b339 100644
--- a/mlserver/codecs/base64.py
+++ b/mlserver/codecs/base64.py
@@ -46,6 +46,7 @@ class Base64Codec(InputCodec):
     """
 
     ContentType = "base64"
+    TypeHint = List[bytes]
 
     @classmethod
     def can_encode(cls, payload: Any) -> bool:
diff --git a/mlserver/codecs/datetime.py b/mlserver/codecs/datetime.py
index b7cc56a4a..b6f9ce701 100644
--- a/mlserver/codecs/datetime.py
+++ b/mlserver/codecs/datetime.py
@@ -46,6 +46,7 @@ class DatetimeCodec(InputCodec):
     """
 
     ContentType = "datetime"
+    TypeHint = List[_Datetime]
 
     @classmethod
     def can_encode(cls, payload: Any) -> bool:
diff --git a/mlserver/codecs/decorator.py b/mlserver/codecs/decorator.py
new file mode 100644
index 000000000..88750af4a
--- /dev/null
+++ b/mlserver/codecs/decorator.py
@@ -0,0 +1,237 @@
+from functools import wraps, partial
+from typing import (
+    Any,
+    Callable,
+    Coroutine,
+    Dict,
+    List,
+    Optional,
+    Union,
+    Type,
+    Tuple,
+    get_type_hints,
+    TYPE_CHECKING,
+)
+
+try:
+    from typing import get_origin, get_args
+except ImportError:
+    # The `get_origin` and `get_args` don't exist in Python 3.7, so we'll
+    # backfill them manually.
+    # When we remove support for Python 3.7, remove these methods.
+    def get_origin(t: Type) -> Type:  # type: ignore
+        return getattr(t, "__origin__", type(None))
+
+    def get_args(t: Type) -> tuple:  # type: ignore
+        return getattr(t, "__args__", ())
+
+
+from ..types import InferenceRequest, InferenceResponse, ResponseOutput
+
+from .base import RequestCodec, InputCodec, find_input_codec, find_request_codec
+from .errors import InputsNotFound, OutputNotFound, CodecNotFound
+from .utils import Codec
+
+if TYPE_CHECKING:
+    from ..model import MLModel
+
+PredictFunc = Callable[
+    ["MLModel", InferenceRequest], Coroutine[Any, Any, InferenceResponse]
+]
+
+
+def _as_list(a: Optional[Union[Any, Tuple[Any]]]) -> List[Any]:
+    if a is None:
+        return []
+
+    if isinstance(a, tuple):
+        # Split into components
+        return list(a)
+
+    # Otherwise, assume it's a single element
+    return [a]
+
+
+def _is_codec_type(c: Codec, t: Type) -> bool:
+    if issubclass(c, t):  # type: ignore
+        return True
+
+    if isinstance(c, t):
+        return True
+
+    return False
+
+
+_is_input_codec = partial(_is_codec_type, t=InputCodec)
+_is_request_codec = partial(_is_codec_type, t=RequestCodec)
+
+
+def _is_optional(t: Type) -> bool:
+    origin = get_origin(t)
+    if origin == Optional:
+        return True
+
+    if origin == Union:
+        # Cover case where Optional[a] is reported as Union[a, None]
+        args = get_args(t)
+        if len(args) == 2 and type(None) in args:
+            return True
+
+    return False
+
+
+def _unwrap_optional(t: Type) -> Type:
+    args = get_args(t)
+    for arg in args:
+        if not isinstance(arg, type(None)):
+            return arg
+
+    return t
+
+
+class SignatureCodec(RequestCodec):
+    """
+    Internal codec that knows how to map type hints to codecs.
+    """
+
+    # TODO: Should this receive the whole class as argument?
+    def __init__(self, predict: Callable):
+        self._predict = predict
+        self._input_codecs, self._output_codecs = self._get_codecs(predict)
+
+    def _get_codecs(self, pred: Callable) -> Tuple[Dict[str, Codec], List[Codec]]:
+        self._input_hints = self._get_type_hints(pred)
+        self._output_hints = _as_list(self._input_hints.pop("return", None))
+
+        input_codecs = {}
+        for name, type_hint in self._input_hints.items():
+            codec = self._find_codec(name=name, type_hint=type_hint, is_input=True)
+            # TODO: Consider metadata as well! (needs to be done at runtime)
+            input_codecs[name] = codec
+
+        output_codecs = []
+        for type_hint in self._output_hints:
+            # Try either as an input or as a request codec
+            codec = self._find_codec(name=None, type_hint=type_hint, is_input=False)
+            output_codecs.append(codec)
+
+        return input_codecs, output_codecs
+
+    def _get_type_hints(self, pred: Callable) -> Dict[str, Type]:
+        type_hints = get_type_hints(pred)
+        # For us, `typing.Optional` is just syntactic sugar, so let's ensure we
+        # unwrap it
+        for name, hint in type_hints.items():
+            if _is_optional(hint):
+                type_hints[name] = _unwrap_optional(hint)
+
+        return type_hints
+
+    def _find_codec(
+        self, name: Optional[str], type_hint: Type, is_input: bool = False
+    ) -> Codec:
+        codec = find_input_codec(type_hint=type_hint)
+        if codec is not None:
+            return codec
+
+        codec = find_request_codec(type_hint=type_hint)  # type: ignore
+        if codec is not None:
+            return codec
+
+        raise CodecNotFound(name=name, payload_type=str(type_hint), is_input=is_input)
+
+    def decode_request(  # type: ignore
+        self, request: InferenceRequest
+    ) -> Dict[str, Any]:
+        inputs = {}
+        extra_request_inputs = []
+        for request_input in request.inputs:
+            input_name = request_input.name
+            if input_name not in self._input_codecs:
+                # Aggregate extra request inputs to check later, as they could
+                # be part of aggregated request codecs (e.g. like dataframes)
+                extra_request_inputs.append(request_input)
+                continue
+
+            # Ensure matching codec is an input codec
+            codec = self._input_codecs[input_name]
+            if not _is_input_codec(codec):
+                raise CodecNotFound(name=input_name, is_input=True)
+
+            inputs[input_name] = codec.decode_input(request_input)  # type: ignore
+
+        if extra_request_inputs:
+            request_codec = self._get_request_codec()
+            if not request_codec:
+                # If there are no request codecs that can aggregate all
+                # remaining inputs, raise an error
+                raise InputsNotFound(extra_request_inputs, self._input_codecs)
+
+            # We create a fake request built from the extra request inputs
+            name, codec = request_codec
+            extra_inputs = InferenceRequest(inputs=extra_request_inputs)
+            inputs[name] = codec.decode_request(extra_inputs)
+
+        return inputs
+
+    def _get_request_codec(self) -> Optional[Tuple[str, RequestCodec]]:
+        for name, codec in self._input_codecs.items():
+            if _is_request_codec(codec):
+                return name, codec  # type: ignore
+
+        return None
+
+    def encode_response(  # type: ignore
+        self, model_name: str, payload: Any, model_version: str = None
+    ) -> InferenceResponse:
+        payloads = _as_list(payload)
+        outputs = []
+        for idx, payload in enumerate(payloads):
+            outputs += self._encode_outputs(idx, payload)
+
+        return InferenceResponse(
+            model_name=model_name, model_version=model_version, outputs=outputs
+        )
+
+    def _encode_outputs(self, idx: int, payload: Any) -> List[ResponseOutput]:
+        output_type = type(payload)
+        if idx >= len(self._output_codecs):
+            raise OutputNotFound(idx, output_type, self._output_hints)
+
+        # TODO: Fallback to encode_by_payload?
+        codec = self._output_codecs[idx]
+        if not codec.can_encode(payload):
+            raise OutputNotFound(idx, output_type, self._output_hints)
+
+        if _is_input_codec(codec):
+            # TODO: Check model metadata for output names
+            output_name = f"output-{idx}"
+            response_output = codec.encode_output(  # type: ignore
+                name=output_name, payload=payload
+            )
+            return [response_output]
+
+        if _is_request_codec(codec):
+            # NOTE: We will ignore `model_name` and only grab the outputs
+            response = codec.encode_response(  # type: ignore
+                model_name="", payload=payload
+            )
+            return response.outputs
+
+        return []
+
+
+def decode_args(predict: Callable) -> PredictFunc:
+    codec = SignatureCodec(predict)
+
+    @wraps(predict)
+    async def _f(self: "MLModel", request: InferenceRequest) -> InferenceResponse:
+        inputs = codec.decode_request(request=request)
+
+        outputs = await predict(self, **inputs)
+
+        return codec.encode_response(
+            model_name=self.name, payload=outputs, model_version=self.version
+        )
+
+    return _f
diff --git a/mlserver/codecs/errors.py b/mlserver/codecs/errors.py
index f6fd97a9f..809b2693b 100644
--- a/mlserver/codecs/errors.py
+++ b/mlserver/codecs/errors.py
@@ -1,3 +1,6 @@
+from typing import Any, Dict, List, Type
+
+from ..types import RequestInput
 from ..errors import MLServerError
 
 
@@ -11,7 +14,7 @@ def __init__(
     ):
         msg = ""
         if name:
-            msg = f"with name {name}"
+            msg = f"with name '{name}'"
 
             if payload_type:
                 msg = f"{msg} and type {payload_type}"
@@ -38,3 +41,24 @@ class CodecError(MLServerError):
     def __init__(self, msg: str):
         msg = f"There was an error encoding / decoding the payload: {msg}"
         super().__init__(msg)
+
+
+class OutputNotFound(MLServerError):
+    def __init__(self, output_idx: int, output_type: Type, output_hints: List[Type]):
+        expected_outputs = [f"'{output_hint}'" for output_hint in output_hints]
+        msg = (
+            f"Unexpected output value at position '{output_idx}' ({output_type}). "
+            f"Expected outputs are {', '.join(expected_outputs)} outputs."
+        )
+        super().__init__(msg)
+
+
+class InputsNotFound(MLServerError):
+    def __init__(self, inputs: List[RequestInput], input_hints: Dict[str, Any]):
+        input_names = [f"'{inp.name}'" for inp in inputs]
+        available_inputs = [f"'{input_name}'" for input_name in input_hints.keys()]
+        msg = (
+            f"Input {', '.join(input_names)} was not found. "
+            f"Available inputs are {', '.join(available_inputs)}."
+        )
+        super().__init__(msg)
diff --git a/mlserver/codecs/numpy.py b/mlserver/codecs/numpy.py
index 558851c47..e0e12ce13 100644
--- a/mlserver/codecs/numpy.py
+++ b/mlserver/codecs/numpy.py
@@ -101,6 +101,7 @@ class NumpyCodec(InputCodec):
     """
 
     ContentType = "np"
+    TypeHint = np.ndarray
 
     @classmethod
     def can_encode(csl, payload: Any) -> bool:
diff --git a/mlserver/codecs/pandas.py b/mlserver/codecs/pandas.py
index ab697254a..fceebbec3 100644
--- a/mlserver/codecs/pandas.py
+++ b/mlserver/codecs/pandas.py
@@ -54,6 +54,7 @@ def _ensure_bytes(elem: PackElement) -> bytes:
 @register_request_codec
 class PandasCodec(RequestCodec):
     ContentType = "pd"
+    TypeHint = pd.DataFrame
 
     @classmethod
     def can_encode(cls, payload: Any) -> bool:
diff --git a/mlserver/codecs/string.py b/mlserver/codecs/string.py
index 13f7bd8d0..66be9442f 100644
--- a/mlserver/codecs/string.py
+++ b/mlserver/codecs/string.py
@@ -40,6 +40,7 @@ class StringCodec(InputCodec):
     """
 
     ContentType = "str"
+    TypeHint = List[str]
 
     @classmethod
     def can_encode(cls, payload: Any) -> bool:
@@ -89,3 +90,4 @@ def encode_input(
 class StringRequestCodec(SingleInputRequestCodec):
     InputCodec = StringCodec
     ContentType = StringCodec.ContentType
+    TypeHint = List[str]
diff --git a/mlserver/codecs/utils.py b/mlserver/codecs/utils.py
index 6c4f90dda..72966ec2d 100644
--- a/mlserver/codecs/utils.py
+++ b/mlserver/codecs/utils.py
@@ -15,8 +15,9 @@
     find_input_codec_by_payload,
     find_request_codec,
     find_request_codec_by_payload,
-    InputCodec,
     RequestCodec,
+    InputCodecLike,
+    RequestCodecLike,
 )
 from .errors import CodecError
 
@@ -24,8 +25,7 @@
 DefaultInputPrefix = "input-"
 
 InputOrOutput = Union[RequestInput, ResponseOutput]
-InputCodecLike = Union[Type[InputCodec], InputCodec]
-RequestCodecLike = Union[Type[RequestCodec], RequestCodec]
+Codec = Union[InputCodecLike, RequestCodecLike]
 
 Parametrised = Union[
     InferenceRequest, RequestInput, RequestOutput, ResponseOutput, InferenceResponse
@@ -175,7 +175,7 @@ class SingleInputRequestCodec(RequestCodec):
     element.
     """
 
-    InputCodec: Optional[Type[InputCodec]] = None
+    InputCodec: Optional[InputCodecLike] = None
 
     @classmethod
     def can_encode(cls, payload: Any) -> bool:
diff --git a/runtimes/sklearn/mlserver_sklearn/sklearn.py b/runtimes/sklearn/mlserver_sklearn/sklearn.py
index 8f730a1fc..4c959ca38 100644
--- a/runtimes/sklearn/mlserver_sklearn/sklearn.py
+++ b/runtimes/sklearn/mlserver_sklearn/sklearn.py
@@ -76,7 +76,8 @@ def _check_request(self, payload: InferenceRequest) -> InferenceRequest:
                     )
 
         # Regression models do not support `predict_proba`
-        if PREDICT_PROBA_OUTPUT in [o.name for o in payload.outputs]:
+        output_names = [o.name for o in payload.outputs]  # type: ignore
+        if PREDICT_PROBA_OUTPUT in output_names:
             # Ensure model supports it
             maybe_regressor = self._model
             if isinstance(self._model, Pipeline):


Test patch:
diff --git a/runtimes/alibi-explain/tests/conftest.py b/runtimes/alibi-explain/tests/conftest.py
index 31eeda2d9..766abe231 100644
--- a/runtimes/alibi-explain/tests/conftest.py
+++ b/runtimes/alibi-explain/tests/conftest.py
@@ -35,7 +35,7 @@
 
 
 @pytest.fixture
-async def inference_pool(settings: Settings) -> InferencePool:
+async def inference_pool(settings: Settings) -> AsyncIterable[InferencePool]:
     pool = InferencePool(settings)
     yield pool
 
@@ -72,7 +72,7 @@ def settings() -> Settings:
 @pytest.fixture
 async def model_registry(
     custom_runtime_tf_settings, inference_pool
-) -> MultiModelRegistry:
+) -> AsyncIterable[MultiModelRegistry]:
     model_registry = MultiModelRegistry(
         on_model_load=[inference_pool.load_model],
         on_model_reload=[inference_pool.reload_model],
@@ -155,7 +155,7 @@ def anchor_image_directory() -> Path:
 @pytest.fixture
 async def anchor_image_runtime_with_remote_predict_patch(
     anchor_image_directory, custom_runtime_tf: MLModel, mocker
-) -> AlibiExplainRuntime:
+) -> AsyncIterable[AlibiExplainRuntime]:
     def _mock_metadata(*args, **kwargs):
         return MetadataModelResponse(name="dummy", platform="dummy")
 
diff --git a/tests/codecs/test_base.py b/tests/codecs/test_base.py
index dd05a8e40..95f3a4865 100644
--- a/tests/codecs/test_base.py
+++ b/tests/codecs/test_base.py
@@ -1,9 +1,28 @@
 import numpy as np
+import pytest
 
-from mlserver.codecs.numpy import NumpyCodec
+from datetime import datetime
+from typing import List, Optional
+
+from mlserver.codecs.numpy import NumpyRequestCodec, NumpyCodec
+from mlserver.codecs.string import StringRequestCodec, StringCodec
+from mlserver.codecs.datetime import DatetimeCodec
+from mlserver.codecs.base import _CodecRegistry, InputCodec, RequestCodec
 from mlserver.types import RequestInput
 
 
+@pytest.fixture
+def codec_registry():
+    registry = _CodecRegistry()
+    registry.register_request_codec(NumpyRequestCodec.ContentType, NumpyRequestCodec)
+    registry.register_request_codec(StringRequestCodec.ContentType, StringRequestCodec)
+    registry.register_input_codec(NumpyCodec.ContentType, NumpyCodec)
+    registry.register_input_codec(StringCodec.ContentType, StringCodec)
+    registry.register_input_codec(DatetimeCodec.ContentType, DatetimeCodec)
+
+    return registry
+
+
 def test_deprecated_methods(caplog):
     request_input = RequestInput(
         name="foo", shape=[3], data=[1, 2, 3], datatype="INT32"
@@ -14,3 +33,37 @@ def test_deprecated_methods(caplog):
 
     assert any(["DEPRECATED" in rec.message for rec in caplog.records])
     np.testing.assert_array_equal(decoded, expected)
+
+
+@pytest.mark.parametrize(
+    "kwargs, expected",
+    [
+        ({"content_type": NumpyCodec.ContentType}, NumpyCodec),
+        ({"content_type": StringCodec.ContentType}, StringCodec),
+        ({"type_hint": List[str]}, StringCodec),
+        ({"type_hint": dict}, None),
+        ({"payload": [datetime.now()]}, DatetimeCodec),
+    ],
+)
+def test_find_input_codec(
+    codec_registry: _CodecRegistry, kwargs: dict, expected: Optional[InputCodec]
+):
+    input_codec = codec_registry.find_input_codec(**kwargs)
+    assert input_codec == expected
+
+
+@pytest.mark.parametrize(
+    "kwargs, expected",
+    [
+        ({"content_type": NumpyRequestCodec.ContentType}, NumpyRequestCodec),
+        ({"content_type": StringRequestCodec.ContentType}, StringRequestCodec),
+        ({"type_hint": List[str]}, StringRequestCodec),
+        ({"type_hint": dict}, None),
+        ({"payload": ["foo"]}, StringRequestCodec),
+    ],
+)
+def test_find_request_codec(
+    codec_registry: _CodecRegistry, kwargs: dict, expected: Optional[RequestCodec]
+):
+    request_codec = codec_registry.find_request_codec(**kwargs)
+    assert request_codec == expected
diff --git a/tests/codecs/test_decorator.py b/tests/codecs/test_decorator.py
new file mode 100644
index 000000000..2b256d7a3
--- /dev/null
+++ b/tests/codecs/test_decorator.py
@@ -0,0 +1,289 @@
+import pytest
+import numpy as np
+import pandas as pd
+
+from typing import Any, Callable, Dict, Optional, List
+
+from mlserver.types import (
+    InferenceRequest,
+    InferenceResponse,
+    RequestInput,
+    ResponseOutput,
+    Parameters,
+)
+from mlserver.codecs.base import InputCodec
+from mlserver.codecs.utils import Codec
+from mlserver.codecs.decorator import SignatureCodec
+from mlserver.codecs.errors import InputsNotFound, OutputNotFound
+from mlserver.codecs.numpy import NumpyCodec, NumpyRequestCodec
+from mlserver.codecs.string import StringCodec
+from mlserver.codecs.pandas import PandasCodec
+
+from ..fixtures import SimpleModel
+
+
+async def predict_fn(foo: np.ndarray, bar: List[str]) -> np.ndarray:
+    return np.array([3])
+
+
+def _implicit_optional(foo: np.ndarray, bar: List[str] = None) -> np.ndarray:
+    return np.array([2])
+
+
+def _explicit_optional(foo: np.ndarray, bar: Optional[List[str]]) -> np.ndarray:
+    return np.array([2])
+
+
+@pytest.fixture
+def signature_codec() -> SignatureCodec:
+    return SignatureCodec(predict_fn)
+
+
+@pytest.fixture
+def input_values() -> dict:
+    return {"foo": np.array([[1, 2]], dtype=np.int32), "bar": ["asd", "qwe"]}
+
+
+@pytest.fixture
+def output_value() -> np.ndarray:
+    return np.array([3])
+
+
+@pytest.fixture
+def inference_request(input_values: dict) -> InferenceRequest:
+    return InferenceRequest(
+        inputs=[
+            NumpyCodec.encode_input("foo", input_values["foo"]),
+            StringCodec.encode_input("bar", input_values["bar"]),
+        ]
+    )
+
+
+def test_get_codecs(signature_codec: SignatureCodec):
+    assert signature_codec._input_codecs == {"foo": NumpyCodec, "bar": StringCodec}
+    assert signature_codec._output_codecs == [NumpyCodec]
+
+
+def test_get_codecs_with_request():
+    def _f(foo: pd.DataFrame) -> (np.ndarray, pd.DataFrame):
+        return np.array([2]), pd.DataFrame({"bar": [2]})
+
+    signature_codec = SignatureCodec(_f)
+    assert signature_codec._input_codecs == {"foo": PandasCodec}
+    assert signature_codec._output_codecs == [NumpyCodec, PandasCodec]
+
+
+@pytest.mark.parametrize("predict_fn", [_implicit_optional, _explicit_optional])
+def test_get_codecs_with_optional(predict_fn: Callable):
+    signature_codec = SignatureCodec(predict_fn)
+    assert signature_codec._input_codecs == {"foo": NumpyCodec, "bar": StringCodec}
+    assert signature_codec._output_codecs == [NumpyCodec]
+
+
+@pytest.mark.parametrize(
+    "request_inputs, input_codecs, expected_values",
+    [
+        (
+            [
+                RequestInput(name="foo", datatype="INT64", shape=[1], data=[2]),
+                RequestInput(name="bar", datatype="BYTES", shape=[1], data=[b"asd"]),
+            ],
+            {"foo": NumpyCodec, "bar": StringCodec},
+            {"foo": np.array([2]), "bar": ["asd"]},
+        ),
+        (
+            [
+                RequestInput(name="a", datatype="INT64", shape=[1], data=[2]),
+                RequestInput(name="b", datatype="BYTES", shape=[1], data=["asd"]),
+            ],
+            {"foo": PandasCodec},
+            {"foo": pd.DataFrame({"a": [2], "b": ["asd"]})},
+        ),
+        (
+            [
+                RequestInput(name="bar", datatype="BYTES", shape=[1], data=[b"fiu"]),
+                RequestInput(name="a", datatype="INT64", shape=[1], data=[2]),
+                RequestInput(name="b", datatype="BYTES", shape=[1], data=["asd"]),
+            ],
+            {"bar": StringCodec, "foo": PandasCodec},
+            {"bar": ["fiu"], "foo": pd.DataFrame({"a": [2], "b": ["asd"]})},
+        ),
+    ],
+)
+def test_decode_request(
+    signature_codec: SignatureCodec,
+    request_inputs: List[RequestInput],
+    input_codecs: Dict[str, Codec],
+    expected_values: dict,
+):
+    signature_codec._input_codecs = input_codecs
+
+    inference_request = InferenceRequest(inputs=request_inputs)
+    inputs = signature_codec.decode_request(inference_request)
+
+    assert len(inputs) == len(expected_values)
+    for key, value in inputs.items():
+        assert key in expected_values
+        expected_value = expected_values[key]
+
+        if isinstance(expected_value, pd.DataFrame):
+            pd.testing.assert_frame_equal(value, expected_value)
+        else:
+            assert value == expected_value
+
+
+def test_decode_request_not_found(
+    signature_codec: SignatureCodec, inference_request: InferenceRequest
+):
+    inference_request.inputs[0].name = "not-foo"
+    with pytest.raises(InputsNotFound):
+        signature_codec.decode_request(inference_request)
+
+
+@pytest.mark.parametrize(
+    "output_values, output_codecs, expected_outputs",
+    [
+        (
+            np.array([2]),
+            [NumpyCodec],
+            [
+                ResponseOutput(
+                    name="output-0",
+                    datatype="INT64",
+                    shape=[1],
+                    data=[2],
+                )
+            ],
+        ),
+        (
+            ["foo"],
+            [StringCodec],
+            [
+                ResponseOutput(
+                    name="output-0",
+                    datatype="BYTES",
+                    shape=[1],
+                    data=[b"foo"],
+                    parameters=Parameters(content_type=StringCodec.ContentType),
+                )
+            ],
+        ),
+        (
+            (np.array([2]), ["foo"]),
+            [NumpyCodec, StringCodec],
+            [
+                ResponseOutput(
+                    name="output-0",
+                    datatype="INT64",
+                    shape=[1],
+                    data=[2],
+                ),
+                ResponseOutput(
+                    name="output-1",
+                    datatype="BYTES",
+                    shape=[1],
+                    data=[b"foo"],
+                    parameters=Parameters(content_type=StringCodec.ContentType),
+                ),
+            ],
+        ),
+        (
+            (["foo"], np.array([2])),
+            [StringCodec, NumpyCodec],
+            [
+                ResponseOutput(
+                    name="output-0",
+                    datatype="BYTES",
+                    shape=[1],
+                    data=[b"foo"],
+                    parameters=Parameters(content_type=StringCodec.ContentType),
+                ),
+                ResponseOutput(
+                    name="output-1",
+                    datatype="INT64",
+                    shape=[1],
+                    data=[2],
+                ),
+            ],
+        ),
+        (
+            pd.DataFrame({"a": [2], "b": ["foo"]}),
+            [PandasCodec],
+            [
+                ResponseOutput(name="a", datatype="INT64", shape=[1], data=[2]),
+                ResponseOutput(name="b", datatype="BYTES", shape=[1], data=[b"foo"]),
+            ],
+        ),
+        (
+            (pd.DataFrame({"a": [2], "b": ["foo"]}), ["bar"]),
+            [PandasCodec, StringCodec],
+            [
+                ResponseOutput(name="a", datatype="INT64", shape=[1], data=[2]),
+                ResponseOutput(name="b", datatype="BYTES", shape=[1], data=[b"foo"]),
+                ResponseOutput(
+                    name="output-1",
+                    datatype="BYTES",
+                    shape=[1],
+                    data=[b"bar"],
+                    parameters=Parameters(content_type=StringCodec.ContentType),
+                ),
+            ],
+        ),
+        (
+            (np.array([3]), pd.DataFrame({"a": [2], "b": ["foo"]})),
+            [NumpyCodec, PandasCodec],
+            [
+                ResponseOutput(
+                    name="output-0",
+                    datatype="INT64",
+                    shape=[1],
+                    data=[3],
+                ),
+                ResponseOutput(name="a", datatype="INT64", shape=[1], data=[2]),
+                ResponseOutput(name="b", datatype="BYTES", shape=[1], data=[b"foo"]),
+            ],
+        ),
+    ],
+)
+def test_encode_response(
+    signature_codec: SignatureCodec,
+    output_values: np.ndarray,
+    output_codecs: List[InputCodec],
+    expected_outputs: List[ResponseOutput],
+):
+    signature_codec._output_codecs = output_codecs
+    response = signature_codec.encode_response(model_name="foo", payload=output_values)
+
+    assert response.model_name == "foo"
+    assert response.outputs == expected_outputs
+
+
+@pytest.mark.parametrize(
+    "invalid_values",
+    [
+        ("foo", np.array([2])),
+        (np.array([2]), "foo"),
+    ],
+)
+def test_encode_response_not_found(
+    signature_codec: SignatureCodec, invalid_values: List[Any]
+):
+    """
+    Ensure the `SignatureCodec` detects when an output does not match the
+    method's signature and raise an error.
+    """
+    with pytest.raises(OutputNotFound):
+        signature_codec.encode_response(model_name="foo", payload=invalid_values)
+
+
+async def test_decode_args(
+    simple_model: SimpleModel,
+    inference_request: InferenceRequest,
+    output_value: np.ndarray,
+):
+    inference_response = await simple_model.predict(inference_request)
+
+    assert isinstance(inference_response, InferenceResponse)
+
+    res = NumpyRequestCodec.decode_response(inference_response)
+    assert res == output_value
diff --git a/tests/conftest.py b/tests/conftest.py
index e20d20943..3e3ff8293 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -13,7 +13,7 @@
 from mlserver.logging import get_logger
 from mlserver import types, Settings, ModelSettings
 
-from .fixtures import SumModel, ErrorModel
+from .fixtures import SumModel, ErrorModel, SimpleModel
 
 TESTS_PATH = os.path.dirname(__file__)
 TESTDATA_PATH = os.path.join(TESTS_PATH, "testdata")
@@ -58,6 +58,15 @@ def sum_model_settings() -> ModelSettings:
     return ModelSettings.parse_file(model_settings_path)
 
 
+@pytest.fixture
+def simple_model_settings() -> ModelSettings:
+    model_settings_path = os.path.join(TESTDATA_PATH, DEFAULT_MODEL_SETTINGS_FILENAME)
+    model_settings = ModelSettings.parse_file(model_settings_path)
+    model_settings.name = "simple-model"
+    model_settings.implementation = SimpleModel
+    return model_settings
+
+
 @pytest.fixture
 def error_model_settings() -> ModelSettings:
     model_settings_path = os.path.join(TESTDATA_PATH, DEFAULT_MODEL_SETTINGS_FILENAME)
@@ -75,6 +84,14 @@ async def error_model(
     return await model_registry.get_model(error_model_settings.name)
 
 
+@pytest.fixture
+async def simple_model(
+    model_registry: MultiModelRegistry, simple_model_settings: ModelSettings
+) -> SimpleModel:
+    await model_registry.load(simple_model_settings)
+    return await model_registry.get_model(simple_model_settings.name)
+
+
 @pytest.fixture
 async def sum_model(
     model_registry: MultiModelRegistry, sum_model_settings: ModelSettings
diff --git a/tests/fixtures.py b/tests/fixtures.py
index e6a07173f..a557ba99c 100644
--- a/tests/fixtures.py
+++ b/tests/fixtures.py
@@ -1,12 +1,13 @@
 import asyncio
 import random
 import string
+import numpy as np
 
-from typing import Dict
+from typing import Dict, List
 
 from mlserver import MLModel
 from mlserver.types import InferenceRequest, InferenceResponse, Parameters
-from mlserver.codecs import NumpyCodec
+from mlserver.codecs import NumpyCodec, decode_args
 from mlserver.handlers.custom import custom_handler
 from mlserver.errors import MLServerError
 
@@ -54,6 +55,12 @@ async def predict(self, payload: InferenceRequest) -> InferenceResponse:
         raise MLServerError(self.error_message)
 
 
+class SimpleModel(MLModel):
+    @decode_args
+    async def predict(self, foo: np.ndarray, bar: List[str]) -> np.ndarray:
+        return foo.sum(axis=1, keepdims=True)
+
+
 class SlowModel(MLModel):
     async def load(self) -> bool:
         await asyncio.sleep(10)