diff --git a/src/python/grpcio/grpc/framework/core/_context.py b/src/python/grpcio/grpc/framework/core/_context.py
index 24a12b6..76b3534 100644
--- a/src/python/grpcio/grpc/framework/core/_context.py
+++ b/src/python/grpcio/grpc/framework/core/_context.py
@@ -60,7 +60,7 @@
     with self._lock:
       if self._termination_manager.outcome is None:
         self._termination_manager.abort(outcome)
-        self._transmission_manager.abort(outcome)
+        self._transmission_manager.abort(outcome, None, None)
         self._expiration_manager.terminate()
 
   def outcome(self):
diff --git a/src/python/grpcio/grpc/framework/core/_emission.py b/src/python/grpcio/grpc/framework/core/_emission.py
index 7c702ab..2d7b2e2 100644
--- a/src/python/grpcio/grpc/framework/core/_emission.py
+++ b/src/python/grpcio/grpc/framework/core/_emission.py
@@ -82,7 +82,8 @@
             completion_present and self._completion_seen or
             allowance_present and allowance <= 0):
           self._termination_manager.abort(base.Outcome.LOCAL_FAILURE)
-          self._transmission_manager.abort(base.Outcome.LOCAL_FAILURE)
+          self._transmission_manager.abort(
+              base.Outcome.LOCAL_FAILURE, None, None)
           self._expiration_manager.terminate()
         else:
           self._initial_metadata_seen |= initial_metadata_present
diff --git a/src/python/grpcio/grpc/framework/core/_expiration.py b/src/python/grpcio/grpc/framework/core/_expiration.py
index d94bdf2..d8690b3 100644
--- a/src/python/grpcio/grpc/framework/core/_expiration.py
+++ b/src/python/grpcio/grpc/framework/core/_expiration.py
@@ -73,7 +73,7 @@
         if self._future is not None and index == self._index:
           self._future = None
           self._termination_manager.expire()
-          self._transmission_manager.abort(base.Outcome.EXPIRED)
+          self._transmission_manager.abort(base.Outcome.EXPIRED, None, None)
     return expire
 
   def start(self):
diff --git a/src/python/grpcio/grpc/framework/core/_ingestion.py b/src/python/grpcio/grpc/framework/core/_ingestion.py
index 59f7f8a..7b8127f 100644
--- a/src/python/grpcio/grpc/framework/core/_ingestion.py
+++ b/src/python/grpcio/grpc/framework/core/_ingestion.py
@@ -31,6 +31,7 @@
 
 import abc
 import collections
+import enum
 
 from grpc.framework.core import _constants
 from grpc.framework.core import _interfaces
@@ -42,21 +43,31 @@
 _INGESTION_EXCEPTION_LOG_MESSAGE = 'Exception during ingestion!'
 
 
-class _SubscriptionCreation(collections.namedtuple(
-    '_SubscriptionCreation', ('subscription', 'remote_error', 'abandoned'))):
+class _SubscriptionCreation(
+    collections.namedtuple(
+        '_SubscriptionCreation',
+        ('kind', 'subscription', 'code', 'message',))):
   """A sum type for the outcome of ingestion initialization.
 
-  Either subscription will be non-None, remote_error will be True, or abandoned
-  will be True.
-
   Attributes:
-    subscription: A base.Subscription describing the customer's interest in
-      operation values from the other side.
-    remote_error: A boolean indicating that the subscription could not be
-      created due to an error on the remote side of the operation.
-    abandoned: A boolean indicating that subscription creation was abandoned.
+    kind: A Kind value coarsely indicating how subscription creation completed.
+    subscription: The created subscription. Only present if kind is
+      Kind.SUBSCRIPTION.
+    code: A code value to be sent to the other side of the operation along with
+      an indication that the operation is being aborted due to an error on the
+      remote side of the operation. Only present if kind is Kind.REMOTE_ERROR.
+    message: A message value to be sent to the other side of the operation
+      along with an indication that the operation is being aborted due to an
+      error on the remote side of the operation. Only present if kind is
+      Kind.REMOTE_ERROR.
   """
 
