diff --git a/examples/modular/models/BUILD.gn b/examples/modular/models/BUILD.gn
index c5941ae..18bcbcb 100644
--- a/examples/modular/models/BUILD.gn
+++ b/examples/modular/models/BUILD.gn
@@ -12,7 +12,6 @@
   sources = [
     "shape.dart",
     "src/shape.dart",
-    "src/shape_codec.dart",
   ]
 
   deps = [
diff --git a/examples/modular/models/lib/shape.dart b/examples/modular/models/lib/shape.dart
index 7cd7d32..99e76c5 100644
--- a/examples/modular/models/lib/shape.dart
+++ b/examples/modular/models/lib/shape.dart
@@ -3,4 +3,3 @@
 // found in the LICENSE file.
 
 export 'src/shape.dart';
-export 'src/shape_codec.dart';
diff --git a/examples/modular/models/lib/src/shape.dart b/examples/modular/models/lib/src/shape.dart
index e23b5b2..c7b13a6 100644
--- a/examples/modular/models/lib/src/shape.dart
+++ b/examples/modular/models/lib/src/shape.dart
@@ -2,8 +2,12 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+import 'dart:typed_data';
+
 /// A Simple model object representing a shape
 class Shape {
+  static const String entityType = 'com.fuchsia.shapes_mod.shape';
+
   /// The size of the shape
   double size = 0.0;
 
@@ -15,4 +19,16 @@
 
   /// The default constructor
   Shape(this.size);
+
+  Shape.fromBytes(Uint8List bytes) {
+    size = ByteData.view(bytes.buffer).getFloat64(0);
+  }
+
+  Uint8List toBytes() {
+    final list = Uint8List(8);
+
+    ByteData.view(list.buffer).setFloat64(0, size);
+
+    return list;
+  }
 }
diff --git a/examples/modular/models/lib/src/shape_codec.dart b/examples/modular/models/lib/src/shape_codec.dart
deleted file mode 100644
index dd486d2..0000000
--- a/examples/modular/models/lib/src/shape_codec.dart
+++ /dev/null
@@ -1,32 +0,0 @@
-// Copyright 2018 The Fuchsia Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-import 'dart:typed_data';
-
-import 'package:fuchsia_modular/entity.dart';
-
-import 'shape.dart';
-
-/// An [EntityCodec] which encodes and decodes a shape object
-class ShapeCodec extends SimpleEntityCodec<Shape> {
-  ShapeCodec()
-      : super(
-            type: 'com.fuchsia.shapes_mod.shape',
-            encoding: '*',
-            encode: _encode,
-            decode: _decode);
-
-  static Uint8List _encode(Shape shape) {
-    final list = Uint8List(8);
-
-    ByteData.view(list.buffer).setFloat64(0, shape.size);
-
-    return list;
-  }
-
-  static Shape _decode(Uint8List list) {
-    final size = ByteData.view(list.buffer).getFloat64(0);
-    return Shape(size);
-  }
-}
diff --git a/examples/modular/shapes_mod/lib/src/intent_handlers/root_intent_handler.dart b/examples/modular/shapes_mod/lib/src/intent_handlers/root_intent_handler.dart
index 1207504..4a082ce 100644
--- a/examples/modular/shapes_mod/lib/src/intent_handlers/root_intent_handler.dart
+++ b/examples/modular/shapes_mod/lib/src/intent_handlers/root_intent_handler.dart
@@ -15,9 +15,9 @@
   @override
   void handleIntent(Intent intent) {
     final stream = intent
-        .getEntity(name: 'shape', codec: ShapeCodec())
+        .getEntity(name: 'shape', type: Shape.entityType)
         .watch()
-        .cast<Shape>();
+        .map((b) => Shape.fromBytes(b));
 
     if (intent.action == _circleAction) {
       CircleRenderer().render(stream);
diff --git a/examples/modular/shapes_mod/meta/shapes_mod.cmx b/examples/modular/shapes_mod/meta/shapes_mod.cmx
index d297d04..be82af8 100644
--- a/examples/modular/shapes_mod/meta/shapes_mod.cmx
+++ b/examples/modular/shapes_mod/meta/shapes_mod.cmx
@@ -9,6 +9,7 @@
             "fuchsia.logger.LogSink",
             "fuchsia.modular.Clipboard",
             "fuchsia.modular.ContextWriter",
+            "fuchsia.modular.ComponentContext",
             "fuchsia.modular.ModuleContext",
             "fuchsia.netstack.Netstack",
             "fuchsia.sys.Environment",
diff --git a/examples/modular/slider_mod/lib/src/blocs/app_bloc.dart b/examples/modular/slider_mod/lib/src/blocs/app_bloc.dart
index c23045e..2411a38 100644
--- a/examples/modular/slider_mod/lib/src/blocs/app_bloc.dart
+++ b/examples/modular/slider_mod/lib/src/blocs/app_bloc.dart
@@ -13,18 +13,22 @@
 
   AppBloc(this.shapeEntity);
 
-  void launchSquare() => Module().addModuleToStory(
-      name: 'shape_module',
-      intent: _makeIntent('com.fuchsia.shapes_mod.show_square'));
+  void launchSquare() =>
+      _launchModule('shape_module', 'com.fuchsia.shapes_mod.show_square');
 
-  void launchCircle() => Module().addModuleToStory(
-      name: 'shape_module',
-      intent: _makeIntent('com.fuchsia.shapes_mod.show_circle'));
+  void launchCircle() =>
+      _launchModule('shape_module', 'com.fuchsia.shapes_mod.show_circle');
 
-  Intent _makeIntent(String action) => Intent(
+  void _launchModule(String name, String action) {
+    _makeIntent(action).then(
+        (intent) => Module().addModuleToStory(name: name, intent: intent));
+  }
+
+  Future<Intent> _makeIntent(String action) async => Intent(
         action: action,
-        handler: 'shapes_mod',
-      )..addParameterFromEntity('shape', shapeEntity);
+        handler: 'fuchsia-pkg://fuchsia.com/shapes_mod#meta/shapes_mod.cmx',
+      )..addParameterFromEntityReference(
+          'shape', await shapeEntity.getEntityReference());
 
   @override
   void dispose() {}
diff --git a/examples/modular/slider_mod/lib/src/blocs/slider_bloc.dart b/examples/modular/slider_mod/lib/src/blocs/slider_bloc.dart
index ab6e744..dce6ba5 100644
--- a/examples/modular/slider_mod/lib/src/blocs/slider_bloc.dart
+++ b/examples/modular/slider_mod/lib/src/blocs/slider_bloc.dart
@@ -12,7 +12,7 @@
 /// The [SliderBloc] provides actions and streams associated with
 /// the onscreen slider.
 class SliderBloc implements BlocBase {
-  final Entity<Shape> shapeEntity;
+  final Entity shapeEntity;
   final Shape shape;
 
   final _valueController = StreamController<double>.broadcast();
@@ -34,6 +34,6 @@
     _valueController.add(newValue);
     shape.size = newValue;
 
-    shapeEntity.write(shape);
+    shapeEntity.write(shape.toBytes());
   }
 }
diff --git a/examples/modular/slider_mod/lib/src/handlers/root_intent_handler.dart b/examples/modular/slider_mod/lib/src/handlers/root_intent_handler.dart
index 4b5b08f..0d1680e 100644
--- a/examples/modular/slider_mod/lib/src/handlers/root_intent_handler.dart
+++ b/examples/modular/slider_mod/lib/src/handlers/root_intent_handler.dart
@@ -13,14 +13,16 @@
 
 class RootIntentHandler extends IntentHandler {
   @override
-  void handleIntent(Intent intent) {
+  void handleIntent(Intent intent) async {
     // if (intent.action != '') {
     //   throw Exception('Unknown intent action');
     // }
 
     final shape = Shape(5.0);
-
-    final shapeEntity = Entity(codec: ShapeCodec())..write(shape);
+    final shapeEntity = await Entity.createStoryScoped(
+      type: Shape.entityType,
+      initialData: shape.toBytes(),
+    );
 
     runApp(
       MaterialApp(
diff --git a/examples/modular/slider_mod/lib/src/widgets/slider_scaffold.dart b/examples/modular/slider_mod/lib/src/widgets/slider_scaffold.dart
index c860b8c..0634ae6 100644
--- a/examples/modular/slider_mod/lib/src/widgets/slider_scaffold.dart
+++ b/examples/modular/slider_mod/lib/src/widgets/slider_scaffold.dart
@@ -12,7 +12,7 @@
 import 'value_slider.dart';
 
 class SliderScaffold extends StatelessWidget {
-  final Entity<Shape> shapeEntity;
+  final Entity shapeEntity;
   final Shape shape;
 
   const SliderScaffold(this.shapeEntity, this.shape);
diff --git a/examples/modular/slider_mod/meta/slider_mod.cmx b/examples/modular/slider_mod/meta/slider_mod.cmx
index bdc92e8..5318f02 100644
--- a/examples/modular/slider_mod/meta/slider_mod.cmx
+++ b/examples/modular/slider_mod/meta/slider_mod.cmx
@@ -21,7 +21,8 @@
     "facets": {
         "fuchsia.module": {
             "@version": 2,
-            "suggestion_headline": "Show a slider"
+            "suggestion_headline": "Show a slider",
+            "intent_filters": []
         }
     }
 }
diff --git a/public/dart/fuchsia_modular/BUILD.gn b/public/dart/fuchsia_modular/BUILD.gn
index 4cbef16..c8e22ef 100644
--- a/public/dart/fuchsia_modular/BUILD.gn
+++ b/public/dart/fuchsia_modular/BUILD.gn
@@ -17,7 +17,6 @@
     "agent.dart",
     "codecs.dart",
     "entity.dart",
-    "entity_codec.dart",
     "lifecycle.dart",
     "logger.dart",
     "module.dart",
@@ -28,8 +27,8 @@
     "src/agent/internal/_agent_context.dart",
     "src/agent/internal/_agent_impl.dart",
     "src/entity/entity.dart",
-    "src/entity/entity_codec.dart",
-    "src/entity/internal/_link_entity.dart",
+    "src/entity/entity_exceptions.dart",
+    "src/entity/internal/_entity_impl.dart",
     "src/internal/_component_context.dart",
     "src/lifecycle/internal/_lifecycle_impl.dart",
     "src/lifecycle/lifecycle.dart",
@@ -53,8 +52,8 @@
     "//garnet/public/fidl/fuchsia.sys",
     "//garnet/public/fidl/fuchsia.ui.viewsv1token",
     "//peridot/public/fidl/fuchsia.modular",
+    "//third_party/dart-pkg/pub/async",
     "//third_party/dart-pkg/pub/meta",
-    "//third_party/dart-pkg/pub/uuid",
     "//topaz/public/dart/fidl",
     "//topaz/public/dart/fuchsia",
     "//topaz/public/dart/fuchsia_logger",
@@ -105,6 +104,7 @@
   ]
 
   sources = [
+    "entity/internal/entity_impl_test.dart",
     "internal/component_context_integ_test.dart",
     "lifecycle/internal/lifecycle_impl_test.dart",
     "module/internal/module_impl_integ_test.dart",
@@ -113,6 +113,7 @@
 
   deps = [
     ":fuchsia_modular",
+    "//garnet/public/fidl/fuchsia.testing.runner",
     "//third_party/dart-pkg/pub/mockito",  # Remove after DX-470 is fixed
     "//third_party/dart-pkg/pub/test",
   ]
diff --git a/public/dart/fuchsia_modular/lib/entity.dart b/public/dart/fuchsia_modular/lib/entity.dart
index c88f58a..9c4782e 100644
--- a/public/dart/fuchsia_modular/lib/entity.dart
+++ b/public/dart/fuchsia_modular/lib/entity.dart
@@ -4,4 +4,4 @@
 
 /// The classes required for working with enities
 export 'src/entity/entity.dart';
-export 'src/entity/entity_codec.dart';
+export 'src/entity/entity_exceptions.dart';
diff --git a/public/dart/fuchsia_modular/lib/entity_codec.dart b/public/dart/fuchsia_modular/lib/entity_codec.dart
deleted file mode 100644
index 63facf2..0000000
--- a/public/dart/fuchsia_modular/lib/entity_codec.dart
+++ /dev/null
@@ -1,6 +0,0 @@
-// Copyright 2018 The Fuchsia Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-/// The base class for implementing typed codecs for structured entity values.
-library entity_codec;
diff --git a/public/dart/fuchsia_modular/lib/src/entity/entity.dart b/public/dart/fuchsia_modular/lib/src/entity/entity.dart
index 6c0f51d..89c6e6f 100644
--- a/public/dart/fuchsia_modular/lib/src/entity/entity.dart
+++ b/public/dart/fuchsia_modular/lib/src/entity/entity.dart
@@ -3,35 +3,72 @@
 // found in the LICENSE file.
 
 import 'dart:async';
+import 'dart:typed_data';
 
+import 'package:fidl_fuchsia_mem/fidl_async.dart' as fuchsia_mem;
+import 'package:fidl_fuchsia_modular/fidl_async.dart' as fidl_modular;
 import 'package:meta/meta.dart';
-import 'package:uuid/uuid.dart';
+import 'package:zircon/zircon.dart';
 
-import 'entity_codec.dart';
-import 'internal/_link_entity.dart';
+import '../internal/_component_context.dart';
+import '../module/internal/_module_context.dart';
+import 'entity_exceptions.dart';
+import 'internal/_entity_impl.dart';
 
 /// An [Entity] provides a mechanism for communicating
-/// data between compoonents.
+/// data between components.
 ///
 /// Note: this is a preliminary API that is likely to change.
 @experimental
-abstract class Entity<T> {
-  /// Creates an entity that will live for the scope of this story.
-  /// The entity that is created will be backed by the framework and
-  /// can be treated as if it was received from any other entity provider.
+abstract class Entity {
+  /// Creates an Entity instance.
+  ///
+  /// This method will lazily connect to the entity proxy for data transmission.
+  /// This object should be treated like it has a valid connection until
+  /// otherwise informed via a failed future.
+  ///
+  /// ```
+  ///   final entity = Entity(entityReference: 'foo', type: 'com.foo.bar');
+  ///   // fetch the data assuming that the entity resolved correctly. If it
+  ///   // did not the call to getData() will fail.
+  ///   final data = await entity.getData();
+  /// ```
   factory Entity({
-    @required EntityCodec<T> codec,
+    @required String entityReference,
+    @required String type,
   }) {
-    // This is temporary and go away when we remove link entities.
-    final linkName = Uuid().v4().toString();
-    return LinkEntity<T>(linkName: linkName, codec: codec);
+    ArgumentError.checkNotNull(entityReference, 'entityReference');
+    ArgumentError.checkNotNull(type, 'type');
+
+    return EntityImpl(
+      proxyFactory: () async {
+        final resolver = fidl_modular.EntityResolverProxy();
+        await getComponentContext().getEntityResolver(resolver.ctrl.request());
+
+        final proxy = fidl_modular.EntityProxy();
+        await resolver.resolveEntity(entityReference, proxy.ctrl.request());
+
+        final types = await proxy.getTypes();
+        if (!types.contains(type)) {
+          throw EntityTypeException(type);
+        }
+
+        return proxy;
+      },
+      type: type,
+    );
   }
 
-  /// Returns the data stored in the entity.
-  Future<T> getData();
+  /// The type of data that this object represents.
+  String get type;
 
-  /// Writes the object stored in value
-  Future<void> write(T object);
+  /// Returns the data stored in the entity.
+  Future<Uint8List> getData();
+
+  /// Returns the reference for this entity. Entity references will never change
+  /// for a given entity so this value can be cached and used to access the
+  /// entity from a different process or at a later time.
+  Future<String> getEntityReference();
 
   /// Watches the entity for updates.
   ///
@@ -41,19 +78,40 @@
   /// The returned stream is a single subscription stream
   /// which, when closed, will close the underlying fidl
   /// connection.
-  Stream<T> watch();
-}
+  Stream<Uint8List> watch();
 
-/// An exception which is thrown when an Entity does not
-/// support a given type.
-class EntityTypeException implements Exception {
-  /// The unsuported type.
-  final String type;
+  /// Writes the object stored in value
+  Future<void> write(Uint8List object);
 
-  /// Create a new [EntityTypeException].
-  EntityTypeException(this.type);
+  /// Creates an entity that will live for the scope of this story.
+  /// The entity that is created will be backed by the framework and
+  /// can be treated as if it was received from any other entity provider.
+  static Future<Entity> createStoryScoped({
+    @required String type,
+    @required Uint8List initialData,
+  }) async {
+    ArgumentError.checkNotNull(type, 'type');
+    ArgumentError.checkNotNull(initialData, 'initialData');
 
-  @override
-  String toString() =>
-      'EntityTypeError: type "$type" is not available for Entity';
+    if (type.isEmpty) {
+      throw ArgumentError.value(type, 'type cannot be an empty string');
+    }
+
+    final context = getModuleContext();
+
+    // need to create the proxy and write data immediately so other modules
+    // can extract values
+    final proxy = fidl_modular.EntityProxy();
+    final vmo = SizedVmo.fromUint8List(initialData);
+    final buffer = fuchsia_mem.Buffer(vmo: vmo, size: initialData.length);
+    final ref = await context.createEntity(type, buffer, proxy.ctrl.request());
+
+    // use the ref value to determine if creation was successful
+    if (ref == null || ref.isEmpty) {
+      throw Exception('Entity.createStoryScopedentity creation failed because'
+          ' the framework was unable to create the entity.');
+    }
+
+    return EntityImpl(type: type, proxyFactory: () => proxy);
+  }
 }
diff --git a/public/dart/fuchsia_modular/lib/src/entity/entity_codec.dart b/public/dart/fuchsia_modular/lib/src/entity/entity_codec.dart
deleted file mode 100644
index 0403ea4..0000000
--- a/public/dart/fuchsia_modular/lib/src/entity/entity_codec.dart
+++ /dev/null
@@ -1,121 +0,0 @@
-// Copyright 2018 The Fuchsia Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-import 'dart:async';
-import 'dart:convert';
-import 'dart:typed_data';
-
-import 'package:meta/meta.dart';
-
-/// The [EntityCodec] is a typed [Codec] which translates
-/// entity data between the [Uint8List] wire format and a
-/// dart object.
-///
-/// [EntityCodec]s have a specific semantic type and encoding
-/// which allow them to operate with the entity system.
-abstract class EntityCodec<T> extends Codec<T, Uint8List> {
-  /// The semantic type of the entity which this codec supports.
-  /// This type will be used when requesting entities from the
-  /// framework.
-  final String type;
-
-  /// The encoding that this [EntityCodec] knows how to encode/decode data
-  final String encoding;
-
-  /// The default constructor.
-  const EntityCodec({
-    @required this.type,
-    @required this.encoding,
-  })  : assert(type != null),
-        assert(encoding != null);
-}
-
-/// A [Codec] used for handling the the automatic translation to and from an
-/// Entity's source data to the specified Dart type [T].
-class SimpleEntityCodec<T> extends EntityCodec<T> {
-  final _EntityEncoder<T> _encoder;
-  final _EntityDecoder<T> _decoder;
-
-  /// Create a new [EntityCodec].
-  SimpleEntityCodec({
-    @required String type,
-    @required String encoding,
-    @required _EncodeEntity<T> encode,
-    @required _DecodeEntity<T> decode,
-  })  : assert(type != null),
-        assert(encoding != null),
-        assert(encode != null),
-        assert(decode != null),
-        _encoder = _EntityEncoder<T>(encode),
-        _decoder = _EntityDecoder<T>(decode),
-        super(type: type, encoding: encoding);
-
-  @override
-  _EntityEncoder<T> get encoder => _encoder;
-
-  @override
-  _EntityDecoder<T> get decoder => _decoder;
-}
-
-typedef _EncodeEntity<T> = Uint8List Function(T value);
-
-class _EntityEncoder<T> extends Converter<T, Uint8List> {
-  final _EncodeEntity<T> encode;
-
-  const _EntityEncoder(this.encode);
-
-  @override
-  Uint8List convert(T source) {
-    return encode(source);
-  }
-}
-
-typedef _DecodeEntity<T> = T Function(Uint8List data);
-
-class _EntityDecoder<T> extends Converter<Uint8List, T> {
-  final _DecodeEntity<T> decode;
-
-  const _EntityDecoder(this.decode);
-
-  @override
-  T convert(Uint8List data) => decode(data);
-
-  @override
-  Stream<T> bind(Stream<Uint8List> source) {
-    _EntityDecoderSink<T> map(EventSink<T> out) =>
-        new _EntityDecoderSink<T>(out, this);
-
-    return new Stream<T>.eventTransformed(source, map);
-  }
-}
-
-/// Entity data [String]s in, [T] out.
-class _EntityDecoderSink<T> extends EventSink<Uint8List> {
-  /// The [EventSink] that values are decoded into. Errors generated by the
-  /// decoder are also added to [out].
-  final EventSink<T> out;
-
-  /// The decoder to used to convert the source ([Stream<String>]) events.
-  final _EntityDecoder<T> decoder;
-
-  /// Create an instance of [_EntityDecoderSink], usually via the
-  /// [Stream#eventTransformed] constructor.
-  _EntityDecoderSink(this.out, this.decoder);
-
-  @override
-  void add(Uint8List data) {
-    try {
-      T value = decoder.decode(data);
-      out.add(value);
-    } on Object catch (err, stackTrace) {
-      addError(err, stackTrace);
-    }
-  }
-
-  @override
-  void addError(Object e, [StackTrace s]) => out.addError(e, s);
-
-  @override
-  void close() => out.close();
-}
diff --git a/public/dart/fuchsia_modular/lib/src/entity/entity_exceptions.dart b/public/dart/fuchsia_modular/lib/src/entity/entity_exceptions.dart
new file mode 100644
index 0000000..7d49b12
--- /dev/null
+++ b/public/dart/fuchsia_modular/lib/src/entity/entity_exceptions.dart
@@ -0,0 +1,33 @@
+// Copyright 2018 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'package:fidl_fuchsia_modular/fidl_async.dart' as fidl_modular;
+
+/// An exception which is thrown when an Entity does not
+/// support a given type.
+class EntityTypeException implements Exception {
+  /// The unsuported type.
+  final String type;
+
+  /// Create a new [EntityTypeException].
+  EntityTypeException(this.type);
+
+  @override
+  String toString() =>
+      'EntityTypeError: type "$type" is not available for Entity';
+}
+
+/// An exception which is thrown when writes to an Entity
+/// fail with the give status.
+class EntityWriteException implements Exception {
+  /// The status code for this failure.
+  final fidl_modular.EntityWriteStatus status;
+
+  /// Create a new [EntityWriteException].
+  EntityWriteException(this.status);
+
+  @override
+  String toString() =>
+      'EntityWriteException: entity write failed with status $status';
+}
diff --git a/public/dart/fuchsia_modular/lib/src/entity/internal/_entity_impl.dart b/public/dart/fuchsia_modular/lib/src/entity/internal/_entity_impl.dart
new file mode 100644
index 0000000..1e6ab75
--- /dev/null
+++ b/public/dart/fuchsia_modular/lib/src/entity/internal/_entity_impl.dart
@@ -0,0 +1,113 @@
+// Copyright 2018 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:typed_data';
+
+import 'package:async/async.dart';
+import 'package:fidl/fidl.dart';
+import 'package:fidl_fuchsia_mem/fidl_async.dart' as fuchsia_mem;
+import 'package:fidl_fuchsia_modular/fidl_async.dart' as fidl_modular;
+import 'package:meta/meta.dart';
+import 'package:zircon/zircon.dart';
+
+import '../entity.dart';
+import '../entity_exceptions.dart';
+
+typedef EntityProxyFactory = FutureOr<fidl_modular.Entity> Function();
+
+/// A concrete implementation of the [Entity] class.
+class EntityImpl implements Entity {
+  final _proxyMemo = AsyncMemoizer<fidl_modular.Entity>();
+
+  @override
+  final String type;
+
+  final _entityReferenceMemo = AsyncMemoizer<String>();
+
+  /// A function which returns a bound proxy on demand. This method will only be
+  /// called once.
+  final EntityProxyFactory proxyFactory;
+
+  /// Constructs an instance of [EntityImpl].
+  EntityImpl({
+    @required this.proxyFactory,
+    @required this.type,
+  })  : assert(proxyFactory != null),
+        assert(type != null && type.isNotEmpty);
+
+  @override
+  Future<Uint8List> getData() =>
+      _getProxy().then((p) => p.getData(type)).then(_unwrapBuffer);
+
+  @override
+  Future<String> getEntityReference() => _entityReferenceMemo
+      .runOnce(() => _getProxy().then((p) => p.getReference()));
+
+  @override
+  Stream<Uint8List> watch() {
+    final controller = StreamController<fuchsia_mem.Buffer>();
+    final watcher = _EntityWatcher(controller.add);
+
+    // if the user closed the stream we close the binding to the watcher
+    controller.onCancel = watcher.binding.close;
+
+    // if the connection closes then we close the stream
+    watcher.binding.whenClosed.then((_) {
+      if (!controller.isClosed) {
+        controller.close();
+      }
+    });
+
+    _getProxy().then((proxy) {
+      proxy.watch(type, watcher.getInterfaceHandle());
+    });
+
+    return controller.stream.asyncMap(_unwrapBuffer);
+  }
+
+  @override
+  Future<void> write(Uint8List value) async {
+    ArgumentError.checkNotNull(value, 'value');
+
+    final vmo = SizedVmo.fromUint8List(value);
+    final buffer = fuchsia_mem.Buffer(vmo: vmo, size: value.length);
+
+    final proxy = await _getProxy();
+    final result = await proxy.writeData(type, buffer);
+    if (result != fidl_modular.EntityWriteStatus.ok) {
+      throw EntityWriteException(result);
+    }
+  }
+
+  Future<fidl_modular.Entity> _getProxy() => _proxyMemo.runOnce(proxyFactory);
+
+  Uint8List _unwrapBuffer(fuchsia_mem.Buffer buffer) {
+    final dataVmo = SizedVmo(buffer.vmo.handle, buffer.size);
+    final result = dataVmo.read(buffer.size);
+
+    if (result.status != 0) {
+      throw Exception(
+          'VMO read faile with status [${result.status}] when trying to read entity data with type [$type]');
+    }
+
+    dataVmo.close();
+    return result.bytesAsUint8List();
+  }
+}
+
+class _EntityWatcher implements fidl_modular.EntityWatcher {
+  final fidl_modular.EntityWatcherBinding binding =
+      fidl_modular.EntityWatcherBinding();
+  final void Function(fuchsia_mem.Buffer) onUpdatedFunc;
+
+  _EntityWatcher(this.onUpdatedFunc) : assert(onUpdatedFunc != null);
+
+  InterfaceHandle<fidl_modular.EntityWatcher> getInterfaceHandle() {
+    return binding.wrap(this);
+  }
+
+  @override
+  Future<void> onUpdated(fuchsia_mem.Buffer data) async => onUpdatedFunc(data);
+}
diff --git a/public/dart/fuchsia_modular/lib/src/entity/internal/_link_entity.dart b/public/dart/fuchsia_modular/lib/src/entity/internal/_link_entity.dart
deleted file mode 100644
index dc59146..0000000
--- a/public/dart/fuchsia_modular/lib/src/entity/internal/_link_entity.dart
+++ /dev/null
@@ -1,180 +0,0 @@
-// Copyright 2018 The Fuchsia Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-import 'dart:async';
-import 'dart:convert';
-import 'dart:typed_data';
-
-import 'package:fidl/fidl.dart';
-import 'package:fidl_fuchsia_mem/fidl_async.dart' as fuchsia_mem;
-import 'package:fidl_fuchsia_modular/fidl_async.dart' as fidl;
-import 'package:meta/meta.dart';
-import 'package:zircon/zircon.dart';
-
-import '../../internal/_component_context.dart';
-import '../../module/internal/_module_context.dart';
-import '../entity.dart';
-import '../entity_codec.dart';
-
-/// An [Entity] implementation which supports
-/// data that lives inside links.
-///
-/// Note: This is a temporary solution that will go away when links are removed.
-class LinkEntity<T> implements Entity<T> {
-  /// The name of the link to pull data from
-  final String linkName;
-
-  /// The codec used to decode/encode data
-  final EntityCodec codec;
-
-  fidl.Link _link;
-
-  /// Constructor
-  LinkEntity({
-    @required this.linkName,
-    @required this.codec,
-  })  : assert(linkName != null),
-        assert(codec != null);
-
-  @override
-  Future<T> getData() async => _getSnapshot();
-
-  @override
-  Stream<T> watch() {
-    final link = _getLink();
-    final controller = StreamController<fuchsia_mem.Buffer>();
-    final watcher = _LinkWatcher(onNotify: (buffer) async {
-      controller.add(buffer);
-    });
-
-    link.watch(watcher.getInterfaceHandle());
-
-    // if the user closed the stream we close the binding to the watcher
-    controller.onCancel = watcher.binding.close;
-
-    // if the connection closes then we close the stream
-    watcher.binding.whenClosed.then((_) {
-      if (!controller.isClosed) {
-        controller.close();
-      }
-    });
-
-    // Use _getSnapshot here instead of using the buffer directly because
-    // most uses of links are using the entityRef value on the link instead
-    // of raw json.
-    return controller.stream.asyncMap((_) => _getSnapshot());
-  }
-
-  @override
-  Future<void> write(T value) async {
-    final link = _getLink();
-
-    // Convert the object to raw bytes
-    final bytes = codec.encode(value);
-
-    // need to base64 encode so we can encode it as json.
-    final b64ByteString = base64.encode(bytes);
-    final jsonString = json.encode(b64ByteString);
-
-    // convert the json encoded string into bytes so we can put it in a vmo
-    final jsonData = Uint8List.fromList(utf8.encode(jsonString));
-    final vmo = SizedVmo.fromUint8List(jsonData);
-    final buffer = fuchsia_mem.Buffer(vmo: vmo, size: jsonData.length);
-
-    await link.set(null, buffer);
-  }
-
-  Future<T> _getEntityData<T>(String entityReference) async {
-    final resolver = fidl.EntityResolverProxy();
-    await getComponentContext().getEntityResolver(resolver.ctrl.request());
-
-    final entity = fidl.EntityProxy();
-    await resolver.resolveEntity(entityReference, entity.ctrl.request());
-
-    final types = await entity.getTypes();
-    if (!types.contains(codec.type)) {
-      throw EntityTypeException(codec.type);
-    }
-
-    final buffer = await entity.getData(codec.type);
-    final dataVmo = SizedVmo(buffer.vmo.handle, buffer.size);
-    final result = dataVmo.read(buffer.size);
-
-    if (result.status != 0) {
-      throw new Exception('Failed to read VMO');
-    }
-
-    dataVmo.close();
-
-    return codec.decode(result.bytesAsUint8List());
-  }
-
-  T _getJsonData<T>(fuchsia_mem.Buffer jsonBuffer) {
-    final vmo = SizedVmo(jsonBuffer.vmo.handle, jsonBuffer.size);
-    final result = vmo.read(jsonBuffer.size);
-    if (result.status != 0) {
-      throw Exception('Failed to read VMO');
-    }
-    vmo.close();
-
-    final resultBytes = result.bytesAsUint8List();
-    Uint8List bytesToDecode;
-
-    try {
-      // Try to decode the values in the format that this entity encoded them
-      final utf8String = utf8.decode(resultBytes);
-      final jsonDecoded = json.decode(utf8String);
-      bytesToDecode = base64.decode(jsonDecoded);
-    } on Exception catch (_) {
-      bytesToDecode = resultBytes;
-    }
-
-    return codec.decode(bytesToDecode);
-  }
-
-  fidl.Link _getLink() {
-    if (_link != null) {
-      return _link;
-    }
-
-    // Store the Link instead of the LinkProxy to avoid closing.
-    fidl.LinkProxy linkProxy = fidl.LinkProxy();
-    _link = linkProxy;
-    getModuleContext().getLink(linkName, linkProxy.ctrl.request());
-
-    return _link;
-  }
-
-  Future<T> _getSnapshot() async {
-    final link = _getLink();
-
-    final buffer = await link.get(null);
-    if (buffer == null) {
-      final entityReference = await link.getEntity();
-      if (entityReference == null) {
-        return null;
-      }
-      return _getEntityData(entityReference);
-    }
-    return _getJsonData(buffer);
-  }
-}
-
-class _LinkWatcher extends fidl.LinkWatcher {
-  final fidl.LinkWatcherBinding binding = fidl.LinkWatcherBinding();
-  final Future<void> Function(fuchsia_mem.Buffer) onNotify;
-
-  _LinkWatcher({@required this.onNotify}) : assert(onNotify != null);
-
-  InterfaceHandle<fidl.LinkWatcher> getInterfaceHandle() {
-    if (binding.isBound) {
-      throw Exception(
-          'Attempting to call _LinkWatcher.getInterfaceHandle on already bound binding');
-    }
-    return binding.wrap(this);
-  }
-
-  @override
-  Future<void> notify(fuchsia_mem.Buffer data) => onNotify(data);
-}
diff --git a/public/dart/fuchsia_modular/lib/src/module/intent.dart b/public/dart/fuchsia_modular/lib/src/module/intent.dart
index 38e3531..1c44e8a 100644
--- a/public/dart/fuchsia_modular/lib/src/module/intent.dart
+++ b/public/dart/fuchsia_modular/lib/src/module/intent.dart
@@ -6,8 +6,6 @@
 import 'package:meta/meta.dart';
 
 import '../entity/entity.dart';
-import '../entity/entity_codec.dart';
-import '../entity/internal/_link_entity.dart';
 import 'module_state_exception.dart';
 
 /// An [Intent] is a fundamental building block of module development.
@@ -66,16 +64,6 @@
         );
 
   /// Appends a [fidl.IntentParameter] to the intent's parameters containing
-  /// the [entity] as its data value and the [name] as its name value.
-  void addParameterFromEntity(String name, Entity entity) {
-    if (entity is LinkEntity) {
-      _addParameterFromLinkEntity(name, entity);
-    } else {
-      throw UnimplementedError('Only link entities are supported at this time');
-    }
-  }
-
-  /// Appends a [fidl.IntentParameter] to the intent's parameters containing
   /// the [reference] to an entity as its data value and the [name] as its
   /// name value.
   void addParameterFromEntityReference(String name, String reference) =>
@@ -85,35 +73,27 @@
   /// Returns the entity with the given [name]. An entity's name maps to
   /// the name of a  parameter in your component's manifset.
   ///
-  /// The codec is used to translate the bytes in the entity to the dart
-  /// object type.
-  ///
-  /// The [codec] will need to match the EntityDataType. The name will need
-  /// to match the semantic type and the encoding will need to match the
-  /// encoding such that it can actually encode and decode values
+  /// The type is used to ensure the entity provider provides us with the
+  /// type of entity we are expecting. If the type does not match the entity
+  /// will fail to resolve.
   @experimental
   Entity getEntity({
     @required String name,
-    @required EntityCodec codec,
-  }) =>
-      _entityFromIntentParameter(
-        parameter: _getParameter(name),
-        codec: codec,
-      );
+    @required String type,
+  }) {
+    ArgumentError.checkNotNull(name, 'name');
+    ArgumentError.checkNotNull(type, 'type');
+    return _entityFromIntentParameter(_getParameter(name), type);
+  }
 
   void _addParameter(String name, fidl.IntentParameterData parameterData) =>
       parameters.add(fidl.IntentParameter(name: name, data: parameterData));
 
-  void _addParameterFromLinkEntity(String name, LinkEntity entity) =>
-      _addParameter(
-          name, fidl.IntentParameterData.withLinkName(entity.linkName));
-
-  Entity _entityFromIntentParameter({
-    fidl.IntentParameter parameter,
-    EntityCodec codec,
-  }) {
-    if (parameter.data.tag == fidl.IntentParameterDataTag.linkName) {
-      return LinkEntity(linkName: parameter.data.linkName, codec: codec);
+  Entity _entityFromIntentParameter(
+      fidl.IntentParameter parameter, String type) {
+    if (parameter.data.tag == fidl.IntentParameterDataTag.entityReference) {
+      return Entity(
+          type: type, entityReference: parameter.data.entityReference);
     }
     throw UnimplementedError();
   }
diff --git a/public/dart/fuchsia_modular/meta/fuchsia_modular_package_integration_tests.cmx b/public/dart/fuchsia_modular/meta/fuchsia_modular_package_integration_tests.cmx
index ea24f9b..71676e7 100644
--- a/public/dart/fuchsia_modular/meta/fuchsia_modular_package_integration_tests.cmx
+++ b/public/dart/fuchsia_modular/meta/fuchsia_modular_package_integration_tests.cmx
@@ -8,7 +8,9 @@
             "fuchsia.fonts.Provider",
             "fuchsia.logger.LogSink",
             "fuchsia.modular.Clipboard",
+            "fuchsia.modular.ComponentContext",
             "fuchsia.modular.ContextWriter",
+            "fuchsia.modular.Lifecycle",
             "fuchsia.modular.ModuleContext",
             "fuchsia.netstack.Netstack",
             "fuchsia.sys.Environment",
diff --git a/public/dart/fuchsia_modular/test/entity/internal/entity_impl_test.dart b/public/dart/fuchsia_modular/test/entity/internal/entity_impl_test.dart
new file mode 100644
index 0000000..99bec26
--- /dev/null
+++ b/public/dart/fuchsia_modular/test/entity/internal/entity_impl_test.dart
@@ -0,0 +1,152 @@
+// Copyright 2018 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+// ignore_for_file: implementation_imports
+
+import 'dart:typed_data';
+
+import 'package:fidl/fidl.dart';
+import 'package:fidl_fuchsia_mem/fidl_async.dart' as fuchsia_mem;
+import 'package:fidl_fuchsia_modular/fidl_async.dart' as fuchsia_modular;
+import 'package:fidl_fuchsia_modular/fidl_test.dart' as fuchsia_modular_test;
+import 'package:fuchsia_modular/src/entity/entity_exceptions.dart';
+import 'package:fuchsia_modular/src/entity/internal/_entity_impl.dart';
+import 'package:test/test.dart';
+import 'package:zircon/zircon.dart';
+
+void main() {
+  MockProxy proxy;
+  EntityImpl entity;
+
+  group('enity interactions', () {
+    setUp(() {
+      proxy = MockProxy();
+      entity = EntityImpl(proxyFactory: () => proxy, type: 'foo');
+    });
+
+    tearDown(() {
+      proxy = null;
+      entity = null;
+    });
+    test('getData returns correct data', () async {
+      final data = Uint8List.fromList([1, 2, 3]);
+      await entity.write(data);
+      expect(await entity.getData(), data);
+    });
+
+    test('write returns success for valid write', () async {
+      final data = Uint8List.fromList([1, 2, 3]);
+      await entity.write(data);
+      // test will fail if write fails
+    });
+
+    test('write fails for error in write', () async {
+      proxy.writeStatus = fuchsia_modular.EntityWriteStatus.error;
+      final data = Uint8List.fromList([1, 2, 3]);
+      expect(entity.write(data),
+          throwsA(const TypeMatcher<EntityWriteException>()));
+    });
+
+    test('write fails for readonly', () async {
+      proxy.writeStatus = fuchsia_modular.EntityWriteStatus.readOnly;
+      final data = Uint8List.fromList([1, 2, 3]);
+      expect(entity.write(data),
+          throwsA(const TypeMatcher<EntityWriteException>()));
+    });
+
+    test('getEntityReference returns the correct ref', () async {
+      proxy.entityReference = 'my-entity';
+      expect(await entity.getEntityReference(), 'my-entity');
+    });
+
+    test('watch closes stream when binding is closed', () {
+      proxy.forceCloseWatchers = true;
+      final stream = entity.watch();
+      expect(stream, emitsInOrder([emitsDone]));
+    });
+
+    test('watch udpates stream', () async {
+      Uint8List _bufferWithValue(int v) {
+        final data = ByteData(1)..setInt8(0, v);
+        return data.buffer.asUint8List();
+      }
+
+      int _getValue(Uint8List buffer) {
+        return ByteData.view(buffer.buffer).getInt8(0);
+      }
+
+      final stream = entity.watch();
+
+      expect(stream.map(_getValue), emitsInOrder([1, 2, 3]));
+
+      await entity.write(_bufferWithValue(1));
+      await entity.write(_bufferWithValue(2));
+      await entity.write(_bufferWithValue(3));
+    });
+  });
+}
+
+class MockProxy extends fuchsia_modular_test.Entity$TestBase {
+  Uint8List _data;
+  String entityReference;
+  List<fuchsia_modular.EntityWatcherProxy> watchers = [];
+
+  /// if set to true the watchers will be closed after they are added.
+  bool forceCloseWatchers = false;
+
+  /// Can be set to control how write results are handled.
+  fuchsia_modular.EntityWriteStatus writeStatus =
+      fuchsia_modular.EntityWriteStatus.ok;
+
+  @override
+  Future<fuchsia_mem.Buffer> getData(String _) async {
+    final vmo = SizedVmo.fromUint8List(_data);
+    final buffer = fuchsia_mem.Buffer(vmo: vmo, size: _data.length);
+    return buffer;
+  }
+
+  @override
+  Future<String> getReference() async => entityReference;
+
+  @override
+  Future<void> watch(String type,
+      InterfaceHandle<fuchsia_modular.EntityWatcher> watcher) async {
+    final proxy = fuchsia_modular.EntityWatcherProxy();
+    proxy.ctrl.bind(watcher);
+    watchers.add(proxy);
+
+    if (forceCloseWatchers) {
+      //ignore: unawaited_futures
+      Future(proxy.ctrl.close);
+    }
+  }
+
+  @override
+  Future<fuchsia_modular.EntityWriteStatus> writeData(
+      String type, fuchsia_mem.Buffer buffer) async {
+    if (writeStatus != fuchsia_modular.EntityWriteStatus.ok) {
+      return writeStatus;
+    }
+
+    final dataVmo = SizedVmo(buffer.vmo.handle, buffer.size);
+    final result = dataVmo.read(buffer.size);
+    dataVmo.close();
+    final data = result.bytesAsUint8List();
+    _setData(data, notify: true);
+    return fuchsia_modular.EntityWriteStatus.ok;
+  }
+
+  void _setData(Uint8List data, {bool notify = false}) {
+    _data = data;
+    if (notify) {
+      final vmo = SizedVmo.fromUint8List(data);
+      final buffer = fuchsia_mem.Buffer(vmo: vmo, size: data.length);
+      for (final watcher in watchers) {
+        if (watcher.ctrl.isBound) {
+          watcher.onUpdated(buffer);
+        }
+      }
+    }
+  }
+}
diff --git a/public/dart/fuchsia_modular/test/module/intent_test.dart b/public/dart/fuchsia_modular/test/module/intent_test.dart
index 9ac71e7..1450b8c 100644
--- a/public/dart/fuchsia_modular/test/module/intent_test.dart
+++ b/public/dart/fuchsia_modular/test/module/intent_test.dart
@@ -2,12 +2,8 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-import 'dart:convert';
-import 'dart:typed_data';
-
-import 'package:fidl_fuchsia_modular/fidl_async.dart' as fidl;
 import 'package:fuchsia_modular/src/module/intent.dart'; // ignore: implementation_imports
-import 'package:fuchsia_modular/src/entity/entity_codec.dart'; // ignore: implementation_imports
+
 import 'package:test/test.dart';
 
 import '../matchers.dart';
@@ -41,28 +37,8 @@
 
     test('getEntity throws for missing name', () {
       expect(() {
-        intent.getEntity(name: 'not-a-name', codec: stub);
+        intent.getEntity(name: 'not-a-name', type: 'foo');
       }, throwsModuleStateException);
     });
-
-    test('getEntity returns valid entity for link entity', () {
-      //NOTE: this test can be deleted when _link_entity goes away.
-      intent.parameters.add(fidl.IntentParameter(
-          name: 'name',
-          data: fidl.IntentParameterData.withLinkName('foo-link')));
-      final entity = intent.getEntity(name: 'name', codec: stub);
-      expect(entity, isNotNull);
-    });
   });
 }
-
-const StubEntityCodec stub = StubEntityCodec();
-
-class StubEntityCodec extends EntityCodec {
-  const StubEntityCodec() : super(type: '', encoding: '');
-  @override
-  Converter<Uint8List, dynamic> get decoder => null;
-
-  @override
-  Converter<dynamic, Uint8List> get encoder => null;
-}