docs: def run_callback(self):
      """Runs the callback for the pipeline specified in the request.
  
      Raises:
        _CallbackTaskError if something was wrong with the request parameters.
      """
      pipeline_id = self.request.get('pipeline_id')
      if not pipeline_id:
        raise _CallbackTaskError('"pipeline_id" parameter missing.')
  
      pipeline_key = db.Key.from_path(_PipelineRecord.kind(), pipeline_id)
      pipeline_record = db.get(pipeline_key)
      if pipeline_record is None:
        raise _CallbackTaskError(
            'Pipeline ID "%s" for callback does not exist.' % pipeline_id)
  
      params = pipeline_record.params
      real_class_path = params['class_path']
      try:
        pipeline_func_class = mr_util.for_name(real_class_path)
      except ImportError, e:
        raise _CallbackTaskError(
            'Cannot load class named "%s" for pipeline ID "%s".'
            % (real_class_path, pipeline_id))
  
      if 'HTTP_X_APPENGINE_TASKNAME' not in self.request.environ:
        if pipeline_func_class.public_callbacks:
          pass
        elif pipeline_func_class.admin_callbacks:
          if not users.is_current_user_admin():
            raise _CallbackTaskError(
                'Unauthorized callback for admin-only pipeline ID "%s"'
                % pipeline_id)
        else:
          raise _CallbackTaskError(
              'External callback for internal-only pipeline ID "%s"'
              % pipeline_id)
  
      kwargs = {}
      for key in self.request.arguments():
        if key != 'pipeline_id':
          kwargs[str(key)] = self.request.get(key)
  
      def perform_callback():
        stage = pipeline_func_class.from_id(pipeline_id)
        if stage is None:
          raise _CallbackTaskError(
              'Pipeline ID "%s" deleted during callback' % pipeline_id)
        return stage._callback_internal(kwargs)
  
      # callback_xg_transaction is a 3-valued setting (None=no trans,
      # False=1-eg-trans, True=xg-trans)
      if pipeline_func_class._callback_xg_transaction is not None:
        transaction_options = db.create_transaction_options(
            xg=pipeline_func_class._callback_xg_transaction)
        callback_result = db.run_in_transaction_options(transaction_options,
                                                        perform_callback)
      else:
        callback_result = perform_callback()
  
      if callback_result is not None:
        status_code, content_type, content = callback_result
        self.response.set_status(status_code)
        self.response.headers['Content-Type'] = content_type
        self.response.out.write(content)