+  @enum.unique
+  class Kind(enum.Enum):
+    SUBSCRIPTION = 'subscription'
+    REMOTE_ERROR = 'remote error'
+    ABANDONED = 'abandoned'
+
 
 class _SubscriptionCreator(object):
   """Common specification of subscription-creating behavior."""
@@ -101,12 +112,15 @@
     try:
       subscription = self._servicer.service(
           group, method, self._operation_context, self._output_operator)
-    except base.NoSuchMethodError:
-      return _SubscriptionCreation(None, True, False)
+    except base.NoSuchMethodError as e:
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.REMOTE_ERROR, None, e.code, e.message)
     except abandonment.Abandoned:
-      return _SubscriptionCreation(None, False, True)
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.ABANDONED, None, None, None)
     else:
-      return _SubscriptionCreation(subscription, False, False)
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.SUBSCRIPTION, subscription, None, None)
 
 
 def _wrap(behavior):
@@ -176,10 +190,10 @@
     self._pending_payloads = None
     self._pending_completion = None
 
-  def _abort_and_notify(self, outcome):
+  def _abort_and_notify(self, outcome, code, message):
     self._abort_internal_only()
     self._termination_manager.abort(outcome)
-    self._transmission_manager.abort(outcome)
+    self._transmission_manager.abort(outcome, code, message)
     self._expiration_manager.terminate()
 
   def _operator_next(self):
@@ -236,12 +250,12 @@
         else:
           with self._lock:
             if self._termination_manager.outcome is None:
-              self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+              self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
             return
       else:
         with self._lock:
           if self._termination_manager.outcome is None:
-            self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+            self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
           return
 
   def _operator_post_create(self, subscription):
@@ -260,20 +274,22 @@
 
   def _create(self, subscription_creator, group, name):
     outcome = callable_util.call_logging_exceptions(
-        subscription_creator.create, _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE,
-        group, name)
+        subscription_creator.create,
+        _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE, group, name)
     if outcome.return_value is None:
       with self._lock:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
-    elif outcome.return_value.abandoned:
+          self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
+    elif outcome.return_value.kind is _SubscriptionCreation.Kind.ABANDONED:
       with self._lock:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
-    elif outcome.return_value.remote_error:
+          self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
+    elif outcome.return_value.kind is _SubscriptionCreation.Kind.REMOTE_ERROR:
+      code = outcome.return_value.code
+      message = outcome.return_value.message
       with self._lock:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.REMOTE_FAILURE)
+          self._abort_and_notify(base.Outcome.REMOTE_FAILURE, code, message)
     elif outcome.return_value.subscription.kind is base.Subscription.Kind.FULL:
       self._operator_post_create(outcome.return_value.subscription)
     else:
diff --git a/src/python/grpcio/grpc/framework/core/_interfaces.py b/src/python/grpcio/grpc/framework/core/_interfaces.py
index a626b9f..deb5f34 100644
--- a/src/python/grpcio/grpc/framework/core/_interfaces.py
+++ b/src/python/grpcio/grpc/framework/core/_interfaces.py
@@ -155,13 +155,19 @@
     raise NotImplementedError()
 
   @abc.abstractmethod
-  def abort(self, outcome):
+  def abort(self, outcome, code, message):
     """Indicates that the operation has aborted.
 
     Args:
       outcome: An interfaces.Outcome for the operation. If None, indicates that
         the operation abortion should not be communicated to the other side of
         the operation.
+      code: A code value to communicate to the other side of the operation
+        along with indication of operation abortion. May be None, and has no
+        effect if outcome is None.
+      message: A message value to communicate to the other side of the
+        operation along with indication of operation abortion. May be None, and
+        has no effect if outcome is None.
     """
     raise NotImplementedError()
 
diff --git a/src/python/grpcio/grpc/framework/core/_operation.py b/src/python/grpcio/grpc/framework/core/_operation.py
index d20e40a..cc873c0 100644
--- a/src/python/grpcio/grpc/framework/core/_operation.py
+++ b/src/python/grpcio/grpc/framework/core/_operation.py
@@ -79,7 +79,7 @@
     with self._lock:
       if self._termination_manager.outcome is None:
         self._termination_manager.abort(outcome)
