diff --git a/doc/source/actors.rst b/doc/source/actors.rst
index 6961f4d95..bb214ffac 100644
--- a/doc/source/actors.rst
+++ b/doc/source/actors.rst
@@ -27,7 +27,7 @@ An actor can be defined as follows.
 
   import gym
 
-  @ray.actor
+  @ray.remote
   class GymEnvironment(object):
     def __init__(self, name):
       self.env = gym.make(name)
@@ -63,19 +63,20 @@ We can use the actor by calling one of its methods.
 
 .. code-block:: python
 
-  a1.step(0)
-  a2.step(0)
+  a1.step.remote(0)
+  a2.step.remote(0)
 
-When ``a1.step(0)`` is called, a task is created and scheduled on the first
-actor. This scheduling procedure bypasses the global scheduler, and is assigned
-directly to the local scheduler responsible for the actor by the driver's local
-scheduler. Since the method call is a task, ``a1.step(0)`` returns an object ID.
-We can call `ray.get` on the object ID to retrieve the actual value.
+When ``a1.step.remote(0)`` is called, a task is created and scheduled on the
+first actor. This scheduling procedure bypasses the global scheduler, and is
+assigned directly to the local scheduler responsible for the actor by the
+driver's local scheduler. Since the method call is a task, ``a1.step(0)``
+returns an object ID. We can call `ray.get` on the object ID to retrieve the
+actual value.
 
-The call to ``a2.step(0)`` generates a task which is scheduled on the second
-actor. Since these two tasks run on different actors, they can be executed in
-parallel (note that only actor methods will be scheduled on actor workers, not
-regular remote functions).
+The call to ``a2.step.remote(0)`` generates a task which is scheduled on the
+second actor. Since these two tasks run on different actors, they can be
+executed in parallel (note that only actor methods will be scheduled on actor
+workers, not regular remote functions).
 
 On the other hand, methods called on the same actor are executed serially and
 share in the order that they are called and share state with one another. We
@@ -83,7 +84,7 @@ illustrate this with a simple example.
 
 .. code-block:: python
 
-  @ray.actor
+  @ray.remote
   class Counter(object):
     def __init__(self):
       self.value = 0
@@ -96,12 +97,12 @@ illustrate this with a simple example.
 
   # Increment each counter once and get the results. These tasks all happen in
   # parallel.
-  results = ray.get([c.increment() for c in counters])
+  results = ray.get([c.increment.remote() for c in counters])
   print(results)  # prints [1, 1, 1, 1, 1, 1, 1, 1, 1, 1]
 
   # Increment the first counter five times. These tasks are executed serially
   # and share state.
-  results = ray.get([counters[0].increment() for _ in range(5)])
+  results = ray.get([counters[0].increment.remote() for _ in range(5)])
   print(results)  # prints [2, 3, 4, 5, 6]
 
 Using GPUs on actors
@@ -136,8 +137,8 @@ We can then define an actor for this network as follows.
   import os
 
   # Define an actor that runs on GPUs. If there are no GPUs, then simply use
-  # ray.actor without any arguments and no parentheses.
-  @ray.actor(num_gpus=1)
+  # ray.remote without any arguments and no parentheses.
+  @ray.remote(num_gpus=1)
   class NeuralNetOnGPU(object):
     def __init__(self):
       # Set an environment variable to tell TensorFlow which GPUs to use. Note
@@ -154,15 +155,15 @@ We can then define an actor for this network as follows.
           self.sess.run(init)
 
 To indicate that an actor requires one GPU, we pass in ``num_gpus=1`` to
-``ray.actor``. Note that in order for this to work, Ray must have been started
+``ray.remote``. Note that in order for this to work, Ray must have been started
 with some GPUs, e.g., via ``ray.init(num_gpus=2)``. Otherwise, when you try to
-instantiate the GPU version with ``NeuralNetOnGPU()``, an exception will be
-thrown saying that there aren't enough GPUs in the system.
+instantiate the GPU version with ``NeuralNetOnGPU.remote()``, an exception will
+be thrown saying that there aren't enough GPUs in the system.
 
 When the actor is created, it will have access to a list of the IDs of the GPUs
 that it is allowed to use via ``ray.get_gpu_ids()``. This is a list of integers,
 like ``[]``, or ``[1]``, or ``[2, 5, 6]``. Since we passed in
-``ray.actor(num_gpus=1)``, this list will have length one.
+``ray.remote(num_gpus=1)``, this list will have length one.
 
 We can put this all together as follows.
 
@@ -190,7 +191,7 @@ We can put this all together as follows.
 
     return x, y_, train_step, accuracy
 
-  @ray.actor(num_gpus=1)
+  @ray.remote(num_gpus=1)
   class NeuralNetOnGPU(object):
     def __init__(self, mnist_data):
       self.mnist = mnist_data
@@ -223,9 +224,9 @@ We can put this all together as follows.
   ray.register_class(type(mnist.train))
 
   # Create the actor.
-  nn = NeuralNetOnGPU(mnist)
+  nn = NeuralNetOnGPU.remote(mnist)
 
   # Run a few steps of training and print the accuracy.
-  nn.train(100)
-  accuracy = ray.get(nn.get_accuracy())
+  nn.train.remote(100)
+  accuracy = ray.get(nn.get_accuracy.remote())
   print("Accuracy is {}.".format(accuracy))
diff --git a/doc/source/example-a3c.rst b/doc/source/example-a3c.rst
index de7b060ae..553889ff4 100644
--- a/doc/source/example-a3c.rst
+++ b/doc/source/example-a3c.rst
@@ -73,7 +73,7 @@ We use a Ray Actor to simulate the environment.
   import numpy as np
   import ray
 
-  @ray.actor
+  @ray.remote
   class Runner(object):
     """Actor object to start running simulation on workers.
         Gradient computation is also executed on this object."""
@@ -127,7 +127,7 @@ global model parameters. The main training script looks like the following.
 
     # Start gradient calculation tasks on each actor
     parameters = policy.get_weights()
-    gradient_list = [agent.compute_gradient(parameters) for agent in agents]
+    gradient_list = [agent.compute_gradient.remote(parameters) for agent in agents]
 
     while True: # Replace with your termination condition
       # wait for some gradient to be computed - unblock as soon as the earliest arrives
@@ -147,6 +147,12 @@ global model parameters. The main training script looks like the following.
 Benchmarks and Visualization
 ----------------------------
 
-For the :code:`PongDeterministic-v3` and an Amazon EC2 m4.16xlarge instance, we are able to train the agent with 16 workers in around 15 minutes. With 8 workers, we can train the agent in around 25 minutes.
+For the :code:`PongDeterministic-v3` and an Amazon EC2 m4.16xlarge instance, we
+are able to train the agent with 16 workers in around 15 minutes. With 8
+workers, we can train the agent in around 25 minutes.
 
-You can visualize performance by running :code:`tensorboard --logdir [directory]` in a separate screen, where :code:`[directory]` is defaulted to :code:`./results/`. If you are running multiple experiments, be sure to vary the directory to which Tensorflow saves its progress (found in :code:`driver.py`).
+You can visualize performance by running
+:code:`tensorboard --logdir [directory]` in a separate screen, where
+:code:`[directory]` is defaulted to :code:`./results/`. If you are running
+multiple experiments, be sure to vary the directory to which Tensorflow saves
+its progress (found in :code:`driver.py`).
diff --git a/doc/source/example-resnet.rst b/doc/source/example-resnet.rst
index 6be26d9c9..3b0313c0a 100644
--- a/doc/source/example-resnet.rst
+++ b/doc/source/example-resnet.rst
@@ -53,7 +53,7 @@ The core of the script is the actor definition.
 
 .. code-block:: python
 
-  @ray.actor(num_gpus=1)
+  @ray.remote(num_gpus=1)
   class ResNetTrainActor(object):
     def __init__(self, path, num_gpus):
       # Set the CUDA_VISIBLE_DEVICES environment variable in order to restrict
@@ -78,7 +78,7 @@ The main script first creates one actor for each GPU.
 
 .. code-block:: python
 
-  train_actors = [ResNetTrainActor(train_data, num_gpus) for _ in range(num_gpus)]
+  train_actors = [ResNetTrainActor.remote(train_data, num_gpus) for _ in range(num_gpus)]
 
 Then after initializing the actors with the same weights, the main loop performs
 updates on each model, averages the updates, and puts the new weights in the
@@ -87,7 +87,7 @@ object store.
 .. code-block:: python
 
   while True:
-    all_weights = ray.get([actor.compute_steps(weight_id) for actor in train_actors])
+    all_weights = ray.get([actor.compute_steps.remote(weight_id) for actor in train_actors])
     mean_weights = {k: sum([weights[k] for weights in all_weights]) / num_gpus for k in all_weights[0]}
     weight_id = ray.put(mean_weights)
 
diff --git a/doc/source/example-rl-pong.rst b/doc/source/example-rl-pong.rst
index 43bc4a8b0..899191048 100644
--- a/doc/source/example-rl-pong.rst
+++ b/doc/source/example-rl-pong.rst
@@ -58,7 +58,7 @@ the actor.
 
 .. code-block:: python
 
-  @ray.actor
+  @ray.remote
   class PongEnv(object):
     def __init__(self):
       # Tell numpy to only use one core. If we don't do this, each actor may try
@@ -93,7 +93,7 @@ perform rollouts and compute gradients in parallel.
   actions = []
   # Launch tasks to compute gradients from multiple rollouts in parallel.
   for i in range(batch_size):
-    action_id = actors[i].compute_gradient(model_id)
+    action_id = actors[i].compute_gradient.remote(model_id)
     actions.append(action_id)
 
 
