diff --git a/staging/src/k8s.io/apimachinery/pkg/runtime/BUILD b/staging/src/k8s.io/apimachinery/pkg/runtime/BUILD
index f26d903a0e5..f63f13cce76 100644
--- a/staging/src/k8s.io/apimachinery/pkg/runtime/BUILD
+++ b/staging/src/k8s.io/apimachinery/pkg/runtime/BUILD
@@ -50,6 +50,7 @@ go_library(
         "helper.go",
         "interfaces.go",
         "mapper.go",
+        "negotiate.go",
         "register.go",
         "scheme.go",
         "scheme_builder.go",
diff --git a/staging/src/k8s.io/apimachinery/pkg/runtime/interfaces.go b/staging/src/k8s.io/apimachinery/pkg/runtime/interfaces.go
index 33c377c3eec..f44693c0c6f 100644
--- a/staging/src/k8s.io/apimachinery/pkg/runtime/interfaces.go
+++ b/staging/src/k8s.io/apimachinery/pkg/runtime/interfaces.go
@@ -155,6 +155,28 @@ type NegotiatedSerializer interface {
 	DecoderToVersion(serializer Decoder, gv GroupVersioner) Decoder
 }
 
+// ClientNegotiator handles turning an HTTP content type into the appropriate encoder.
+// Use NewClientNegotiator or NewVersionedClientNegotiator to create this interface from
+// a NegotiatedSerializer.
+type ClientNegotiator interface {
+	// Encoder returns the appropriate encoder for the provided contentType (e.g. application/json)
+	// and any optional mediaType parameters (e.g. pretty=1), or an error. If no serializer is found
+	// a NegotiateError will be returned. The current client implementations consider params to be
+	// optional modifiers to the contentType and will ignore unrecognized parameters.
+	Encoder(contentType string, params map[string]string) (Encoder, error)
+	// Decoder returns the appropriate decoder for the provided contentType (e.g. application/json)
+	// and any optional mediaType parameters (e.g. pretty=1), or an error. If no serializer is found
+	// a NegotiateError will be returned. The current client implementations consider params to be
+	// optional modifiers to the contentType and will ignore unrecognized parameters.
+	Decoder(contentType string, params map[string]string) (Decoder, error)
+	// StreamDecoder returns the appropriate stream decoder for the provided contentType (e.g.
+	// application/json) and any optional mediaType parameters (e.g. pretty=1), or an error. If no
+	// serializer is found a NegotiateError will be returned. The Serializer and Framer will always
+	// be returned if a Decoder is returned. The current client implementations consider params to be
+	// optional modifiers to the contentType and will ignore unrecognized parameters.
+	StreamDecoder(contentType string, params map[string]string) (Decoder, Serializer, Framer, error)
+}
+
 // StorageSerializer is an interface used for obtaining encoders, decoders, and serializers
 // that can read and write data at rest. This would commonly be used by client tools that must
 // read files, or server side storage interfaces that persist restful objects.
diff --git a/staging/src/k8s.io/apimachinery/pkg/runtime/negotiate.go b/staging/src/k8s.io/apimachinery/pkg/runtime/negotiate.go
new file mode 100644
index 00000000000..159b301206a
--- /dev/null
+++ b/staging/src/k8s.io/apimachinery/pkg/runtime/negotiate.go
@@ -0,0 +1,146 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+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.
+*/
+
+package runtime
+
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// NegotiateError is returned when a ClientNegotiator is unable to locate
+// a serializer for the requested operation.
+type NegotiateError struct {
+	ContentType string
+	Stream      bool
+}
+
+func (e NegotiateError) Error() string {
+	if e.Stream {
+		return fmt.Sprintf("no stream serializers registered for %s", e.ContentType)
+	}
+	return fmt.Sprintf("no serializers registered for %s", e.ContentType)
+}
+
+type clientNegotiator struct {
+	serializer     NegotiatedSerializer
+	encode, decode GroupVersioner
+}
+
+func (n *clientNegotiator) Encoder(contentType string, params map[string]string) (Encoder, error) {
+	// TODO: `pretty=1` is handled in NegotiateOutputMediaType, consider moving it to this method
+	// if client negotiators truly need to use it
+	mediaTypes := n.serializer.SupportedMediaTypes()
+	info, ok := SerializerInfoForMediaType(mediaTypes, contentType)
+	if !ok {
+		if len(contentType) != 0 || len(mediaTypes) == 0 {
+			return nil, NegotiateError{ContentType: contentType}
+		}
+		info = mediaTypes[0]
+	}
+	return n.serializer.EncoderForVersion(info.Serializer, n.encode), nil
+}
+
+func (n *clientNegotiator) Decoder(contentType string, params map[string]string) (Decoder, error) {
+	mediaTypes := n.serializer.SupportedMediaTypes()
+	info, ok := SerializerInfoForMediaType(mediaTypes, contentType)
+	if !ok {
+		if len(contentType) != 0 || len(mediaTypes) == 0 {
+			return nil, NegotiateError{ContentType: contentType}
+		}
+		info = mediaTypes[0]
+	}
+	return n.serializer.DecoderToVersion(info.Serializer, n.decode), nil
+}
+
+func (n *clientNegotiator) StreamDecoder(contentType string, params map[string]string) (Decoder, Serializer, Framer, error) {
+	mediaTypes := n.serializer.SupportedMediaTypes()
+	info, ok := SerializerInfoForMediaType(mediaTypes, contentType)
+	if !ok {
+		if len(contentType) != 0 || len(mediaTypes) == 0 {
+			return nil, nil, nil, NegotiateError{ContentType: contentType, Stream: true}
+		}
+		info = mediaTypes[0]
+	}
+	if info.StreamSerializer == nil {
+		return nil, nil, nil, NegotiateError{ContentType: info.MediaType, Stream: true}
+	}
+	return n.serializer.DecoderToVersion(info.Serializer, n.decode), info.StreamSerializer.Serializer, info.StreamSerializer.Framer, nil
+}
+
+// NewClientNegotiator will attempt to retrieve the appropriate encoder, decoder, or
+// stream decoder for a given content type. Does not perform any conversion, but will
+// encode the object to the desired group, version, and kind. Use when creating a client.
+func NewClientNegotiator(serializer NegotiatedSerializer, gv schema.GroupVersion) ClientNegotiator {
+	return &clientNegotiator{
+		serializer: serializer,
+		encode:     gv,
+	}
+}
+
+// NewInternalClientNegotiator applies the default client rules for connecting to a Kubernetes apiserver
+// where objects are converted to gv prior to sending and decoded to their internal representation prior
+// to retrieval.
+//
+// DEPRECATED: Internal clients are deprecated and will be removed in a future Kubernetes release.
+func NewInternalClientNegotiator(serializer NegotiatedSerializer, gv schema.GroupVersion) ClientNegotiator {
+	decode := schema.GroupVersions{
+		{
+			Group:   gv.Group,
+			Version: APIVersionInternal,
+		},
+		// always include the legacy group as a decoding target to handle non-error `Status` return types
+		{
+			Group:   "",
+			Version: APIVersionInternal,
+		},
+	}
+	return &clientNegotiator{
+		encode:     gv,
+		decode:     decode,
+		serializer: serializer,
+	}
+}
+
+// NewSimpleClientNegotiator will negotiate for a single serializer. This should only be used
+// for testing or when the caller is taking responsibility for setting the GVK on encoded objects.
+func NewSimpleClientNegotiator(info SerializerInfo, gv schema.GroupVersion) ClientNegotiator {
+	return &clientNegotiator{
+		serializer: &simpleNegotiatedSerializer{info: info},
+		encode:     gv,
+	}
+}
+
+type simpleNegotiatedSerializer struct {
+	info SerializerInfo
+}
+
+func NewSimpleNegotiatedSerializer(info SerializerInfo) NegotiatedSerializer {
+	return &simpleNegotiatedSerializer{info: info}
+}
+
+func (n *simpleNegotiatedSerializer) SupportedMediaTypes() []SerializerInfo {
+	return []SerializerInfo{n.info}
+}
+
+func (n *simpleNegotiatedSerializer) EncoderForVersion(e Encoder, _ GroupVersioner) Encoder {
+	return e
+}
+
+func (n *simpleNegotiatedSerializer) DecoderToVersion(d Decoder, _gv GroupVersioner) Decoder {
+	return d
+}
diff --git a/test/e2e/apimachinery/BUILD b/test/e2e/apimachinery/BUILD
index a60c8f93dc1..286a9248b58 100644
--- a/test/e2e/apimachinery/BUILD
+++ b/test/e2e/apimachinery/BUILD
@@ -21,6 +21,7 @@ go_library(
         "garbage_collector.go",
         "generated_clientset.go",
         "namespace.go",
+        "protocol.go",
         "resource_quota.go",
         "table_conversion.go",
         "watch.go",