-        self._transmission_manager.abort(outcome)
+        self._transmission_manager.abort(outcome, None, None)
         self._expiration_manager.terminate()
 
 
diff --git a/src/python/grpcio/grpc/framework/core/_reception.py b/src/python/grpcio/grpc/framework/core/_reception.py
index 0858f64..1cebe38 100644
--- a/src/python/grpcio/grpc/framework/core/_reception.py
+++ b/src/python/grpcio/grpc/framework/core/_reception.py
@@ -73,7 +73,7 @@
     self._aborted = True
     if self._termination_manager.outcome is None:
       self._termination_manager.abort(outcome)
-      self._transmission_manager.abort(None)
+      self._transmission_manager.abort(None, None, None)
       self._expiration_manager.terminate()
 
   def _sequence_failure(self, ticket):
diff --git a/src/python/grpcio/grpc/framework/core/_transmission.py b/src/python/grpcio/grpc/framework/core/_transmission.py
index 03644f4..efef87d 100644
--- a/src/python/grpcio/grpc/framework/core/_transmission.py
+++ b/src/python/grpcio/grpc/framework/core/_transmission.py
@@ -104,9 +104,13 @@
           return None
         else:
           self._abortion_outcome = None
+          if self._completion is None:
+            code, message = None, None
+          else:
+            code, message = self._completion.code, self._completion.message
           return links.Ticket(
               self._operation_id, self._lowest_unused_sequence_number, None,
-              None, None, None, None, None, None, None, None, None,
+              None, None, None, None, None, None, None, code, message,
               termination, None)
 
     action = False
@@ -277,7 +281,7 @@
     self._remote_complete = True
     self._local_allowance = 0
 
-  def abort(self, outcome):
+  def abort(self, outcome, code, message):
     """See _interfaces.TransmissionManager.abort for specification."""
     if self._transmitting:
       self._aborted, self._abortion_outcome = True, outcome
@@ -287,8 +291,12 @@
         termination = _constants.ABORTION_OUTCOME_TO_TICKET_TERMINATION[
             outcome]
         if termination is not None:
+          if self._completion is None:
+            code, message = None, None
+          else:
+            code, message = self._completion.code, self._completion.message
           ticket = links.Ticket(
               self._operation_id, self._lowest_unused_sequence_number, None,
-              None, None, None, None, None, None, None, None, None,
+              None, None, None, None, None, None, None, code, message,
               termination, None)
           self._transmit(ticket)
diff --git a/src/python/grpcio/grpc/framework/interfaces/base/base.py b/src/python/grpcio/grpc/framework/interfaces/base/base.py
index 76e0a5b..bc52efb 100644
--- a/src/python/grpcio/grpc/framework/interfaces/base/base.py
+++ b/src/python/grpcio/grpc/framework/interfaces/base/base.py
@@ -47,7 +47,26 @@
 
 
 class NoSuchMethodError(Exception):
-  """Indicates that an unrecognized operation has been called."""
+  """Indicates that an unrecognized operation has been called.
+
+  Attributes:
+    code: A code value to communicate to the other side of the operation along
+      with indication of operation termination. May be None.
+    details: A details value to communicate to the other side of the operation
+      along with indication of operation termination. May be None.
+  """
+
+  def __init__(self, code, details):
+    """Constructor.
+
+    Args:
+      code: A code value to communicate to the other side of the operation
+        along with indication of operation termination. May be None.
+      details: A details value to communicate to the other side of the
+        operation along with indication of operation termination. May be None.
+    """
+    self.code = code
+    self.details = details
 
 
 @enum.unique
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
index 5c8b176..87332cf 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
@@ -134,7 +134,7 @@
       if group != self._group or method != self._method:
         controller.fail(
             '%s != %s or %s != %s' % (group, self._group, method, self._method))
-        raise base.NoSuchMethodError()
+        raise base.NoSuchMethodError(None, None)
       else:
         operator = _Operator(
             controller, controller.on_service_advance, self._pool,