diff --git a/doc/source/using-ray-with-tensorflow.rst b/doc/source/using-ray-with-tensorflow.rst
index 0a2bdefcd..39caf7724 100644
--- a/doc/source/using-ray-with-tensorflow.rst
+++ b/doc/source/using-ray-with-tensorflow.rst
@@ -158,11 +158,11 @@ complex Python objects.
   x_test, y_test = ray.get(generate_fake_x_y_data.remote(BATCH_SIZE, seed=NUM_BATCHES))
 
   # Create actors to store the networks.
-  remote_network = ray.actor(Network)
-  actor_list = [remote_network(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)]
+  remote_network = ray.remote(Network)
+  actor_list = [remote_network.remote(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)]
 
   # Get initial weights of some actor.
-  weights = ray.get(actor_list[0].get_weights())
+  weights = ray.get(actor_list[0].get_weights.remote())
 
   # Do some steps of training.
   for iteration in range(NUM_ITERS):
@@ -173,7 +173,7 @@ complex Python objects.
     # more efficient.
     weights_id = ray.put(weights)
     # Call the remote function multiple times in parallel.
-    new_weights_ids = [actor.step(weights_id) for actor in actor_list]
+    new_weights_ids = [actor.step.remote(weights_id) for actor in actor_list]
     # Get all of the weights.
     new_weights_list = ray.get(new_weights_ids)
     # Add up all the different weights. Each element of new_weights_list is a dict
@@ -288,8 +288,8 @@ For reference, the full code is below:
   x_test, y_test = ray.get(generate_fake_x_y_data.remote(BATCH_SIZE, seed=NUM_BATCHES))
 
   # Create actors to store the networks.
-  remote_network = ray.actor(Network)
-  actor_list = [remote_network(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)]
+  remote_network = ray.remote(Network)
+  actor_list = [remote_network.remote(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)]
   local_network = Network(x_test, y_test)
 
   # Get initial weights of local network.
@@ -304,7 +304,7 @@ For reference, the full code is below:
     # more efficient.
     weights_id = ray.put(weights)
     # Call the remote function multiple times in parallel.
-    gradients_ids = [actor.step(weights_id) for actor in actor_list]
+    gradients_ids = [actor.step.remote(weights_id) for actor in actor_list]
     # Get all of the weights.
     gradients_list = ray.get(gradients_ids)
 
diff --git a/examples/a3c/driver.py b/examples/a3c/driver.py
index fe1c97d5d..e4fc0624a 100644
--- a/examples/a3c/driver.py
+++ b/examples/a3c/driver.py
@@ -15,7 +15,7 @@ from datetime import datetime, timedelta
 from misc import timestamp, time_string
 from envs import create_env
 
-@ray.actor
+@ray.remote
 class Runner(object):
     """Actor object to start running simulation on workers.
         Gradient computation is also executed from this object."""
@@ -58,9 +58,9 @@ class Runner(object):
 def train(num_workers, env_name="PongDeterministic-v3"):
     env = create_env(env_name)
     policy = LSTMPolicy(env.observation_space.shape, env.action_space.n, 0)
-    agents = [Runner(env_name, i) for i in range(num_workers)]
+    agents = [Runner.remote(env_name, i) for i in range(num_workers)]
     parameters = policy.get_weights()
-    gradient_list = [agent.compute_gradient(parameters) for agent in agents]
+    gradient_list = [agent.compute_gradient.remote(parameters) for agent in agents]
     steps = 0
     obs = 0
     while True:
@@ -70,7 +70,7 @@ def train(num_workers, env_name="PongDeterministic-v3"):
         parameters = policy.get_weights()
         steps += 1
         obs += info["size"]
-        gradient_list.extend([agents[info["id"]].compute_gradient(parameters)])
+        gradient_list.extend([agents[info["id"]].compute_gradient.remote(parameters)])
     return policy
 
 if __name__ == '__main__':
diff --git a/examples/lbfgs/driver.py b/examples/lbfgs/driver.py
index 68639ca37..6475208c6 100644
--- a/examples/lbfgs/driver.py
+++ b/examples/lbfgs/driver.py
@@ -60,7 +60,7 @@ class LinearModel(object):
     """Computes the gradients of the network."""
     return self.sess.run(self.cross_entropy_grads, feed_dict={self.x: xs, self.y_: ys})
 
-@ray.actor
+@ray.remote
 class NetActor(object):
   def __init__(self, xs, ys):
     os.environ["CUDA_VISIBLE_DEVICES"] = ""
@@ -88,13 +88,13 @@ class NetActor(object):
 # Compute the loss on the entire dataset.
 def full_loss(theta):
   theta_id = ray.put(theta)
-  loss_ids = [actor.loss(theta_id) for actor in actors]
+  loss_ids = [actor.loss.remote(theta_id) for actor in actors]
   return sum(ray.get(loss_ids))
 
 # Compute the gradient of the loss on the entire dataset.
 def full_grad(theta):
   theta_id = ray.put(theta)
-  grad_ids = [actor.grad(theta_id) for actor in actors]
+  grad_ids = [actor.grad.remote(theta_id) for actor in actors]
   return sum(ray.get(grad_ids)).astype("float64") # This conversion is necessary for use with fmin_l_bfgs_b.
 
 if __name__ == "__main__":
@@ -117,9 +117,9 @@ if __name__ == "__main__":
   batch_size = mnist.train.num_examples // num_batches
   batches = [mnist.train.next_batch(batch_size) for _ in range(num_batches)]
   print("Putting MNIST in the object store.")
-  actors = [NetActor(xs, ys) for (xs, ys) in batches]
+  actors = [NetActor.remote(xs, ys) for (xs, ys) in batches]
   # Initialize the weights for the network to the vector of all zeros.
-  dim = ray.get(actors[0].get_flat_size())
+  dim = ray.get(actors[0].get_flat_size.remote())
   theta_init = 1e-2 * np.random.normal(size=dim)
 
   # Use L-BFGS to minimize the loss function.
diff --git a/examples/policy_gradient/examples/example.py b/examples/policy_gradient/examples/example.py
index 49f823206..7b9a01ab7 100644
--- a/examples/policy_gradient/examples/example.py
+++ b/examples/policy_gradient/examples/example.py
@@ -47,7 +47,7 @@ if __name__ == "__main__":
     preprocessor = AtariPixelPreprocessor()
 
   print("Using the environment {}.".format(mdp_name))
-  agents = [RemoteAgent(mdp_name, 1, preprocessor, config, False) for _ in range(5)]
+  agents = [RemoteAgent.remote(mdp_name, 1, preprocessor, config, False) for _ in range(5)]
   agent = Agent(mdp_name, 1, preprocessor, config, True)
 
   kl_coeff = config["kl_coeff"]
@@ -55,7 +55,7 @@ if __name__ == "__main__":
   for j in range(1000):
     print("== iteration", j)
     weights = ray.put(agent.get_weights())
-    [a.load_weights(weights) for a in agents]
+    [a.load_weights.remote(weights) for a in agents]
     trajectory, total_reward, traj_len_mean = collect_samples(agents, config["timesteps_per_batch"], 0.995, 1.0, 2000)
     print("total reward is ", total_reward)
     print("trajectory length mean is ", traj_len_mean)
diff --git a/examples/policy_gradient/reinforce/agent.py b/examples/policy_gradient/reinforce/agent.py
index c023bcada..0ae0d937d 100644
--- a/examples/policy_gradient/reinforce/agent.py
+++ b/examples/policy_gradient/reinforce/agent.py
@@ -40,4 +40,4 @@ class Agent(object):
     add_advantage_values(trajectory, gamma, lam, self.reward_filter)
     return trajectory
 
-RemoteAgent = ray.actor(Agent)
+RemoteAgent = ray.remote(Agent)
diff --git a/examples/policy_gradient/reinforce/rollout.py b/examples/policy_gradient/reinforce/rollout.py
index 094e71609..1c23d8f24 100644
--- a/examples/policy_gradient/reinforce/rollout.py
+++ b/examples/policy_gradient/reinforce/rollout.py
@@ -79,7 +79,7 @@ def collect_samples(agents, num_timesteps, gamma, lam, horizon, observation_filt
   total_rewards = []
   traj_len_means = []
   while num_timesteps_so_far < num_timesteps:
-    trajectory_batch = ray.get([agent.compute_trajectory(gamma, lam, horizon) for agent in agents])
+    trajectory_batch = ray.get([agent.compute_trajectory.remote(gamma, lam, horizon) for agent in agents])
     trajectory = concatenate(trajectory_batch)
     total_rewards.append(trajectory["raw_rewards"].sum(axis=0).mean() / len(agents))
     trajectory = flatten(trajectory)
diff --git a/examples/resnet/resnet_main.py b/examples/resnet/resnet_main.py
index 0e4fc4e2b..356752e08 100644
--- a/examples/resnet/resnet_main.py
+++ b/examples/resnet/resnet_main.py
@@ -44,7 +44,7 @@ def get_data(path, size, dataset):
           images[int(2 * size / 3):, :],
           labels)
 
-@ray.actor(num_gpus=use_gpu)
+@ray.remote(num_gpus=use_gpu)
 class ResNetTrainActor(object):
   def __init__(self, data, dataset, num_gpus):
     if num_gpus > 0:
@@ -89,7 +89,7 @@ class ResNetTrainActor(object):
   def get_weights(self):
     return self.model.variables.get_weights()
 
