diff --git a/plugin/conversions.go b/plugin/conversions.go
new file mode 100644
index 0000000000..561e501370
--- /dev/null
+++ b/plugin/conversions.go
@@ -0,0 +1,108 @@
+package plugin
+
+import (
+	"encoding/json"
+
+	"github.com/hashicorp/terraform/configs/configschema"
+	"github.com/hashicorp/terraform/plugin/proto"
+	"github.com/hashicorp/terraform/providers"
+	"github.com/hashicorp/terraform/tfdiags"
+	"github.com/zclconf/go-cty/cty"
+)
+
+// ProtoToProviderSchema takes a proto.Schema and converts it to a providers.Schema.
+func ProtoToProviderSchema(s *proto.Schema) providers.Schema {
+	return providers.Schema{
+		Version: int(s.Version),
+		Block:   schemaBlock(s.Block),
+	}
+}
+
+// schemaBlock takes the GetSchcema_Block from a grpc response and converts it
+// to a terraform *configschema.Block.
+func schemaBlock(b *proto.Schema_Block) *configschema.Block {
+	block := &configschema.Block{
+		Attributes: make(map[string]*configschema.Attribute),
+		BlockTypes: make(map[string]*configschema.NestedBlock),
+	}
+
+	for _, a := range b.Attributes {
+		attr := &configschema.Attribute{
+			Description: a.Description,
+			Required:    a.Required,
+			Optional:    a.Optional,
+			Computed:    a.Computed,
+			Sensitive:   a.Sensitive,
+		}
+
+		if err := json.Unmarshal(a.Type, &attr.Type); err != nil {
+			panic(err)
+		}
+
+		block.Attributes[a.Name] = attr
+	}
+
+	for _, b := range b.BlockTypes {
+		block.BlockTypes[b.TypeName] = schemaNestedBlock(b)
+	}
+
+	return block
+}
+
+func schemaNestedBlock(b *proto.Schema_NestedBlock) *configschema.NestedBlock {
+	nb := &configschema.NestedBlock{
+		Nesting:  configschema.NestingMode(b.Nesting),
+		MinItems: int(b.MinItems),
+		MaxItems: int(b.MaxItems),
+	}
+
+	nested := schemaBlock(b.Block)
+	nb.Block = *nested
+	return nb
+}
+
+// ProtoToDiagnostics converts a list of proto.Diagnostics to a tf.Diagnostics.
+// for now we assume these only contain a basic message
+func ProtoToDiagnostics(ds []*proto.Diagnostic) tfdiags.Diagnostics {
+	var diags tfdiags.Diagnostics
+	for _, d := range ds {
+		var severity tfdiags.Severity
+
+		switch d.Severity {
+		case proto.Diagnostic_ERROR:
+			severity = tfdiags.Error
+		case proto.Diagnostic_WARNING:
+			severity = tfdiags.Warning
+		}
+
+		var newDiag tfdiags.Diagnostic
+
+		// if there's an attribute path, we need to create a AttributeValue diagnostic
+		if d.Attribute != nil {
+			path := attributePath(d.Attribute)
+			newDiag = tfdiags.AttributeValue(severity, d.Summary, d.Detail, path)
+		} else {
+			newDiag = tfdiags.Sourceless(severity, d.Summary, d.Detail)
+		}
+
+		diags = diags.Append(newDiag)
+	}
+
+	return diags
+}
+
+// attributePath takes the proto encoded path and converts it to a cty.Path
+func attributePath(ap *proto.AttributePath) cty.Path {
+	var p cty.Path
+	for _, step := range ap.Steps {
+		switch selector := step.Selector.(type) {
+		case *proto.AttributePath_Step_AttributeName:
+			p = p.GetAttr(selector.AttributeName)
+		case *proto.AttributePath_Step_ElementKeyString:
+			p = p.Index(cty.StringVal(selector.ElementKeyString))
+		case *proto.AttributePath_Step_ElementKeyInt:
+			p = p.Index(cty.NumberIntVal(selector.ElementKeyInt))
+		}
+	}
+	return p
+}
diff --git a/plugin/conversions_test.go b/plugin/conversions_test.go
new file mode 100644
index 0000000000..f9af3ca77e
--- /dev/null
+++ b/plugin/conversions_test.go
@@ -0,0 +1,502 @@
+package plugin
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/hashicorp/terraform/configs/configschema"
+	"github.com/hashicorp/terraform/plugin/proto"
+	"github.com/hashicorp/terraform/tfdiags"
+	"github.com/zclconf/go-cty/cty"
+)
+
+var (
+	equateEmpty   = cmpopts.EquateEmpty()
+	typeComparer  = cmp.Comparer(cty.Type.Equals)
+	valueComparer = cmp.Comparer(cty.Value.RawEquals)
+)
+
+// Test that we can convert configschema to protobuf types and back again.
+func TestConvertSchemaBlocks(t *testing.T) {
+	tests := map[string]struct {
+		Block *proto.Schema_Block
+		Want  *configschema.Block
+	}{
+		"attributes": {
+			&proto.Schema_Block{
+				Attributes: []*proto.Schema_Attribute{
+					{
+						Name:     "computed",
+						Type:     []byte(`["list","bool"]`),
+						Computed: true,
+					},
+					{
+						Name:     "optional",
+						Type:     []byte(`"string"`),
+						Optional: true,
+					},
+					{
+						Name:     "optional_computed",
+						Type:     []byte(`["map","bool"]`),
+						Optional: true,
+						Computed: true,
+					},
+					{
+						Name:     "required",
+						Type:     []byte(`"number"`),
+						Required: true,
+					},
+				},
+			},
+			&configschema.Block{
+				Attributes: map[string]*configschema.Attribute{
+					"computed": {
+						Type:     cty.List(cty.Bool),
+						Computed: true,
+					},
+					"optional": {
+						Type:     cty.String,
+						Optional: true,
+					},
+					"optional_computed": {
+						Type:     cty.Map(cty.Bool),
+						Optional: true,
+						Computed: true,
+					},
+					"required": {
+						Type:     cty.Number,
+						Required: true,
+					},
+				},
+			},
+		},
+		"blocks": {
+			&proto.Schema_Block{
+				BlockTypes: []*proto.Schema_NestedBlock{
+					{
+						TypeName: "list",
+						Nesting:  proto.Schema_NestedBlock_LIST,
+						Block:    &proto.Schema_Block{},
+					},
+					{
+						TypeName: "map",
+						Nesting:  proto.Schema_NestedBlock_MAP,
+						Block:    &proto.Schema_Block{},
+					},
+					{
+						TypeName: "set",
+						Nesting:  proto.Schema_NestedBlock_SET,
+						Block:    &proto.Schema_Block{},
+					},
+					{
+						TypeName: "single",
+						Nesting:  proto.Schema_NestedBlock_SINGLE,
+						Block: &proto.Schema_Block{
+							Attributes: []*proto.Schema_Attribute{
+								{
+									Name:     "foo",
+									Type:     []byte(`"dynamic"`),
+									Required: true,
+								},
+							},
+						},
+					},
+				},
+			},
+			&configschema.Block{
+				BlockTypes: map[string]*configschema.NestedBlock{
+					"list": &configschema.NestedBlock{
+						Nesting: configschema.NestingList,
+					},
+					"map": &configschema.NestedBlock{
+						Nesting: configschema.NestingMap,
+					},
+					"set": &configschema.NestedBlock{
+						Nesting: configschema.NestingSet,
+					},
+					"single": &configschema.NestedBlock{
+						Nesting: configschema.NestingSingle,
+						Block: configschema.Block{
+							Attributes: map[string]*configschema.Attribute{
+								"foo": {
+									Type:     cty.DynamicPseudoType,
+									Required: true,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		"deep block nesting": {
+			&proto.Schema_Block{
+				BlockTypes: []*proto.Schema_NestedBlock{
+					{
+						TypeName: "single",
+						Nesting:  proto.Schema_NestedBlock_SINGLE,
+						Block: &proto.Schema_Block{
+							BlockTypes: []*proto.Schema_NestedBlock{
+								{
+									TypeName: "list",
+									Nesting:  proto.Schema_NestedBlock_LIST,
+									Block: &proto.Schema_Block{
+										BlockTypes: []*proto.Schema_NestedBlock{
+											{
+												TypeName: "set",
+												Nesting:  proto.Schema_NestedBlock_SET,
+												Block:    &proto.Schema_Block{},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			&configschema.Block{
+				BlockTypes: map[string]*configschema.NestedBlock{
+					"single": &configschema.NestedBlock{
+						Nesting: configschema.NestingSingle,
+						Block: configschema.Block{
+							BlockTypes: map[string]*configschema.NestedBlock{
+								"list": &configschema.NestedBlock{
+									Nesting: configschema.NestingList,
+									Block: configschema.Block{
+										BlockTypes: map[string]*configschema.NestedBlock{
+											"set": &configschema.NestedBlock{
+												Nesting: configschema.NestingSet,
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			converted := schemaBlock(tc.Block)
+			if !cmp.Equal(converted, tc.Want, typeComparer, valueComparer, equateEmpty) {
+				t.Fatal(cmp.Diff(converted, tc.Want, typeComparer, valueComparer, equateEmpty))
+			}
+		})
+	}
+}
+
+func TestDiagnostics(t *testing.T) {
+	type diagFlat struct {
+		Severity tfdiags.Severity
+		Attr     []interface{}
+		Summary  string
+		Detail   string
+	}
+
+	tests := map[string]struct {
+		Cons func([]*proto.Diagnostic) []*proto.Diagnostic
+		Want []diagFlat
+	}{
+		"nil": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				return diags
+			},
+			nil,
+		},
+		"error": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				return append(diags, &proto.Diagnostic{
+					Severity: proto.Diagnostic_ERROR,
+					Summary:  "simple error",
+				})
+			},
+			[]diagFlat{
+				{
+					Severity: tfdiags.Error,
+					Summary:  "simple error",
+				},
+			},
+		},
+		"detailed error": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				return append(diags, &proto.Diagnostic{
+					Severity: proto.Diagnostic_ERROR,
+					Summary:  "simple error",
+					Detail:   "detailed error",
+				})
+			},
+			[]diagFlat{
+				{
+					Severity: tfdiags.Error,
+					Summary:  "simple error",
+					Detail:   "detailed error",
+				},
+			},
+		},
+		"warning": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				return append(diags, &proto.Diagnostic{
+					Severity: proto.Diagnostic_WARNING,
+					Summary:  "simple warning",
+				})
+			},
+			[]diagFlat{
+				{
+					Severity: tfdiags.Warning,
+					Summary:  "simple warning",
+				},
+			},
+		},
+		"detailed warning": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				return append(diags, &proto.Diagnostic{
+					Severity: proto.Diagnostic_WARNING,
+					Summary:  "simple warning",
+					Detail:   "detailed warning",
+				})
+			},
+			[]diagFlat{
+				{
+					Severity: tfdiags.Warning,
+					Summary:  "simple warning",
+					Detail:   "detailed warning",
+				},
+			},
+		},
+		"multi error": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				diags = append(diags, &proto.Diagnostic{
+					Severity: proto.Diagnostic_ERROR,
+					Summary:  "first error",
+				}, &proto.Diagnostic{
+					Severity: proto.Diagnostic_ERROR,
+					Summary:  "second error",
+				})
+				return diags
+			},
+			[]diagFlat{
+				{
+					Severity: tfdiags.Error,
+					Summary:  "first error",
+				},
+				{
+					Severity: tfdiags.Error,
+					Summary:  "second error",
+				},
+			},
+		},
+		"warning and error": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				diags = append(diags, &proto.Diagnostic{
+					Severity: proto.Diagnostic_WARNING,
+					Summary:  "warning",
+				}, &proto.Diagnostic{
+					Severity: proto.Diagnostic_ERROR,
+					Summary:  "error",
+				})
+				return diags
+			},
+			[]diagFlat{
+				{
+					Severity: tfdiags.Warning,
+					Summary:  "warning",
+				},
+				{
+					Severity: tfdiags.Error,
+					Summary:  "error",
+				},
+			},
+		},
+		"attr error": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				diags = append(diags, &proto.Diagnostic{
+					Severity: proto.Diagnostic_ERROR,
+					Summary:  "error",
+					Detail:   "error detail",
+					Attribute: &proto.AttributePath{
+						Steps: []*proto.AttributePath_Step{
+							{
+								Selector: &proto.AttributePath_Step_AttributeName{
+									AttributeName: "attribute_name",
+								},
+							},
+						},
+					},
+				})
+				return diags
+			},
+			[]diagFlat{
+				{
+					Severity: tfdiags.Error,
+					Summary:  "error",
+					Detail:   "error detail",
+					Attr:     []interface{}{"attribute_name"},
+				},
+			},
+		},
+		"multi attr": {
+			func(diags []*proto.Diagnostic) []*proto.Diagnostic {
+				diags = append(diags,
+					&proto.Diagnostic{
+						Severity: proto.Diagnostic_ERROR,
+						Summary:  "error 1",
+						Detail:   "error 1 detail",
+						Attribute: &proto.AttributePath{
+							Steps: []*proto.AttributePath_Step{
+								{
+									Selector: &proto.AttributePath_Step_AttributeName{
+										AttributeName: "attr",
+									},
+								},
+							},
+						},
+					},
+					&proto.Diagnostic{
+						Severity: proto.Diagnostic_ERROR,
+						Summary:  "error 2",
+						Detail:   "error 2 detail",
+						Attribute: &proto.AttributePath{
+							Steps: []*proto.AttributePath_Step{
+								{
+									Selector: &proto.AttributePath_Step_AttributeName{
+										AttributeName: "attr",
+									},
+								},
+								{
+									Selector: &proto.AttributePath_Step_AttributeName{
+										AttributeName: "sub",
+									},
+								},
+							},
+						},
+					},
+					&proto.Diagnostic{
+						Severity: proto.Diagnostic_WARNING,
+						Summary:  "warning",
+						Detail:   "warning detail",
+						Attribute: &proto.AttributePath{
+							Steps: []*proto.AttributePath_Step{
+								{
+									Selector: &proto.AttributePath_Step_AttributeName{
+										AttributeName: "attr",
+									},
+								},
+								{
+									Selector: &proto.AttributePath_Step_ElementKeyInt{
+										ElementKeyInt: 1,
+									},
+								},
+								{
+									Selector: &proto.AttributePath_Step_AttributeName{
+										AttributeName: "sub",
+									},
+								},
+							},
+						},
+					},
+					&proto.Diagnostic{
+						Severity: proto.Diagnostic_ERROR,
+						Summary:  "error 3",
+						Detail:   "error 3 detail",
+						Attribute: &proto.AttributePath{
+							Steps: []*proto.AttributePath_Step{
+								{
+									Selector: &proto.AttributePath_Step_AttributeName{
+										AttributeName: "attr",
+									},
+								},
+								{
+									Selector: &proto.AttributePath_Step_ElementKeyString{
+										ElementKeyString: "idx",
+									},
+								},
+								{
+									Selector: &proto.AttributePath_Step_AttributeName{
+										AttributeName: "sub",
+									},
+								},
+							},
+						},
+					},
+				)
+
+				return diags
+			},
+			[]diagFlat{
+				{
+					Severity: tfdiags.Error,
+					Summary:  "error 1",
+					Detail:   "error 1 detail",
+					Attr:     []interface{}{"attr"},
+				},
+				{
+					Severity: tfdiags.Error,
+					Summary:  "error 2",
+					Detail:   "error 2 detail",
+					Attr:     []interface{}{"attr", "sub"},
+				},
+				{
+					Severity: tfdiags.Warning,
+					Summary:  "warning",
+					Detail:   "warning detail",
+					Attr:     []interface{}{"attr", 1, "sub"},
+				},
+				{
+					Severity: tfdiags.Error,
+					Summary:  "error 3",
+					Detail:   "error 3 detail",
+					Attr:     []interface{}{"attr", "idx", "sub"},
+				},
+			},
+		},
+	}
+
+	flattenTFDiags := func(ds tfdiags.Diagnostics) []diagFlat {
+		var flat []diagFlat
+		for _, item := range ds {
+			desc := item.Description()
+
+			var attr []interface{}
+
+			for _, a := range tfdiags.GetAttribute(item) {
+				switch step := a.(type) {
+				case cty.GetAttrStep:
+					attr = append(attr, step.Name)
+				case cty.IndexStep:
+					switch step.Key.Type() {
+					case cty.Number:
+						i, _ := step.Key.AsBigFloat().Int64()
+						attr = append(attr, int(i))
+					case cty.String:
+						attr = append(attr, step.Key.AsString())
+					}
+				}
+			}
+
+			flat = append(flat, diagFlat{
+				Severity: item.Severity(),
+				Attr:     attr,
+				Summary:  desc.Summary,
+				Detail:   desc.Detail,
+			})
+		}
+		return flat
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			// we take the
+			tfDiags := ProtoToDiagnostics(tc.Cons(nil))
+
+			flat := flattenTFDiags(tfDiags)
+
+			if !cmp.Equal(flat, tc.Want, typeComparer, valueComparer, equateEmpty) {
+				t.Fatal(cmp.Diff(flat, tc.Want, typeComparer, valueComparer, equateEmpty))
+			}
+		})
+	}
+}
diff --git a/plugin/grpc_provider.go b/plugin/grpc_provider.go
new file mode 100644
index 0000000000..0edc7f762a
--- /dev/null
+++ b/plugin/grpc_provider.go
@@ -0,0 +1,439 @@
+package plugin
+
+import (
+	"context"
+	"errors"
+	"sync"
+
+	"github.com/hashicorp/terraform/plugin/proto"
+	"github.com/hashicorp/terraform/providers"
+	"github.com/hashicorp/terraform/version"
+	"github.com/zclconf/go-cty/cty/msgpack"
+	"google.golang.org/grpc"
+)
+
+// GRPCProvider handles the client, or core side of the plugin rpc connection.
+// The GRPCProvider methods are mostly a translation layer between the
+// terraform provioders types and the grpc proto types, directly converting
+// between the two.
+type GRPCProvider struct {
+	conn   *grpc.ClientConn
+	client proto.ProviderClient
+
+	// this context is created by the plugin package, and is canceled when the
+	// plugin process ends.
+	ctx context.Context
+
+	// schema stores the schema for this provider. This is used to properly
+	// serialize the state for requests.
+	mu      sync.Mutex
+	schemas providers.GetSchemaResponse
+}
+
+// getSchema is used internally to get the saved provider schema.  The schema
+// should have already been fetched from the provider, but we have to
+// synchronize access to avoid being called concurrently with GetSchema.
+func (p *GRPCProvider) getSchema() providers.GetSchemaResponse {
+	p.mu.Lock()
+	// unlock inline in case GetSchema needs to be called
+	if p.schemas.Provider.Block != nil {
+		return p.schemas
+	}
+	p.mu.Unlock()
+
+	// the schema should have been fetched already, but give it another shot
+	// just in case things are being called out of order. This may happen for
+	// tests.
+	schemas := p.GetSchema()
+	if schemas.Diagnostics.HasErrors() {
+		panic(schemas.Diagnostics.Err())
+	}
+
+	return schemas
+}
+
+// getResourceSchema is a helper to extract the schema for a resource, and
+// panics if the schema is not available.
+func (p *GRPCProvider) getResourceSchema(name string) providers.Schema {
+	schema := p.getSchema()
+	resSchema, ok := schema.ResourceTypes[name]
+	if !ok {
+		panic("unknown resource type " + name)
+	}
+	return resSchema
+}
+
+// gettDatasourceSchema is a helper to extract the schema for a datasource, and
+// panics if that schema is not available.
+func (p *GRPCProvider) getDatasourceSchema(name string) providers.Schema {
+	schema := p.getSchema()
+	dataSchema, ok := schema.DataSources[name]
+	if !ok {
+		panic("unknown data source " + name)
+	}
+	return dataSchema
+}
+
+func (p *GRPCProvider) GetSchema() (resp providers.GetSchemaResponse) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.schemas.Provider.Block != nil {
+		return p.schemas
+	}
+
+	resp.ResourceTypes = make(map[string]providers.Schema)
+	resp.DataSources = make(map[string]providers.Schema)
+
+	protoResp, err := p.client.GetSchema(p.ctx, new(proto.GetProviderSchema_Request))
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	if protoResp.Provider == nil {
+		resp.Diagnostics = resp.Diagnostics.Append(errors.New("missing provider schema"))
+		return resp
+	}
+
+	resp.Provider = ProtoToProviderSchema(protoResp.Provider)
+
+	for name, res := range protoResp.ResourceSchemas {
+		resp.ResourceTypes[name] = ProtoToProviderSchema(res)
+	}
+
+	for name, data := range protoResp.DataSourceSchemas {
+		resp.DataSources[name] = ProtoToProviderSchema(data)
+	}
+
+	p.schemas = resp
+
+	return resp
+}
+
+func (p *GRPCProvider) ValidateProviderConfig(r providers.ValidateProviderConfigRequest) (resp providers.ValidateProviderConfigResponse) {
+	schema := p.getSchema()
+	mp, err := msgpack.Marshal(r.Config, schema.Provider.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.ValidateProviderConfig_Request{
+		Config: &proto.DynamicValue{Msgpack: mp},
+	}
+
+	protoResp, err := p.client.ValidateProviderConfig(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+	resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(protoResp.Diagnostics))
+	return resp
+}
+
+func (p *GRPCProvider) ValidateResourceTypeConfig(r providers.ValidateResourceTypeConfigRequest) (resp providers.ValidateResourceTypeConfigResponse) {
+	resourceSchema := p.getResourceSchema(r.TypeName)
+
+	mp, err := msgpack.Marshal(r.Config, resourceSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.ValidateResourceTypeConfig_Request{
+		TypeName: r.TypeName,
+		Config:   &proto.DynamicValue{Msgpack: mp},
+	}
+
+	protoResp, err := p.client.ValidateResourceTypeConfig(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(protoResp.Diagnostics))
+	return resp
+}
+
+func (p *GRPCProvider) ValidateDataSourceConfig(r providers.ValidateDataSourceConfigRequest) (resp providers.ValidateDataSourceConfigResponse) {
+	dataSchema := p.getDatasourceSchema(r.TypeName)
+
+	mp, err := msgpack.Marshal(r.Config, dataSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.ValidateDataSourceConfig_Request{
+		TypeName: r.TypeName,
+		Config:   &proto.DynamicValue{Msgpack: mp},
+	}
+
+	protoResp, err := p.client.ValidateDataSourceConfig(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+	resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(protoResp.Diagnostics))
+	return resp
+}
+
+func (p *GRPCProvider) UpgradeResourceState(r providers.UpgradeResourceStateRequest) (resp providers.UpgradeResourceStateResponse) {
+	resSchema := p.getResourceSchema(r.TypeName)
+
+	protoReq := &proto.UpgradeResourceState_Request{
+		TypeName: r.TypeName,
+		Version:  int64(r.Version),
+		RawState: &proto.RawState{
+			Json:    r.RawStateJSON,
+			Flatmap: r.RawStateFlatmap,
+		},
+	}
+
+	protoResp, err := p.client.UpgradeResourceState(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	state, err := msgpack.Unmarshal(protoResp.UpgradedState.Msgpack, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.UpgradedState = state
+
+	resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(protoResp.Diagnostics))
+	return resp
+}
+
+func (p *GRPCProvider) Configure(r providers.ConfigureRequest) (resp providers.ConfigureResponse) {
+	schema := p.getSchema()
+
+	var mp []byte
+
+	// we don't have anything to marshal if there's no config
+	mp, err := msgpack.Marshal(r.Config, schema.Provider.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.Configure_Request{
+		TerraformVersion: version.Version,
+		Config: &proto.DynamicValue{
+			Msgpack: mp,
+		},
+	}
+
+	protoResp, err := p.client.Configure(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+	resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(protoResp.Diagnostics))
+	return resp
+}
+
+func (p *GRPCProvider) Stop() error {
+	resp, err := p.client.Stop(p.ctx, new(proto.Stop_Request))
+	if err != nil {
+		return err
+	}
+
+	if resp.Error != "" {
+		return errors.New(resp.Error)
+	}
+	return nil
+}
+
+func (p *GRPCProvider) ReadResource(r providers.ReadResourceRequest) (resp providers.ReadResourceResponse) {
+	resSchema := p.getResourceSchema(r.TypeName)
+
+	mp, err := msgpack.Marshal(r.PriorState, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.ReadResource_Request{
+		TypeName:     r.TypeName,
+		CurrentState: &proto.DynamicValue{Msgpack: mp},
+	}
+
+	protoResp, err := p.client.ReadResource(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(protoResp.Diagnostics))
+
+	state, err := msgpack.Unmarshal(protoResp.NewState.Msgpack, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.NewState = state
+	return resp
+}
+
+func (p *GRPCProvider) PlanResourceChange(r providers.PlanResourceChangeRequest) (resp providers.PlanResourceChangeResponse) {
+	resSchema := p.getResourceSchema(r.TypeName)
+
+	priorMP, err := msgpack.Marshal(r.PriorState, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+	propMP, err := msgpack.Marshal(r.ProposedNewState, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.PlanResourceChange_Request{
+		TypeName:         r.TypeName,
+		PriorState:       &proto.DynamicValue{Msgpack: priorMP},
+		ProposedNewState: &proto.DynamicValue{Msgpack: propMP},
+		PriorPrivate:     r.PriorPrivate,
+	}
+
+	protoResp, err := p.client.PlanResourceChange(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(protoResp.Diagnostics))
+
+	state, err := msgpack.Unmarshal(protoResp.PlannedState.Msgpack, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.PlannedState = state
+
+	for _, p := range protoResp.RequiresReplace {
+		resp.RequiresReplace = append(resp.RequiresReplace, attributePath(p))
+	}
+
+	resp.PlannedPrivate = protoResp.PlannedPrivate
+
+	return resp
+}
+
+func (p *GRPCProvider) ApplyResourceChange(r providers.ApplyResourceChangeRequest) (resp providers.ApplyResourceChangeResponse) {
+	resSchema := p.getResourceSchema(r.TypeName)
+
+	priorMP, err := msgpack.Marshal(r.PriorState, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+	plannedMP, err := msgpack.Marshal(r.PlannedState, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.ApplyResourceChange_Request{
+		TypeName:       r.TypeName,
+		PriorState:     &proto.DynamicValue{Msgpack: priorMP},
+		PlannedState:   &proto.DynamicValue{Msgpack: plannedMP},
+		PlannedPrivate: r.PlannedPrivate,
+	}
+
+	protoResp, err := p.client.ApplyResourceChange(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.Private = protoResp.Private
+
+	state, err := msgpack.Unmarshal(protoResp.NewState.Msgpack, resSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.NewState = state
+
+	return resp
+}
+
+func (p *GRPCProvider) ImportResourceState(r providers.ImportResourceStateRequest) (resp providers.ImportResourceStateResponse) {
+	resSchema := p.getResourceSchema(r.TypeName)
+
+	protoReq := &proto.ImportResourceState_Request{
+		TypeName: r.TypeName,
+		Id:       r.ID,
+	}
+
+	protoResp, err := p.client.ImportResourceState(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	for _, imported := range protoResp.ImportedResources {
+		resource := providers.ImportedResource{
+			TypeName: imported.TypeName,
+			Private:  imported.Private,
+		}
+		state, err := msgpack.Unmarshal(imported.State.Msgpack, resSchema.Block.ImpliedType())
+		if err != nil {
+			resp.Diagnostics = resp.Diagnostics.Append(err)
+			return resp
+		}
+
+		resource.State = state
+		resp.ImportedResources = append(resp.ImportedResources, resource)
+	}
+
+	return resp
+}
+
+func (p *GRPCProvider) ReadDataSource(r providers.ReadDataSourceRequest) (resp providers.ReadDataSourceResponse) {
+	dataSchema := p.getDatasourceSchema(r.TypeName)
+
+	config, err := msgpack.Marshal(r.Config, dataSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.ReadDataSource_Request{
+		TypeName: r.TypeName,
+		Config: &proto.DynamicValue{
+			Msgpack: config,
+		},
+	}
+
+	protoResp, err := p.client.ReadDataSource(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	state, err := msgpack.Unmarshal(protoResp.State.Msgpack, dataSchema.Block.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	resp.State = state
+	return resp
+
+}
+
+// closing the grpc connection is final, and terraform will call it at the end of every phase.
+// FIXME: do we need this, and if so, how do we fix it?
+func (p *GRPCProvider) Close() error {
+	return nil
+}
diff --git a/plugin/grpc_provider_test.go b/plugin/grpc_provider_test.go
new file mode 100644
index 0000000000..f61af0a8dc
--- /dev/null
+++ b/plugin/grpc_provider_test.go
@@ -0,0 +1,433 @@
+package plugin
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/google/go-cmp/cmp"
+	"github.com/hashicorp/terraform/config/hcl2shim"
+	"github.com/hashicorp/terraform/providers"
+	"github.com/hashicorp/terraform/tfdiags"
+	"github.com/zclconf/go-cty/cty"
+
+	mockproto "github.com/hashicorp/terraform/plugin/mock_proto"
+	"github.com/hashicorp/terraform/plugin/proto"
+)
+
+var _ providers.Interface = (*GRPCProvider)(nil)
+
+func mockProviderClient(t *testing.T) *mockproto.MockProviderClient {
+	ctrl := gomock.NewController(t)
+	client := mockproto.NewMockProviderClient(ctrl)
+
+	// we always need a GetSchema method
+	client.EXPECT().GetSchema(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(providerProtoSchema(), nil)
+
+	return client
+}
+
+func checkDiags(t *testing.T, d tfdiags.Diagnostics) {
+	t.Helper()
+	if d.HasErrors() {
+		t.Fatal(d.Err())
+	}
+}
+
+func providerProtoSchema() *proto.GetProviderSchema_Response {
+	return &proto.GetProviderSchema_Response{
+		Provider: &proto.Schema{
+			Block: &proto.Schema_Block{
+				Attributes: []*proto.Schema_Attribute{
+					{
+						Name:     "attr",
+						Type:     []byte(`"string"`),
+						Required: true,
+					},
+				},
+			},
+		},
+		ResourceSchemas: map[string]*proto.Schema{
+			"resource": &proto.Schema{
+				Version: 1,
+				Block: &proto.Schema_Block{
+					Attributes: []*proto.Schema_Attribute{
+						{
+							Name:     "attr",
+							Type:     []byte(`"string"`),
+							Required: true,
+						},
+					},
+				},
+			},
+		},
+		DataSourceSchemas: map[string]*proto.Schema{
+			"data": &proto.Schema{
+				Version: 1,
+				Block: &proto.Schema_Block{
+					Attributes: []*proto.Schema_Attribute{
+						{
+							Name:     "attr",
+							Type:     []byte(`"string"`),
+							Required: true,
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func TestGRPCProvider_GetSchema(t *testing.T) {
+	p := &GRPCProvider{
+		client: mockProviderClient(t),
+	}
+
+	resp := p.GetSchema()
+	checkDiags(t, resp.Diagnostics)
+}
+
+func TestGRPCProvider_ValidateProviderConfig(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().ValidateProviderConfig(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.ValidateProviderConfig_Response{}, nil)
+
+	cfg := hcl2shim.HCL2ValueFromConfigValue(map[string]interface{}{"attr": "value"})
+	resp := p.ValidateProviderConfig(providers.ValidateProviderConfigRequest{Config: cfg})
+	checkDiags(t, resp.Diagnostics)
+}
+
+func TestGRPCProvider_ValidateResourceTypeConfig(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().ValidateResourceTypeConfig(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.ValidateResourceTypeConfig_Response{}, nil)
+
+	cfg := hcl2shim.HCL2ValueFromConfigValue(map[string]interface{}{"attr": "value"})
+	resp := p.ValidateResourceTypeConfig(providers.ValidateResourceTypeConfigRequest{
+		TypeName: "resource",
+		Config:   cfg,
+	})
+	checkDiags(t, resp.Diagnostics)
+}
+
+func TestGRPCProvider_ValidateDataSourceConfig(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().ValidateDataSourceConfig(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.ValidateDataSourceConfig_Response{}, nil)
+
+	cfg := hcl2shim.HCL2ValueFromConfigValue(map[string]interface{}{"attr": "value"})
+	resp := p.ValidateDataSourceConfig(providers.ValidateDataSourceConfigRequest{
+		TypeName: "data",
+		Config:   cfg,
+	})
+	checkDiags(t, resp.Diagnostics)
+}
+
+func TestGRPCProvider_UpgradeResourceState(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().UpgradeResourceState(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.UpgradeResourceState_Response{
+		UpgradedState: &proto.DynamicValue{
+			Msgpack: []byte("\x81\xa4attr\xa3bar"),
+		},
+	}, nil)
+
+	resp := p.UpgradeResourceState(providers.UpgradeResourceStateRequest{
+		TypeName:     "resource",
+		Version:      0,
+		RawStateJSON: []byte(`{"old_attr":"bar"}`),
+	})
+	checkDiags(t, resp.Diagnostics)
+
+	expected := cty.ObjectVal(map[string]cty.Value{
+		"attr": cty.StringVal("bar"),
+	})
+
+	if !cmp.Equal(expected, resp.UpgradedState, typeComparer, valueComparer, equateEmpty) {
+		t.Fatal(cmp.Diff(expected, resp.UpgradedState, typeComparer, valueComparer, equateEmpty))
+	}
+}
+
+func TestGRPCProvider_Configure(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().Configure(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.Configure_Response{}, nil)
+
+	resp := p.Configure(providers.ConfigureRequest{
+		Config: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("foo"),
+		}),
+	})
+	checkDiags(t, resp.Diagnostics)
+}
+
+func TestGRPCProvider_Stop(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().Stop(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.Stop_Response{}, nil)
+
+	err := p.Stop()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestGRPCProvider_ReadResource(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().ReadResource(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.ReadResource_Response{
+		NewState: &proto.DynamicValue{
+			Msgpack: []byte("\x81\xa4attr\xa3bar"),
+		},
+	}, nil)
+
+	resp := p.ReadResource(providers.ReadResourceRequest{
+		TypeName: "resource",
+		PriorState: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("foo"),
+		}),
+	})
+
+	checkDiags(t, resp.Diagnostics)
+
+	expected := cty.ObjectVal(map[string]cty.Value{
+		"attr": cty.StringVal("bar"),
+	})
+
+	if !cmp.Equal(expected, resp.NewState, typeComparer, valueComparer, equateEmpty) {
+		t.Fatal(cmp.Diff(expected, resp.NewState, typeComparer, valueComparer, equateEmpty))
+	}
+}
+
+func TestGRPCProvider_PlanResourceChange(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	expectedPrivate := []byte(`{"meta": "data"}`)
+
+	client.EXPECT().PlanResourceChange(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.PlanResourceChange_Response{
+		PlannedState: &proto.DynamicValue{
+			Msgpack: []byte("\x81\xa4attr\xa3bar"),
+		},
+		RequiresReplace: []*proto.AttributePath{
+			{
+				Steps: []*proto.AttributePath_Step{
+					{
+						Selector: &proto.AttributePath_Step_AttributeName{
+							AttributeName: "attr",
+						},
+					},
+				},
+			},
+		},
+		PlannedPrivate: expectedPrivate,
+	}, nil)
+
+	resp := p.PlanResourceChange(providers.PlanResourceChangeRequest{
+		TypeName: "resource",
+		PriorState: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("foo"),
+		}),
+		ProposedNewState: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("bar"),
+		}),
+		Config: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("bar"),
+		}),
+	})
+
+	checkDiags(t, resp.Diagnostics)
+
+	expectedState := cty.ObjectVal(map[string]cty.Value{
+		"attr": cty.StringVal("bar"),
+	})
+
+	if !cmp.Equal(expectedState, resp.PlannedState, typeComparer, valueComparer, equateEmpty) {
+		t.Fatal(cmp.Diff(expectedState, resp.PlannedState, typeComparer, valueComparer, equateEmpty))
+	}
+
+	expectedReplace := `[]cty.Path{cty.Path{cty.GetAttrStep{Name:"attr"}}}`
+	replace := fmt.Sprintf("%#v", resp.RequiresReplace)
+	if expectedReplace != replace {
+		t.Fatalf("expected %q, got %q", expectedReplace, replace)
+	}
+
+	if !bytes.Equal(expectedPrivate, resp.PlannedPrivate) {
+		t.Fatalf("expected %q, got %q", expectedPrivate, resp.PlannedPrivate)
+	}
+}
+
+func TestGRPCProvider_ApplyResourceChange(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	expectedPrivate := []byte(`{"meta": "data"}`)
+
+	client.EXPECT().ApplyResourceChange(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.ApplyResourceChange_Response{
+		NewState: &proto.DynamicValue{
+			Msgpack: []byte("\x81\xa4attr\xa3bar"),
+		},
+		Private: expectedPrivate,
+	}, nil)
+
+	resp := p.ApplyResourceChange(providers.ApplyResourceChangeRequest{
+		TypeName: "resource",
+		PriorState: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("foo"),
+		}),
+		PlannedState: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("bar"),
+		}),
+		Config: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("bar"),
+		}),
+		PlannedPrivate: expectedPrivate,
+	})
+
+	checkDiags(t, resp.Diagnostics)
+
+	expectedState := cty.ObjectVal(map[string]cty.Value{
+		"attr": cty.StringVal("bar"),
+	})
+
+	if !cmp.Equal(expectedState, resp.NewState, typeComparer, valueComparer, equateEmpty) {
+		t.Fatal(cmp.Diff(expectedState, resp.NewState, typeComparer, valueComparer, equateEmpty))
+	}
+
+	if !bytes.Equal(expectedPrivate, resp.Private) {
+		t.Fatalf("expected %q, got %q", expectedPrivate, resp.Private)
+	}
+}
+
+func TestGRPCProvider_ImportResourceState(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	expectedPrivate := []byte(`{"meta": "data"}`)
+
+	client.EXPECT().ImportResourceState(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.ImportResourceState_Response{
+		ImportedResources: []*proto.ImportResourceState_ImportedResource{
+			{
+				TypeName: "resource",
+				State: &proto.DynamicValue{
+					Msgpack: []byte("\x81\xa4attr\xa3bar"),
+				},
+				Private: expectedPrivate,
+			},
+		},
+	}, nil)
+
+	resp := p.ImportResourceState(providers.ImportResourceStateRequest{
+		TypeName: "resource",
+		ID:       "foo",
+	})
+
+	checkDiags(t, resp.Diagnostics)
+
+	expectedResource := providers.ImportedResource{
+		TypeName: "resource",
+		State: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("bar"),
+		}),
+		Private: expectedPrivate,
+	}
+
+	imported := resp.ImportedResources[0]
+	if !cmp.Equal(expectedResource, imported, typeComparer, valueComparer, equateEmpty) {
+		t.Fatal(cmp.Diff(expectedResource, imported, typeComparer, valueComparer, equateEmpty))
+	}
+}
+
+func TestGRPCProvider_ReadDataSource(t *testing.T) {
+	client := mockProviderClient(t)
+	p := &GRPCProvider{
+		client: client,
+	}
+
+	client.EXPECT().ReadDataSource(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.ReadDataSource_Response{
+		State: &proto.DynamicValue{
+			Msgpack: []byte("\x81\xa4attr\xa3bar"),
+		},
+	}, nil)
+
+	resp := p.ReadDataSource(providers.ReadDataSourceRequest{
+		TypeName: "data",
+		Config: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("foo"),
+		}),
+	})
+
+	checkDiags(t, resp.Diagnostics)
+
+	expected := cty.ObjectVal(map[string]cty.Value{
+		"attr": cty.StringVal("bar"),
+	})
+
+	if !cmp.Equal(expected, resp.State, typeComparer, valueComparer, equateEmpty) {
+		t.Fatal(cmp.Diff(expected, resp.State, typeComparer, valueComparer, equateEmpty))
+	}
+}
diff --git a/plugin/grpc_provisioner.go b/plugin/grpc_provisioner.go
new file mode 100644
index 0000000000..b2ad0f20b4
--- /dev/null
+++ b/plugin/grpc_provisioner.go
@@ -0,0 +1,146 @@
+package plugin
+
+import (
+	"context"
+	"errors"
+	"io"
+	"sync"
+
+	"github.com/hashicorp/terraform/configs/configschema"
+	"github.com/hashicorp/terraform/plugin/proto"
+	"github.com/hashicorp/terraform/provisioners"
+	"github.com/zclconf/go-cty/cty"
+	"github.com/zclconf/go-cty/cty/msgpack"
+	"google.golang.org/grpc"
+)
+
+// provisioners.Interface grpc implementation
+type GRPCProvisioner struct {
+	conn   *grpc.ClientConn
+	client proto.ProvisionerClient
+	ctx    context.Context
+
+	// Cache the schema since we need it for serialization in each method call.
+	mu     sync.Mutex
+	schema *configschema.Block
+}
+
+func (p *GRPCProvisioner) GetSchema() (resp provisioners.GetSchemaResponse) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.schema != nil {
+		return provisioners.GetSchemaResponse{
+			Provisioner: p.schema,
+		}
+	}
+
+	protoResp, err := p.client.GetSchema(p.ctx, new(proto.GetProvisionerSchema_Request))
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	if protoResp.Provisioner == nil {
+		resp.Diagnostics = resp.Diagnostics.Append(errors.New("missing provisioner schema"))
+		return resp
+	}
+
+	resp.Provisioner = schemaBlock(protoResp.Provisioner.Block)
+
+	p.schema = resp.Provisioner
+
+	return resp
+}
+
+func (p *GRPCProvisioner) ValidateProvisionerConfig(r provisioners.ValidateProvisionerConfigRequest) (resp provisioners.ValidateProvisionerConfigResponse) {
+	schema := p.GetSchema()
+	if schema.Diagnostics.HasErrors() {
+		resp.Diagnostics = resp.Diagnostics.Append(schema.Diagnostics)
+		return resp
+	}
+
+	mp, err := msgpack.Marshal(r.Config, schema.Provisioner.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.ValidateProvisionerConfig_Request{
+		Config: &proto.DynamicValue{Msgpack: mp},
+	}
+	protoResp, err := p.client.ValidateProvisionerConfig(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+	resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(protoResp.Diagnostics))
+	return resp
+}
+
+func (p *GRPCProvisioner) ProvisionResource(r provisioners.ProvisionResourceRequest) (resp provisioners.ProvisionResourceResponse) {
+	schema := p.GetSchema()
+	if schema.Diagnostics.HasErrors() {
+		resp.Diagnostics = resp.Diagnostics.Append(schema.Diagnostics)
+		return resp
+	}
+
+	mp, err := msgpack.Marshal(r.Config, schema.Provisioner.ImpliedType())
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	// connection is always assumed to be a simple string map
+	connMP, err := msgpack.Marshal(r.Connection, cty.Map(cty.String))
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	protoReq := &proto.ProvisionResource_Request{
+		Config:     &proto.DynamicValue{Msgpack: mp},
+		Connection: &proto.DynamicValue{Msgpack: connMP},
+	}
+
+	outputClient, err := p.client.ProvisionResource(p.ctx, protoReq)
+	if err != nil {
+		resp.Diagnostics = resp.Diagnostics.Append(err)
+		return resp
+	}
+
+	for {
+		rcv, err := outputClient.Recv()
+		if rcv != nil {
+			r.UIOutput.Output(rcv.Output)
+		}
+		if err != nil {
+			if err != io.EOF {
+				resp.Diagnostics = resp.Diagnostics.Append(err)
+			}
+			break
+		}
+
+		if len(rcv.Diagnostics) > 0 {
+			resp.Diagnostics = resp.Diagnostics.Append(ProtoToDiagnostics(rcv.Diagnostics))
+			break
+		}
+	}
+
+	return resp
+}
+
+func (p *GRPCProvisioner) Stop() error {
+	protoResp, err := p.client.Stop(p.ctx, &proto.Stop_Request{})
+	if err != nil {
+		return err
+	}
+	if protoResp.Error != "" {
+		return errors.New(protoResp.Error)
+	}
+	return nil
+}
+
+func (p *GRPCProvisioner) Close() error {
+	return nil
+}
diff --git a/plugin/grpc_provisioner_test.go b/plugin/grpc_provisioner_test.go
new file mode 100644
index 0000000000..c26b3edef2
--- /dev/null
+++ b/plugin/grpc_provisioner_test.go
@@ -0,0 +1,138 @@
+package plugin
+
+import (
+	"io"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/hashicorp/terraform/config/hcl2shim"
+	"github.com/hashicorp/terraform/plugin/proto"
+	"github.com/hashicorp/terraform/provisioners"
+	"github.com/zclconf/go-cty/cty"
+
+	mockproto "github.com/hashicorp/terraform/plugin/mock_proto"
+)
+
+var _ provisioners.Interface = (*GRPCProvisioner)(nil)
+
+func mockProvisionerClient(t *testing.T) *mockproto.MockProvisionerClient {
+	ctrl := gomock.NewController(t)
+	client := mockproto.NewMockProvisionerClient(ctrl)
+
+	// we always need a GetSchema method
+	client.EXPECT().GetSchema(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(provisionerProtoSchema(), nil)
+
+	return client
+}
+
+func provisionerProtoSchema() *proto.GetProvisionerSchema_Response {
+	return &proto.GetProvisionerSchema_Response{
+		Provisioner: &proto.Schema{
+			Block: &proto.Schema_Block{
+				Attributes: []*proto.Schema_Attribute{
+					{
+						Name:     "attr",
+						Type:     []byte(`"string"`),
+						Required: true,
+					},
+				},
+			},
+		},
+	}
+}
+
+func TestGRPCProvisioner_GetSchema(t *testing.T) {
+	p := &GRPCProvisioner{
+		client: mockProvisionerClient(t),
+	}
+
+	resp := p.GetSchema()
+	checkDiags(t, resp.Diagnostics)
+}
+
+func TestGRPCProvisioner_ValidateProvisionerConfig(t *testing.T) {
+	client := mockProvisionerClient(t)
+	p := &GRPCProvisioner{
+		client: client,
+	}
+
+	client.EXPECT().ValidateProvisionerConfig(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.ValidateProvisionerConfig_Response{}, nil)
+
+	cfg := hcl2shim.HCL2ValueFromConfigValue(map[string]interface{}{"attr": "value"})
+	resp := p.ValidateProvisionerConfig(provisioners.ValidateProvisionerConfigRequest{Config: cfg})
+	checkDiags(t, resp.Diagnostics)
+}
+
+func TestGRPCProvisioner_ProvisionResource(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	client := mockproto.NewMockProvisionerClient(ctrl)
+
+	// we always need a GetSchema method
+	client.EXPECT().GetSchema(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(provisionerProtoSchema(), nil)
+
+	stream := mockproto.NewMockProvisioner_ProvisionResourceClient(ctrl)
+	stream.EXPECT().Recv().Return(&proto.ProvisionResource_Response{
+		Output: "provisioned",
+	}, io.EOF)
+
+	client.EXPECT().ProvisionResource(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(stream, nil)
+
+	p := &GRPCProvisioner{
+		client: client,
+	}
+
+	rec := &provisionRecorder{}
+
+	resp := p.ProvisionResource(provisioners.ProvisionResourceRequest{
+		Config: cty.ObjectVal(map[string]cty.Value{
+			"attr": cty.StringVal("value"),
+		}),
+		Connection: cty.EmptyObjectVal,
+		UIOutput:   rec,
+	})
+
+	if resp.Diagnostics.HasErrors() {
+		t.Fatal(resp.Diagnostics.Err())
+	}
+
+	if len(rec.output) == 0 || rec.output[0] != "provisioned" {
+		t.Fatalf("expected %q, got %q", []string{"provisioned"}, rec.output)
+	}
+}
+
+type provisionRecorder struct {
+	output []string
+}
+
+func (r *provisionRecorder) Output(s string) {
+	r.output = append(r.output, s)
+}
+
+func TestGRPCProvisioner_Stop(t *testing.T) {
+	client := mockProvisionerClient(t)
+	p := &GRPCProvisioner{
+		client: client,
+	}
+
+	client.EXPECT().Stop(
+		gomock.Any(),
+		gomock.Any(),
+	).Return(&proto.Stop_Response{}, nil)
+
+	err := p.Stop()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/plugin/resource_provisioner.go b/plugin/resource_provisioner.go
index 8fce9d8ae7..7ef435454b 100644
--- a/plugin/resource_provisioner.go
+++ b/plugin/resource_provisioner.go
@@ -4,6 +4,7 @@ import (
 	"net/rpc"
 
 	"github.com/hashicorp/go-plugin"
+	"github.com/hashicorp/terraform/configs/configschema"
 	"github.com/hashicorp/terraform/terraform"
 )
 
@@ -28,6 +29,11 @@ type ResourceProvisioner struct {
 	Client *rpc.Client
 }
 
+func (p *ResourceProvisioner) GetConfigSchema() (*configschema.Block, error) {
+	panic("not implemented")
+	return nil, nil
+}
+
 func (p *ResourceProvisioner) Validate(c *terraform.ResourceConfig) ([]string, []error) {
 	var resp ResourceProvisionerValidateResponse
 	args := ResourceProvisionerValidateArgs{