-@ray.actor
+@ray.remote
 class ResNetTestActor(object):
   def __init__(self, data, dataset, eval_batch_count, eval_dir):
     hps = resnet_model.HParams(batch_size=100,
@@ -162,25 +162,25 @@ def train():
   train_data = get_data.remote(FLAGS.train_data_path, 50000, FLAGS.dataset)
   test_data = get_data.remote(FLAGS.eval_data_path, 10000, FLAGS.dataset)
   if num_gpus > 0:
-    train_actors = [ResNetTrainActor(train_data, FLAGS.dataset, num_gpus) for _ in range(num_gpus)]
+    train_actors = [ResNetTrainActor.remote(train_data, FLAGS.dataset, num_gpus) for _ in range(num_gpus)]
   else:
-    train_actors = [ResNetTrainActor(train_data, num_gpus)]
-  test_actor = ResNetTestActor(test_data, FLAGS.dataset, FLAGS.eval_batch_count, FLAGS.eval_dir)
-  print('The log files for tensorboard are stored at ip {}.'.format(ray.get(test_actor.get_ip_addr())))
+    train_actors = [ResNetTrainActor.remote(train_data, num_gpus, 0)]
+  test_actor = ResNetTestActor.remote(test_data, FLAGS.dataset, FLAGS.eval_batch_count, FLAGS.eval_dir)
+  print('The log files for tensorboard are stored at ip {}.'.format(ray.get(test_actor.get_ip_addr.remote())))
   step = 0
-  weight_id = train_actors[0].get_weights()
-  acc_id = test_actor.accuracy(weight_id, step)
+  weight_id = train_actors[0].get_weights.remote()
+  acc_id = test_actor.accuracy.remote(weight_id, step)
   if num_gpus == 0:
     num_gpus = 1
   print("Starting computation.")
   while True:
-    all_weights = ray.get([actor.compute_steps(weight_id) for actor in train_actors])
+    all_weights = ray.get([actor.compute_steps.remote(weight_id) for actor in train_actors])
     mean_weights = {k: sum([weights[k] for weights in all_weights]) / num_gpus for k in all_weights[0]}
     weight_id = ray.put(mean_weights)
     step += 10
     if step % 200 == 0:
       acc = ray.get(acc_id)
-      acc_id = test_actor.accuracy(weight_id, step)
+      acc_id = test_actor.accuracy.remote(weight_id, step)
       print('Step {0}: {1:.6f}'.format(step - 200, acc))
 
 def main(_):
diff --git a/examples/rl_pong/driver.py b/examples/rl_pong/driver.py
index f887575b1..692ee8de6 100644
--- a/examples/rl_pong/driver.py
+++ b/examples/rl_pong/driver.py
@@ -72,7 +72,7 @@ def policy_backward(eph, epx, epdlogp, model):
   dW1 = np.dot(dh.T, epx)
   return {"W1": dW1, "W2": dW2}
 
-@ray.actor
+@ray.remote
 class PongEnv(object):
   def __init__(self):
     # Tell numpy to only use one core. If we don't do this, each actor may try
@@ -158,14 +158,14 @@ if __name__ == "__main__":
   grad_buffer = {k: np.zeros_like(v) for k, v in model.items()}
   # Update the rmsprop memory.
   rmsprop_cache = {k: np.zeros_like(v) for k, v in model.items()}
-  actors = [PongEnv() for _ in range(batch_size)]
+  actors = [PongEnv.remote() for _ in range(batch_size)]
   while True:
     model_id = ray.put(model)
     actions = []
     # Launch tasks to compute gradients from multiple rollouts in parallel.
     start_time = time.time()
     for i in range(batch_size):
-      action_id = actors[i].compute_gradient(model_id)
+      action_id = actors[i].compute_gradient.remote(model_id)
       actions.append(action_id)
     for i in range(batch_size):
       action_id, actions = ray.wait(actions)
diff --git a/python/ray/actor.py b/python/ray/actor.py
index 2d1c2442e..89d2e1406 100644
--- a/python/ray/actor.py
+++ b/python/ray/actor.py
@@ -269,110 +269,124 @@ def export_actor(actor_id, class_id, actor_method_names, num_cpus, num_gpus,
 
 
 def actor(*args, **kwargs):
-  def make_actor_decorator(num_cpus=1, num_gpus=0):
-    def make_actor(Class):
-      class_id = random_actor_class_id()
-      # The list exported will have length 0 if the class has not been exported
-      # yet, and length one if it has. This is just implementing a bool, but we
-      # don't use a bool because we need to modify it inside of the NewClass
-      # constructor.
-      exported = []
+  raise Exception("The @ray.actor decorator is deprecated. Instead, please "
+                  "use @ray.remote.")
 
-      # The function actor_method_call gets called if somebody tries to call a
-      # method on their local actor stub object.
-      def actor_method_call(actor_id, attr, function_signature, *args,
-                            **kwargs):
-        ray.worker.check_connected()
-        ray.worker.check_main_thread()
-        args = signature.extend_args(function_signature, args, kwargs)
 
-        function_id = get_actor_method_function_id(attr)
-        object_ids = ray.worker.global_worker.submit_task(function_id, "",
-                                                          args,
-                                                          actor_id=actor_id)
-        if len(object_ids) == 1:
-          return object_ids[0]
-        elif len(object_ids) > 1:
-          return object_ids
+def make_actor(Class, num_cpus, num_gpus):
+  class_id = random_actor_class_id()
+  # The list exported will have length 0 if the class has not been exported
+  # yet, and length one if it has. This is just implementing a bool, but we
+  # don't use a bool because we need to modify it inside of the NewClass
+  # constructor.
+  exported = []
 
-      class NewClass(object):
-        def __init__(self, *args, **kwargs):
-          self._ray_actor_id = random_actor_id()
-          self._ray_actor_methods = {
-              k: v for (k, v) in inspect.getmembers(
-                  Class, predicate=(lambda x: (inspect.isfunction(x) or
-                                               inspect.ismethod(x))))}
-          # Extract the signatures of each of the methods. This will be used to
-          # catch some errors if the methods are called with inappropriate
-          # arguments.
-          self._ray_method_signatures = dict()
-          for k, v in self._ray_actor_methods.items():
-            # Print a warning message if the method signature is not supported.
-            # We don't raise an exception because if the actor inherits from a
-            # class that has a method whose signature we don't support, we
-            # there may not be much the user can do about it.
-            signature.check_signature_supported(v, warn=True)
-            self._ray_method_signatures[k] = signature.extract_signature(
-                v, ignore_first=True)
+  # The function actor_method_call gets called if somebody tries to call a
+  # method on their local actor stub object.
+  def actor_method_call(actor_id, attr, function_signature, *args, **kwargs):
+    ray.worker.check_connected()
+    ray.worker.check_main_thread()
+    args = signature.extend_args(function_signature, args, kwargs)
 
-          # Export the actor class if it has not been exported yet.
-          if len(exported) == 0:
-            export_actor_class(class_id, Class, self._ray_actor_methods.keys(),
-                               ray.worker.global_worker)
-            exported.append(0)
-          # Export the actor.
-          export_actor(self._ray_actor_id, class_id,
-                       self._ray_actor_methods.keys(), num_cpus, num_gpus,
-                       ray.worker.global_worker)
-          # Call __init__ as a remote function.
-          if "__init__" in self._ray_actor_methods.keys():
-            actor_method_call(self._ray_actor_id, "__init__",
-                              self._ray_method_signatures["__init__"],
-                              *args, **kwargs)
-          else:
-            print("WARNING: this object has no __init__ method.")
+    function_id = get_actor_method_function_id(attr)
+    object_ids = ray.worker.global_worker.submit_task(function_id, "", args,
+                                                      actor_id=actor_id)
+    if len(object_ids) == 1:
+      return object_ids[0]
+    elif len(object_ids) > 1:
+      return object_ids
 
-        # Make tab completion work.
-        def __dir__(self):
-          return self._ray_actor_methods
+  class ActorMethod(object):
+    def __init__(self, method_name, actor_id, method_signature):
+      self.method_name = method_name
+      self.actor_id = actor_id
+      self.method_signature = method_signature
 
-        def __getattribute__(self, attr):
-          # The following is needed so we can still access self.actor_methods.
-          if attr in ["_ray_actor_id", "_ray_actor_methods",
-                      "_ray_method_signatures"]:
-            return super(NewClass, self).__getattribute__(attr)
-          if attr in self._ray_actor_methods.keys():
-            return lambda *args, **kwargs: actor_method_call(
-                self._ray_actor_id, attr, self._ray_method_signatures[attr],
-                *args, **kwargs)
-          # There is no method with this name, so raise an exception.
-          raise AttributeError("'{}' Actor object has no attribute '{}'"
-                               .format(Class, attr))
+    def __call__(self, *args, **kwargs):
+      raise Exception("Actor methods cannot be called directly. Instead "
+                      "of running 'object.{}()', try 'object.{}.remote()'."
+                      .format(self.method_name, self.method_name))
 
-        def __repr__(self):
-          return "Actor(" + self._ray_actor_id.hex() + ")"
+    def remote(self, *args, **kwargs):
+      return actor_method_call(self.actor_id, self.method_name,
+                               self.method_signature, *args, **kwargs)
 
-      return NewClass
-    return make_actor
+  class NewClass(object):
+    def __init__(self, *args, **kwargs):
+      raise Exception("Actor classes cannot be instantiated directly. "
+                      "Instead of running '{}()', try '{}.remote()'."
+                      .format(Class.__name__, Class.__name__))
 
-  if len(args) == 1 and len(kwargs) == 0 and callable(args[0]):
-    # In this case, the actor decorator was applied directly to a class
-    # definition.
-    Class = args[0]
-    return make_actor_decorator(num_cpus=1, num_gpus=0)(Class)
+    @classmethod
+    def remote(cls, *args, **kwargs):
+      actor_object = cls.__new__(cls)
+      actor_object._manual_init(*args, **kwargs)
+      return actor_object
 
-  # In this case, the actor decorator is something like @ray.actor(num_gpus=1).
-  if len(args) == 0 and len(kwargs) > 0 and all([key
-                                                 in ["num_cpus", "num_gpus"]
-                                                 for key in kwargs.keys()]):
-    num_cpus = kwargs["num_cpus"] if "num_cpus" in kwargs.keys() else 1
-    num_gpus = kwargs["num_gpus"] if "num_gpus" in kwargs.keys() else 0
-    return make_actor_decorator(num_cpus=num_cpus, num_gpus=num_gpus)
+    def _manual_init(self, *args, **kwargs):
+      self._ray_actor_id = random_actor_id()
+      self._ray_actor_methods = {
+          k: v for (k, v) in inspect.getmembers(
+              Class, predicate=(lambda x: (inspect.isfunction(x) or
+                                           inspect.ismethod(x))))}
+      # Extract the signatures of each of the methods. This will be used to
+      # catch some errors if the methods are called with inappropriate
+      # arguments.
+      self._ray_method_signatures = dict()
+      for k, v in self._ray_actor_methods.items():
+        # Print a warning message if the method signature is not supported.
+        # We don't raise an exception because if the actor inherits from a
+        # class that has a method whose signature we don't support, we
+        # there may not be much the user can do about it.
+        signature.check_signature_supported(v, warn=True)
+        self._ray_method_signatures[k] = signature.extract_signature(
+            v, ignore_first=True)
 
-  raise Exception("The ray.actor decorator must either be applied with no "
-                  "arguments as in '@ray.actor', or it must be applied using "
-                  "some of the arguments 'num_cpus' or 'num_gpus' as in "
-                  "'ray.actor(num_gpus=1)'.")
+      # Create objects to wrap method invocations. This is done so that we
+      # can invoke methods with actor.method.remote() instead of
+      # actor.method().
+      self._actor_method_invokers = dict()
+      for k, v in self._ray_actor_methods.items():
+        self._actor_method_invokers[k] = ActorMethod(
+            k, self._ray_actor_id, self._ray_method_signatures[k])
+
+      # Export the actor class if it has not been exported yet.
+      if len(exported) == 0:
+        export_actor_class(class_id, Class, self._ray_actor_methods.keys(),
+                           ray.worker.global_worker)
+        exported.append(0)
+      # Export the actor.
+      export_actor(self._ray_actor_id, class_id,
+                   self._ray_actor_methods.keys(), num_cpus, num_gpus,
+                   ray.worker.global_worker)
+      # Call __init__ as a remote function.
+      if "__init__" in self._ray_actor_methods.keys():
+        actor_method_call(self._ray_actor_id, "__init__",
+                          self._ray_method_signatures["__init__"],
+                          *args, **kwargs)
+      else:
+        print("WARNING: this object has no __init__ method.")
+
+    # Make tab completion work.
+    def __dir__(self):
+      return self._ray_actor_methods
+
+    def __getattribute__(self, attr):
+      # The following is needed so we can still access self.actor_methods.
+      if attr in ["_manual_init", "_ray_actor_id", "_ray_actor_methods",
+                  "_actor_method_invokers", "_ray_method_signatures"]:
+        return super(NewClass, self).__getattribute__(attr)
+      if attr in self._ray_actor_methods.keys():
+        return self._actor_method_invokers[attr]
+      # There is no method with this name, so raise an exception.
+      raise AttributeError("'{}' Actor object has no attribute '{}'"
+                           .format(Class, attr))
+
+    def __repr__(self):
+      return "Actor(" + self._ray_actor_id.hex() + ")"
+
+  return NewClass
 
 
 ray.worker.global_worker.fetch_and_register_actor = fetch_and_register_actor
+ray.worker.global_worker.make_actor = make_actor
diff --git a/python/ray/worker.py b/python/ray/worker.py
index d91363263..7e932852a 100644
--- a/python/ray/worker.py
+++ b/python/ray/worker.py
@@ -448,6 +448,7 @@ class Worker(object):
     self.cached_remote_functions = []
     self.cached_functions_to_run = []
     self.fetch_and_register_actor = None
+    self.make_actor = None
     self.actors = {}
     # Use a defaultdict for the actor counts. If this is accessed with a
     # missing key, the default value of 0 is returned, and that key value pair
@@ -2105,7 +2106,15 @@ def remote(*args, **kwargs):
   worker = global_worker
 
   def make_remote_decorator(num_return_vals, num_cpus, num_gpus, func_id=None):
-    def remote_decorator(func):
+    def remote_decorator(func_or_class):
+      if inspect.isfunction(func_or_class):
+        return remote_function_decorator(func_or_class)
+      if inspect.isclass(func_or_class):
+        return worker.make_actor(func_or_class, num_cpus, num_gpus)
+      raise Exception("The @ray.remote decorator must be applied to either a "
+                      "function or to a class.")
+
+    def remote_function_decorator(func):
       func_name = "{}.{}".format(func.__module__, func.__name__)
       if func_id is None:
         function_id = compute_function_id(func_name, func)
@@ -2195,6 +2204,8 @@ def remote(*args, **kwargs):
     assert len(args) == 0 and ("num_return_vals" in kwargs or
                                "num_cpus" in kwargs or
                                "num_gpus" in kwargs), error_string
+    for key in kwargs:
+      assert key in ["num_return_vals", "num_cpus", "num_gpus"], error_string
     assert "function_id" not in kwargs
     return make_remote_decorator(num_return_vals, num_cpus, num_gpus)
 
diff --git a/test/actor_test.py b/test/actor_test.py
index 7709ca4ea..171d1a937 100644
--- a/test/actor_test.py
+++ b/test/actor_test.py
@@ -16,7 +16,7 @@ class ActorAPI(unittest.TestCase):
   def testKeywordArgs(self):
     ray.init(num_workers=0, driver_mode=ray.SILENT_MODE)
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def __init__(self, arg0, arg1=1, arg2="a"):
         self.arg0 = arg0
@@ -26,43 +26,45 @@ class ActorAPI(unittest.TestCase):
       def get_values(self, arg0, arg1=2, arg2="b"):
         return self.arg0 + arg0, self.arg1 + arg1, self.arg2 + arg2
 
-    actor = Actor(0)
-    self.assertEqual(ray.get(actor.get_values(1)), (1, 3, "ab"))
+    actor = Actor.remote(0)
+    self.assertEqual(ray.get(actor.get_values.remote(1)), (1, 3, "ab"))
 
-    actor = Actor(1, 2)
-    self.assertEqual(ray.get(actor.get_values(2, 3)), (3, 5, "ab"))
+    actor = Actor.remote(1, 2)
+    self.assertEqual(ray.get(actor.get_values.remote(2, 3)), (3, 5, "ab"))
 
-    actor = Actor(1, 2, "c")
-    self.assertEqual(ray.get(actor.get_values(2, 3, "d")), (3, 5, "cd"))
+    actor = Actor.remote(1, 2, "c")
+    self.assertEqual(ray.get(actor.get_values.remote(2, 3, "d")), (3, 5, "cd"))
 
-    actor = Actor(1, arg2="c")
-    self.assertEqual(ray.get(actor.get_values(0, arg2="d")), (1, 3, "cd"))
-    self.assertEqual(ray.get(actor.get_values(0, arg2="d", arg1=0)),
+    actor = Actor.remote(1, arg2="c")
+    self.assertEqual(ray.get(actor.get_values.remote(0, arg2="d")),
+                     (1, 3, "cd"))
+    self.assertEqual(ray.get(actor.get_values.remote(0, arg2="d", arg1=0)),
                      (1, 1, "cd"))
 
-    actor = Actor(1, arg2="c", arg1=2)
-    self.assertEqual(ray.get(actor.get_values(0, arg2="d")), (1, 4, "cd"))
-    self.assertEqual(ray.get(actor.get_values(0, arg2="d", arg1=0)),
+    actor = Actor.remote(1, arg2="c", arg1=2)
+    self.assertEqual(ray.get(actor.get_values.remote(0, arg2="d")),
+                     (1, 4, "cd"))
+    self.assertEqual(ray.get(actor.get_values.remote(0, arg2="d", arg1=0)),
                      (1, 2, "cd"))
 
     # Make sure we get an exception if the constructor is called incorrectly.
     with self.assertRaises(Exception):
-      actor = Actor()
+      actor = Actor.remote()
 
     with self.assertRaises(Exception):
-      actor = Actor(0, 1, 2, arg3=3)
+      actor = Actor.remote(0, 1, 2, arg3=3)
 
     # Make sure we get an exception if the method is called incorrectly.
-    actor = Actor(1)
+    actor = Actor.remote(1)
     with self.assertRaises(Exception):
-      ray.get(actor.get_values())
+      ray.get(actor.get_values.remote())
 
     ray.worker.cleanup()
 
   def testVariableNumberOfArgs(self):
     ray.init(num_workers=0)
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def __init__(self, arg0, arg1=1, *args):
         self.arg0 = arg0
@@ -72,21 +74,21 @@ class ActorAPI(unittest.TestCase):
       def get_values(self, arg0, arg1=2, *args):
         return self.arg0 + arg0, self.arg1 + arg1, self.args, args
 
-    actor = Actor(0)
-    self.assertEqual(ray.get(actor.get_values(1)), (1, 3, (), ()))
+    actor = Actor.remote(0)
+    self.assertEqual(ray.get(actor.get_values.remote(1)), (1, 3, (), ()))
 
-    actor = Actor(1, 2)
-    self.assertEqual(ray.get(actor.get_values(2, 3)), (3, 5, (), ()))
+    actor = Actor.remote(1, 2)
+    self.assertEqual(ray.get(actor.get_values.remote(2, 3)), (3, 5, (), ()))
 
-    actor = Actor(1, 2, "c")
-    self.assertEqual(ray.get(actor.get_values(2, 3, "d")),
+    actor = Actor.remote(1, 2, "c")
+    self.assertEqual(ray.get(actor.get_values.remote(2, 3, "d")),
                      (3, 5, ("c",), ("d",)))
 
-    actor = Actor(1, 2, "a", "b", "c", "d")
-    self.assertEqual(ray.get(actor.get_values(2, 3, 1, 2, 3, 4)),
+    actor = Actor.remote(1, 2, "a", "b", "c", "d")
+    self.assertEqual(ray.get(actor.get_values.remote(2, 3, 1, 2, 3, 4)),
                      (3, 5, ("a", "b", "c", "d"), (1, 2, 3, 4)))
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def __init__(self, *args):
         self.args = args
@@ -94,19 +96,19 @@ class ActorAPI(unittest.TestCase):
       def get_values(self, *args):
         return self.args, args
 
-    a = Actor()
-    self.assertEqual(ray.get(a.get_values()), ((), ()))
-    a = Actor(1)
-    self.assertEqual(ray.get(a.get_values(2)), ((1,), (2,)))
-    a = Actor(1, 2)
-    self.assertEqual(ray.get(a.get_values(3, 4)), ((1, 2), (3, 4)))
+    a = Actor.remote()
+    self.assertEqual(ray.get(a.get_values.remote()), ((), ()))
+    a = Actor.remote(1)
+    self.assertEqual(ray.get(a.get_values.remote(2)), ((1,), (2,)))
+    a = Actor.remote(1, 2)
+    self.assertEqual(ray.get(a.get_values.remote(3, 4)), ((1, 2), (3, 4)))
 
     ray.worker.cleanup()
 
   def testNoArgs(self):
     ray.init(num_workers=0)
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def __init__(self):
         pass
@@ -114,8 +116,8 @@ class ActorAPI(unittest.TestCase):
       def get_values(self):
         pass
 
-    actor = Actor()
-    self.assertEqual(ray.get(actor.get_values()), None)
+    actor = Actor.remote()
+    self.assertEqual(ray.get(actor.get_values.remote()), None)
 
     ray.worker.cleanup()
 
@@ -123,13 +125,13 @@ class ActorAPI(unittest.TestCase):
     # If no __init__ method is provided, that should not be a problem.
     ray.init(num_workers=0)
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def get_values(self):
         pass
 
-    actor = Actor()
-    self.assertEqual(ray.get(actor.get_values()), None)
+    actor = Actor.remote()
+    self.assertEqual(ray.get(actor.get_values.remote()), None)
 
     ray.worker.cleanup()
 
@@ -141,7 +143,7 @@ class ActorAPI(unittest.TestCase):
         self.x = x
     ray.register_class(Foo)
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def __init__(self, f2):
         self.f1 = Foo(1)
@@ -153,11 +155,11 @@ class ActorAPI(unittest.TestCase):
       def get_values2(self, f3):
         return self.f1, self.f2, f3
 
-    actor = Actor(Foo(2))
-    results1 = ray.get(actor.get_values1())
+    actor = Actor.remote(Foo(2))
+    results1 = ray.get(actor.get_values1.remote())
     self.assertEqual(results1[0].x, 1)
     self.assertEqual(results1[1].x, 2)
-    results2 = ray.get(actor.get_values2(Foo(3)))
+    results2 = ray.get(actor.get_values2.remote(Foo(3)))
     self.assertEqual(results2[0].x, 1)
     self.assertEqual(results2[1].x, 2)
     self.assertEqual(results2[2].x, 3)
@@ -173,39 +175,39 @@ class ActorAPI(unittest.TestCase):
 
     # This is an invalid way of using the actor decorator.
     with self.assertRaises(Exception):
-      @ray.actor()
+      @ray.remote()
       class Actor(object):
         def __init__(self):
           pass
 
     # This is an invalid way of using the actor decorator.
     with self.assertRaises(Exception):
-      @ray.actor(invalid_kwarg=0)  # noqa: F811
+      @ray.remote(invalid_kwarg=0)  # noqa: F811
       class Actor(object):
         def __init__(self):
           pass
 
     # This is an invalid way of using the actor decorator.
     with self.assertRaises(Exception):
-      @ray.actor(num_cpus=0, invalid_kwarg=0)  # noqa: F811
+      @ray.remote(num_cpus=0, invalid_kwarg=0)  # noqa: F811
       class Actor(object):
         def __init__(self):
           pass
 
     # This is a valid way of using the decorator.
-    @ray.actor(num_cpus=1)  # noqa: F811
+    @ray.remote(num_cpus=1)  # noqa: F811
     class Actor(object):
       def __init__(self):
         pass
 
     # This is a valid way of using the decorator.
-    @ray.actor(num_gpus=1)  # noqa: F811
+    @ray.remote(num_gpus=1)  # noqa: F811
     class Actor(object):
       def __init__(self):
         pass
 
     # This is a valid way of using the decorator.
-    @ray.actor(num_cpus=1, num_gpus=1)  # noqa: F811
+    @ray.remote(num_cpus=1, num_gpus=1)  # noqa: F811
     class Actor(object):
       def __init__(self):
         pass
@@ -215,7 +217,7 @@ class ActorAPI(unittest.TestCase):
   def testRandomIDGeneration(self):
     ray.init(num_workers=0)
 
-    @ray.actor
+    @ray.remote
     class Foo(object):
       def __init__(self):
         pass
@@ -224,10 +226,10 @@ class ActorAPI(unittest.TestCase):
     # actor IDs.
     np.random.seed(1234)
     random.seed(1234)
-    f1 = Foo()
+    f1 = Foo.remote()
     np.random.seed(1234)
     random.seed(1234)
-    f2 = Foo()
+    f2 = Foo.remote()
 
     self.assertNotEqual(f1._ray_actor_id.id(), f2._ray_actor_id.id())
 
@@ -239,7 +241,7 @@ class ActorMethods(unittest.TestCase):
   def testDefineActor(self):
     ray.init()
 
-    @ray.actor
+    @ray.remote
     class Test(object):
       def __init__(self, x):
         self.x = x
@@ -247,15 +249,19 @@ class ActorMethods(unittest.TestCase):
       def f(self, y):
         return self.x + y
 
-    t = Test(2)
-    self.assertEqual(ray.get(t.f(1)), 3)
+    t = Test.remote(2)
+    self.assertEqual(ray.get(t.f.remote(1)), 3)
+
+    # Make sure that calling an actor method directly raises an exception.
+    with self.assertRaises(Exception):
+      t.f(1)
 
     ray.worker.cleanup()
 
   def testActorState(self):
     ray.init()
 
-    @ray.actor
+    @ray.remote
     class Counter(object):
       def __init__(self):
         self.value = 0
@@ -266,14 +272,14 @@ class ActorMethods(unittest.TestCase):
       def value(self):
         return self.value
 
-    c1 = Counter()
-    c1.increase()
-    self.assertEqual(ray.get(c1.value()), 1)
+    c1 = Counter.remote()
+    c1.increase.remote()
+    self.assertEqual(ray.get(c1.value.remote()), 1)
 
-    c2 = Counter()
-    c2.increase()
-    c2.increase()
-    self.assertEqual(ray.get(c2.value()), 2)
+    c2 = Counter.remote()
+    c2.increase.remote()
+    c2.increase.remote()
+    self.assertEqual(ray.get(c2.value.remote()), 2)
 
     ray.worker.cleanup()
 
@@ -281,7 +287,7 @@ class ActorMethods(unittest.TestCase):
     # Create a bunch of actors and call a bunch of methods on all of them.
     ray.init(num_workers=0)
 
-    @ray.actor
+    @ray.remote
     class Counter(object):
       def __init__(self, value):
         self.value = value
@@ -296,11 +302,11 @@ class ActorMethods(unittest.TestCase):
     num_actors = 20
     num_increases = 50
     # Create multiple actors.
-    actors = [Counter(i) for i in range(num_actors)]
+    actors = [Counter.remote(i) for i in range(num_actors)]
     results = []
     # Call each actor's method a bunch of times.
     for i in range(num_actors):
-      results += [actors[i].increase() for _ in range(num_increases)]
+      results += [actors[i].increase.remote() for _ in range(num_increases)]
     result_values = ray.get(results)
     for i in range(num_actors):
       self.assertEqual(
@@ -308,12 +314,12 @@ class ActorMethods(unittest.TestCase):
           list(range(i + 1, num_increases + i + 1)))
 
     # Reset the actor values.
-    [actor.reset() for actor in actors]
+    [actor.reset.remote() for actor in actors]
 
     # Interweave the method calls on the different actors.
     results = []
     for j in range(num_increases):
-      results += [actor.increase() for actor in actors]
+      results += [actor.increase.remote() for actor in actors]
     result_values = ray.get(results)
     for j in range(num_increases):
       self.assertEqual(result_values[(num_actors * j):(num_actors * (j + 1))],
@@ -340,7 +346,7 @@ class ActorNesting(unittest.TestCase):
     def g(x):
       return ray.get(f.remote(x))
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def __init__(self, x):
         self.x = x
@@ -360,16 +366,16 @@ class ActorNesting(unittest.TestCase):
       def h(self, object_ids):
         return ray.get(object_ids)
 
-    actor = Actor(1)
-    values = ray.get(actor.get_values())
+    actor = Actor.remote(1)
+    values = ray.get(actor.get_values.remote())
     self.assertEqual(values[0], 1)
     self.assertEqual(values[1], val2)
     self.assertEqual(ray.get(values[2]), list(range(1, 6)))
     self.assertEqual(values[3], list(range(1, 6)))
 
-    self.assertEqual(ray.get(ray.get(actor.f())), list(range(1, 6)))
-    self.assertEqual(ray.get(actor.g()), list(range(1, 6)))
-    self.assertEqual(ray.get(actor.h([f.remote(i) for i in range(5)])),
+    self.assertEqual(ray.get(ray.get(actor.f.remote())), list(range(1, 6)))
+    self.assertEqual(ray.get(actor.g.remote()), list(range(1, 6)))
+    self.assertEqual(ray.get(actor.h.remote([f.remote(i) for i in range(5)])),
                      list(range(1, 6)))
 
     ray.worker.cleanup()
@@ -378,27 +384,27 @@ class ActorNesting(unittest.TestCase):
     # Make sure we can use remote funtions within actors.
     ray.init(num_cpus=10)
 
-    @ray.actor
+    @ray.remote
     class Actor1(object):
       def __init__(self, x):
         self.x = x
 
       def new_actor(self, z):
-        @ray.actor
+        @ray.remote
         class Actor2(object):
           def __init__(self, x):
             self.x = x
 
           def get_value(self):
             return self.x
-        self.actor2 = Actor2(z)
+        self.actor2 = Actor2.remote(z)
 
       def get_values(self, z):
         self.new_actor(z)
-        return self.x, ray.get(self.actor2.get_value())
+        return self.x, ray.get(self.actor2.get_value.remote())
 
-    actor1 = Actor1(3)
-    self.assertEqual(ray.get(actor1.get_values(5)), (3, 5))
+    actor1 = Actor1.remote(3)
+    self.assertEqual(ray.get(actor1.get_values.remote(5)), (3, 5))
 
     ray.worker.cleanup()
 
@@ -408,14 +414,14 @@ class ActorNesting(unittest.TestCase):
   #   # Make sure we can use remote funtions within actors.
   #   ray.init(num_cpus=10)
   #
-  #   @ray.actor
+  #   @ray.remote
   #   class Actor1(object):
   #     def __init__(self, x):
   #       self.x = x
   #     def get_val(self):
   #       return self.x
   #
-  #   @ray.actor
+  #   @ray.remote
   #   class Actor2(object):
   #     def __init__(self, x, y):
   #       self.x = x
@@ -435,15 +441,15 @@ class ActorNesting(unittest.TestCase):
 
     @ray.remote
     def f(x, n):
-      @ray.actor
+      @ray.remote
       class Actor1(object):
         def __init__(self, x):
           self.x = x
 
         def get_value(self):
           return self.x
-      actor = Actor1(x)
-      return ray.get([actor.get_value() for _ in range(n)])
+      actor = Actor1.remote(x)
+      return ray.get([actor.get_value.remote() for _ in range(n)])
 
     self.assertEqual(ray.get(f.remote(3, 1)), [3])
     self.assertEqual(ray.get([f.remote(i, 20) for i in range(10)]),
@@ -456,7 +462,7 @@ class ActorNesting(unittest.TestCase):
   #   # Make sure we can create and use actors within remote funtions.
   #   ray.init(num_cpus=10)
   #
-  #   @ray.actor
+  #   @ray.remote
   #   class Actor1(object):
   #     def __init__(self, x):
   #       self.x = x
@@ -487,7 +493,7 @@ class ActorNesting(unittest.TestCase):
 
     @ray.remote
     def g():
-      @ray.actor
+      @ray.remote
       class Actor(object):
         def __init__(self):
           # This should use the last version of f.
@@ -496,8 +502,8 @@ class ActorNesting(unittest.TestCase):
         def get_val(self):
           return self.x
 
-      actor = Actor()
-      return ray.get(actor.get_val())
+      actor = Actor.remote()
+      return ray.get(actor.get_val.remote())
 
     self.assertEqual(ray.get(g.remote()), num_remote_functions - 1)
 
@@ -521,7 +527,7 @@ class ActorInheritance(unittest.TestCase):
       def g(self, y):
         return self.x + y
 
-    @ray.actor
+    @ray.remote
     class Actor(Foo):
       def __init__(self, x):
         Foo.__init__(self, x)
@@ -529,9 +535,9 @@ class ActorInheritance(unittest.TestCase):
       def get_value(self):
         return self.f()
 
-    actor = Actor(1)
-    self.assertEqual(ray.get(actor.get_value()), 1)
-    self.assertEqual(ray.get(actor.g(5)), 6)
+    actor = Actor.remote(1)
+    self.assertEqual(ray.get(actor.get_value.remote()), 1)
+    self.assertEqual(ray.get(actor.g.remote(5)), 6)
 
     ray.worker.cleanup()
 
@@ -542,12 +548,12 @@ class ActorSchedulingProperties(unittest.TestCase):
     # Make sure that regular remote functions are not scheduled on actors.
     ray.init(num_workers=0)
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def __init__(self):
         pass
 
-    Actor()
+    Actor.remote()
 
     @ray.remote
     def f():
@@ -568,13 +574,13 @@ class ActorsOnMultipleNodes(unittest.TestCase):
   def testActorsOnNodesWithNoCPUs(self):
     ray.init(num_cpus=0)
 
-    @ray.actor
+    @ray.remote
     class Foo(object):
       def __init__(self):
         pass
 
     with self.assertRaises(Exception):
-      Foo()
+      Foo.remote()
 
     ray.worker.cleanup()
 
@@ -583,7 +589,7 @@ class ActorsOnMultipleNodes(unittest.TestCase):
     ray.worker._init(start_ray_local=True, num_workers=0,
                      num_local_schedulers=num_local_schedulers)
 
-    @ray.actor
+    @ray.remote
     class Actor1(object):
       def __init__(self):
         pass
@@ -599,8 +605,8 @@ class ActorsOnMultipleNodes(unittest.TestCase):
     # Make sure that actors are spread between the local schedulers.
     attempts = 0
     while attempts < num_attempts:
-      actors = [Actor1() for _ in range(num_actors)]
-      locations = ray.get([actor.get_location() for actor in actors])
+      actors = [Actor1.remote() for _ in range(num_actors)]
+      locations = ray.get([actor.get_location.remote() for actor in actors])
       names = set(locations)
       counts = [locations.count(name) for name in names]
       print("Counts are {}.".format(counts))
@@ -614,7 +620,7 @@ class ActorsOnMultipleNodes(unittest.TestCase):
     results = []
     for _ in range(1000):
       index = np.random.randint(num_actors)
-      results.append(actors[index].get_location())
+      results.append(actors[index].get_location.remote())
     ray.get(results)
 
     ray.worker.cleanup()
@@ -630,7 +636,7 @@ class ActorsWithGPUs(unittest.TestCase):
         num_local_schedulers=num_local_schedulers,
         num_gpus=(num_local_schedulers * [num_gpus_per_scheduler]))
 
-    @ray.actor(num_gpus=1)
+    @ray.remote(num_gpus=1)
     class Actor1(object):
       def __init__(self):
         self.gpu_ids = ray.get_gpu_ids()
@@ -641,10 +647,10 @@ class ActorsWithGPUs(unittest.TestCase):
                 tuple(self.gpu_ids))
 
     # Create one actor per GPU.
-    actors = [Actor1() for _
+    actors = [Actor1.remote() for _
               in range(num_local_schedulers * num_gpus_per_scheduler)]
     # Make sure that no two actors are assigned to the same GPU.
-    locations_and_ids = ray.get([actor.get_location_and_ids()
+    locations_and_ids = ray.get([actor.get_location_and_ids.remote()
                                  for actor in actors])
     node_names = set([location for location, gpu_id in locations_and_ids])
     self.assertEqual(len(node_names), num_local_schedulers)
@@ -656,7 +662,7 @@ class ActorsWithGPUs(unittest.TestCase):
 
     # Creating a new actor should fail because all of the GPUs are being used.
     with self.assertRaises(Exception):
-      Actor1()
+      Actor1.remote()
 
     ray.worker.cleanup()
 
@@ -668,7 +674,7 @@ class ActorsWithGPUs(unittest.TestCase):
         num_local_schedulers=num_local_schedulers,
         num_gpus=(num_local_schedulers * [num_gpus_per_scheduler]))
 
-    @ray.actor(num_gpus=2)
+    @ray.remote(num_gpus=2)
     class Actor1(object):
       def __init__(self):
         self.gpu_ids = ray.get_gpu_ids()
@@ -678,9 +684,9 @@ class ActorsWithGPUs(unittest.TestCase):
                 tuple(self.gpu_ids))
 
     # Create some actors.
-    actors = [Actor1() for _ in range(num_local_schedulers * 2)]
+    actors = [Actor1.remote() for _ in range(num_local_schedulers * 2)]
     # Make sure that no two actors are assigned to the same GPU.
-    locations_and_ids = ray.get([actor.get_location_and_ids()
+    locations_and_ids = ray.get([actor.get_location_and_ids.remote()
                                  for actor in actors])
     node_names = set([location for location, gpu_id in locations_and_ids])
     self.assertEqual(len(node_names), num_local_schedulers)
@@ -694,10 +700,10 @@ class ActorsWithGPUs(unittest.TestCase):
 
     # Creating a new actor should fail because all of the GPUs are being used.
     with self.assertRaises(Exception):
-      Actor1()
+      Actor1.remote()
 
     # We should be able to create more actors that use only a single GPU.
-    @ray.actor(num_gpus=1)
+    @ray.remote(num_gpus=1)
     class Actor2(object):
       def __init__(self):
         self.gpu_ids = ray.get_gpu_ids()
@@ -707,9 +713,9 @@ class ActorsWithGPUs(unittest.TestCase):
                 tuple(self.gpu_ids))
 
     # Create some actors.
-    actors = [Actor2() for _ in range(num_local_schedulers)]
+    actors = [Actor2.remote() for _ in range(num_local_schedulers)]
     # Make sure that no two actors are assigned to the same GPU.
-    locations_and_ids = ray.get([actor.get_location_and_ids()
+    locations_and_ids = ray.get([actor.get_location_and_ids.remote()
                                  for actor in actors])
     self.assertEqual(node_names,
                      set([location for location, gpu_id in locations_and_ids]))
@@ -721,7 +727,7 @@ class ActorsWithGPUs(unittest.TestCase):
 
     # Creating a new actor should fail because all of the GPUs are being used.
     with self.assertRaises(Exception):
-      Actor2()
+      Actor2.remote()
 
     ray.worker.cleanup()
 
@@ -731,7 +737,7 @@ class ActorsWithGPUs(unittest.TestCase):
     ray.worker._init(start_ray_local=True, num_workers=0,
                      num_local_schedulers=3, num_gpus=[0, 5, 10])
 
-    @ray.actor(num_gpus=1)
+    @ray.remote(num_gpus=1)
     class Actor1(object):
       def __init__(self):
         self.gpu_ids = ray.get_gpu_ids()
@@ -741,9 +747,9 @@ class ActorsWithGPUs(unittest.TestCase):
                 tuple(self.gpu_ids))
 
     # Create some actors.
-    actors = [Actor1() for _ in range(0 + 5 + 10)]
+    actors = [Actor1.remote() for _ in range(0 + 5 + 10)]
     # Make sure that no two actors are assigned to the same GPU.
-    locations_and_ids = ray.get([actor.get_location_and_ids()
+    locations_and_ids = ray.get([actor.get_location_and_ids.remote()
                                 for actor in actors])
     node_names = set([location for location, gpu_id in locations_and_ids])
     self.assertEqual(len(node_names), 2)
@@ -756,7 +762,7 @@ class ActorsWithGPUs(unittest.TestCase):
 
     # Creating a new actor should fail because all of the GPUs are being used.
     with self.assertRaises(Exception):
-      Actor1()
+      Actor1.remote()
 
     ray.worker.cleanup()
 
@@ -770,7 +776,7 @@ class ActorsWithGPUs(unittest.TestCase):
 
     @ray.remote
     def create_actors(n):
-      @ray.actor(num_gpus=1)
+      @ray.remote(num_gpus=1)
       class Actor(object):
         def __init__(self):
           self.gpu_ids = ray.get_gpu_ids()
@@ -780,12 +786,12 @@ class ActorsWithGPUs(unittest.TestCase):
                   tuple(self.gpu_ids))
       # Create n actors.
       for _ in range(n):
-        Actor()
+        Actor.remote()
 
     ray.get([create_actors.remote(num_gpus_per_scheduler)
              for _ in range(num_local_schedulers)])
 
-    @ray.actor(num_gpus=1)
+    @ray.remote(num_gpus=1)
     class Actor(object):
       def __init__(self):
         self.gpu_ids = ray.get_gpu_ids()
@@ -796,7 +802,7 @@ class ActorsWithGPUs(unittest.TestCase):
 
     # All the GPUs should be used up now.
     with self.assertRaises(Exception):
-      Actor()
+      Actor.remote()
 
     ray.worker.cleanup()
 
@@ -844,7 +850,7 @@ class ActorsWithGPUs(unittest.TestCase):
       return (ray.worker.global_worker.plasma_client.store_socket_name,
               tuple(gpu_ids), [t1, t2])
 
-    @ray.actor(num_gpus=1)
+    @ray.remote(num_gpus=1)
     class Actor1(object):
       def __init__(self):
         self.gpu_ids = ray.get_gpu_ids()
@@ -883,8 +889,8 @@ class ActorsWithGPUs(unittest.TestCase):
       check_intervals_non_overlapping(locations_to_intervals[locations])
 
     # Create an actor that uses a GPU.
-    a = Actor1()
-    actor_location = ray.get(a.get_location_and_ids())
+    a = Actor1.remote()
+    actor_location = ray.get(a.get_location_and_ids.remote())
     actor_location = (actor_location[0], actor_location[1][0])
     # This check makes sure that actor_location is formatted the same way that
     # the keys of locations_to_intervals are formatted.
@@ -903,8 +909,8 @@ class ActorsWithGPUs(unittest.TestCase):
     self.assertNotIn(actor_location, locations_to_intervals)
 
     # Create several more actors that use GPUs.
-    actors = [Actor1() for _ in range(3)]
-    actor_locations = ray.get([actor.get_location_and_ids()
+    actors = [Actor1.remote() for _ in range(3)]
+    actor_locations = ray.get([actor.get_location_and_ids.remote()
                                for actor in actors])
 
     # Run a bunch of GPU tasks.
@@ -922,11 +928,11 @@ class ActorsWithGPUs(unittest.TestCase):
       self.assertNotIn(location, locations_to_intervals)
 
     # Create more actors to fill up all the GPUs.
-    more_actors = [Actor1() for _ in
+    more_actors = [Actor1.remote() for _ in
                    range(num_local_schedulers *
                          num_gpus_per_scheduler - 1 - 3)]
     # Wait for the actors to finish being created.
-    ray.get([actor.get_location_and_ids() for actor in more_actors])
+    ray.get([actor.get_location_and_ids.remote() for actor in more_actors])
 
     # Now if we run some GPU tasks, they should not be scheduled.
     results = [f1.remote() for _ in range(30)]
@@ -947,7 +953,7 @@ class ActorsWithGPUs(unittest.TestCase):
       assert len(gpu_ids) == 1
       return gpu_ids[0]
 
-    @ray.actor(num_gpus=1)
+    @ray.remote(num_gpus=1)
     class Actor(object):
       def __init__(self):
         self.gpu_ids = ray.get_gpu_ids()
@@ -960,8 +966,8 @@ class ActorsWithGPUs(unittest.TestCase):
     results = []
     for _ in range(5):
       results.append(f.remote())
-      a = Actor()
-      results.append(a.get_gpu_id())
+      a = Actor.remote()
+      results.append(a.get_gpu_id.remote())
 
     gpu_ids = ray.get(results)
     self.assertEqual(set(gpu_ids), set(range(10)))
diff --git a/test/failure_test.py b/test/failure_test.py
index f213d26ab..3b605b9a2 100644
--- a/test/failure_test.py
+++ b/test/failure_test.py
@@ -209,7 +209,7 @@ def temporary_helper_function():
 
     # Define an actor that closes over this temporary module. This should fail
     # when it is unpickled.
-    @ray.actor
+    @ray.remote
     class Foo(object):
       def __init__(self):
         self.x = module.temporary_python_file()
@@ -221,7 +221,7 @@ def temporary_helper_function():
     self.assertEqual(len(ray.error_info()), 0)
 
     # Create an actor.
-    foo = Foo()
+    foo = Foo.remote()
 
     # Wait for the error to arrive.
     wait_for_errors(b"register_actor", 1)
@@ -235,7 +235,7 @@ def temporary_helper_function():
     # Check that if we try to get the function it throws an exception and does
     # not hang.
     with self.assertRaises(Exception):
-      ray.get(foo.get_val())
+      ray.get(foo.get_val.remote())
 
     # Wait for the error from when the call to get_val.
     wait_for_errors(b"task", 2)
@@ -257,7 +257,7 @@ class ActorTest(unittest.TestCase):
     error_message1 = "actor constructor failed"
     error_message2 = "actor method failed"
 
-    @ray.actor
+    @ray.remote
     class FailedActor(object):
       def __init__(self):
         raise Exception(error_message1)
@@ -268,7 +268,7 @@ class ActorTest(unittest.TestCase):
       def fail_method(self):
         raise Exception(error_message2)
 
-    a = FailedActor()
+    a = FailedActor.remote()
 
     # Make sure that we get errors from a failed constructor.
     wait_for_errors(b"task", 1)
@@ -277,7 +277,7 @@ class ActorTest(unittest.TestCase):
                   ray.error_info()[0][b"message"].decode("ascii"))
 
     # Make sure that we get errors from a failed method.
-    a.fail_method()
+    a.fail_method.remote()
     wait_for_errors(b"task", 2)
     self.assertEqual(len(ray.error_info()), 2)
     self.assertIn(error_message2,
@@ -288,7 +288,7 @@ class ActorTest(unittest.TestCase):
   def testIncorrectMethodCalls(self):
     ray.init(num_workers=0, driver_mode=ray.SILENT_MODE)
 
-    @ray.actor
+    @ray.remote
     class Actor(object):
       def __init__(self, missing_variable_name):
         pass
@@ -300,25 +300,27 @@ class ActorTest(unittest.TestCase):
 
     # Create an actor with too few arguments.
     with self.assertRaises(Exception):
-      a = Actor()
+      a = Actor.remote()
 
     # Create an actor with too many arguments.
     with self.assertRaises(Exception):
-      a = Actor(1, 2)
+      a = Actor.remote(1, 2)
 
     # Create an actor the correct number of arguments.
-    a = Actor(1)
+    a = Actor.remote(1)
 
     # Call a method with too few arguments.
     with self.assertRaises(Exception):
-      a.get_val()
+      a.get_val.remote()
 
     # Call a method with too many arguments.
     with self.assertRaises(Exception):
-      a.get_val(1, 2)
+      a.get_val.remote(1, 2)
     # Call a method that doesn't exist.
     with self.assertRaises(AttributeError):
       a.nonexistent_method()
+    with self.assertRaises(AttributeError):
+      a.nonexistent_method.remote()
 
     ray.worker.cleanup()
 
diff --git a/test/jenkins_tests/multi_node_tests/many_drivers_test.py b/test/jenkins_tests/multi_node_tests/many_drivers_test.py
index 09ef13ecd..33218ad35 100644
--- a/test/jenkins_tests/multi_node_tests/many_drivers_test.py
+++ b/test/jenkins_tests/multi_node_tests/many_drivers_test.py
@@ -20,7 +20,7 @@ max_concurrent_drivers = 15
 num_gpus_per_driver = 5
 
 
-@ray.actor(num_gpus=1)
+@ray.remote(num_gpus=1)
 class Actor1(object):
   def __init__(self):
     assert len(ray.get_gpu_ids()) == 1
@@ -50,7 +50,7 @@ def driver(redis_address, driver_index):
     start_time = time.time()
     while time.time() - start_time < timeout:
       try:
-        actor = actor_class()
+        actor = actor_class.remote()
       except Exception as e:
         time.sleep(0.1)
       else:
@@ -64,7 +64,7 @@ def driver(redis_address, driver_index):
     actors_one_gpu.append(try_to_create_actor(Actor1))
 
   for _ in range(100):
-    ray.get([actor.check_ids() for actor in actors_one_gpu])
+    ray.get([actor.check_ids.remote() for actor in actors_one_gpu])
 
   _broadcast_event("DRIVER_{}_DONE".format(driver_index), redis_address)
 
diff --git a/test/jenkins_tests/multi_node_tests/remove_driver_test.py b/test/jenkins_tests/multi_node_tests/remove_driver_test.py
index 2e71224d1..5d0b7d272 100644
--- a/test/jenkins_tests/multi_node_tests/remove_driver_test.py
+++ b/test/jenkins_tests/multi_node_tests/remove_driver_test.py
@@ -60,7 +60,7 @@ def long_running_task(driver_index, task_index, redis_address):
 num_long_running_tasks_per_driver = 2
 
 
-@ray.actor
+@ray.remote
 class Actor0(object):
   def __init__(self, driver_index, actor_index, redis_address):
     _broadcast_event(actor_event_name(driver_index, actor_index),
@@ -77,7 +77,7 @@ class Actor0(object):
       time.sleep(100)
 
 
-@ray.actor(num_gpus=1)
+@ray.remote(num_gpus=1)
 class Actor1(object):
   def __init__(self, driver_index, actor_index, redis_address):
     _broadcast_event(actor_event_name(driver_index, actor_index),
@@ -94,7 +94,7 @@ class Actor1(object):
       time.sleep(100)
 
 
-@ray.actor(num_gpus=2)
+@ray.remote(num_gpus=2)
 class Actor2(object):
   def __init__(self, driver_index, actor_index, redis_address):
     _broadcast_event(actor_event_name(driver_index, actor_index),
@@ -128,18 +128,19 @@ def driver_0(redis_address, driver_index):
     long_running_task.remote(driver_index, i, redis_address)
 
   # Create some actors that require one GPU.
-  actors_one_gpu = [Actor1(driver_index, i, redis_address) for i in range(5)]
+  actors_one_gpu = [Actor1.remote(driver_index, i, redis_address)
+                    for i in range(5)]
   # Create some actors that don't require any GPUs.
-  actors_no_gpus = [Actor0(driver_index, 5 + i, redis_address)
+  actors_no_gpus = [Actor0.remote(driver_index, 5 + i, redis_address)
                     for i in range(5)]
 
   for _ in range(1000):
-    ray.get([actor.check_ids() for actor in actors_one_gpu])
-    ray.get([actor.check_ids() for actor in actors_no_gpus])
+    ray.get([actor.check_ids.remote() for actor in actors_one_gpu])
+    ray.get([actor.check_ids.remote() for actor in actors_no_gpus])
 
   # Start a long-running method on one actor and make sure this doesn't affect
   # anything.
-  actors_no_gpus[0].long_running_method()
+  actors_no_gpus[0].long_running_method.remote()
 
   _broadcast_event("DRIVER_0_DONE", redis_address)
 
@@ -162,22 +163,23 @@ def driver_1(redis_address, driver_index):
     long_running_task.remote(driver_index, i, redis_address)
 
   # Create an actor that requires two GPUs.
-  actors_two_gpus = [Actor2(driver_index, i, redis_address) for i in range(1)]
+  actors_two_gpus = [Actor2.remote(driver_index, i, redis_address)
+                     for i in range(1)]
   # Create some actors that require one GPU.
-  actors_one_gpu = [Actor1(driver_index, 1 + i, redis_address)
+  actors_one_gpu = [Actor1.remote(driver_index, 1 + i, redis_address)
                     for i in range(3)]
   # Create some actors that don't require any GPUs.
-  actors_no_gpus = [Actor0(driver_index, 1 + 3 + i, redis_address)
+  actors_no_gpus = [Actor0.remote(driver_index, 1 + 3 + i, redis_address)
                     for i in range(5)]
 
   for _ in range(1000):
-    ray.get([actor.check_ids() for actor in actors_two_gpus])
-    ray.get([actor.check_ids() for actor in actors_one_gpu])
-    ray.get([actor.check_ids() for actor in actors_no_gpus])
+    ray.get([actor.check_ids.remote() for actor in actors_two_gpus])
+    ray.get([actor.check_ids.remote() for actor in actors_one_gpu])
+    ray.get([actor.check_ids.remote() for actor in actors_no_gpus])
 
   # Start a long-running method on one actor and make sure this doesn't affect
   # anything.
-  actors_one_gpu[0].long_running_method()
+  actors_one_gpu[0].long_running_method.remote()
 
   _broadcast_event("DRIVER_1_DONE", redis_address)
 
@@ -195,7 +197,7 @@ def cleanup_driver(redis_address, driver_index):
     # We go ahead and create some actors that don't require any GPUs. We don't
     # need to wait for the other drivers to finish. We call methods on these
     # actors later to make sure they haven't been killed.
-    actors_no_gpus = [Actor0(driver_index, i, redis_address)
+    actors_no_gpus = [Actor0.remote(driver_index, i, redis_address)
                       for i in range(10)]
 
   _wait_for_event("DRIVER_0_DONE", redis_address)
@@ -207,7 +209,7 @@ def cleanup_driver(redis_address, driver_index):
     start_time = time.time()
     while time.time() - start_time < timeout:
       try:
-        actor = actor_class(driver_index, actor_index, redis_address)
+        actor = actor_class.remote(driver_index, actor_index, redis_address)
       except Exception as e:
         time.sleep(0.1)
       else:
@@ -271,9 +273,9 @@ def cleanup_driver(redis_address, driver_index):
   # Only one of the cleanup drivers should create and use more actors.
   if driver_index == 2:
     for _ in range(1000):
-      ray.get([actor.check_ids() for actor in actors_two_gpus])
-      ray.get([actor.check_ids() for actor in actors_one_gpu])
-      ray.get([actor.check_ids() for actor in actors_no_gpus])
+      ray.get([actor.check_ids.remote() for actor in actors_two_gpus])
+      ray.get([actor.check_ids.remote() for actor in actors_one_gpu])
+      ray.get([actor.check_ids.remote() for actor in actors_no_gpus])
 
   _broadcast_event("DRIVER_{}_DONE".format(driver_index), redis_address)
 
diff --git a/test/tensorflow_test.py b/test/tensorflow_test.py
index 3cf101247..55d7dfe27 100644
--- a/test/tensorflow_test.py
+++ b/test/tensorflow_test.py
@@ -166,10 +166,11 @@ class TensorFlowTest(unittest.TestCase):
     ray.experimental.TensorFlowVariables(loss1, sess1)
     sess1.run(init1)
 
-    net2 = ray.actor(NetActor)()
-    weights2 = ray.get(net2.get_weights())
+    net2 = ray.remote(NetActor).remote()
+    weights2 = ray.get(net2.get_weights.remote())
 
-    new_weights2 = ray.get(net2.set_and_get_weights(net2.get_weights()))
+    new_weights2 = ray.get(net2.set_and_get_weights.remote(
+        net2.get_weights.remote()))
     self.assertEqual(weights2, new_weights2)
 
     ray.worker.cleanup()
@@ -193,23 +194,24 @@ class TensorFlowTest(unittest.TestCase):
   def testRemoteTrainingStep(self):
     ray.init(num_workers=1)
 
-    net = ray.actor(TrainActor)()
-    ray.get(net.training_step(net.get_weights()))
+    net = ray.remote(TrainActor).remote()
+    ray.get(net.training_step.remote(net.get_weights.remote()))
 
     ray.worker.cleanup()
 
   def testRemoteTrainingLoss(self):
     ray.init(num_workers=2)
 
-    net = ray.actor(TrainActor)()
+    net = ray.remote(TrainActor).remote()
     loss, variables, _, sess, grads, train, placeholders = TrainActor().values
 
     before_acc = sess.run(loss, feed_dict=dict(zip(placeholders,
                                                    [[2] * 100, [4] * 100])))
 
     for _ in range(3):
-      gradients_list = ray.get([net.training_step(variables.get_weights())
-                                for _ in range(2)])
+      gradients_list = ray.get(
+          [net.training_step.remote(variables.get_weights())
+           for _ in range(2)])
       mean_grads = [sum([gradients[i] for gradients in gradients_list]) /
                     len(gradients_list) for i in range(len(gradients_list[0]))]
       feed_dict = {grad[0]: mean_grad for (grad, mean_grad)