Repository: proxystore/aeris

Issue:
Create a `Manager` class to reduce boilerplate
### Describe the Request

Right now, apps have a lot of boilerplate setup and teardown: (1) create exchange, (2) create launcher, (3) launch agents, (4) bind handles to agents, (5) shutdown agents, (6) close handles, (7) close launcher, and (8) close exchange. This steps mostly need to be done in this order or things will hang.

A `Manager` class, that owns the exchange and launcher, can improve this workflow. The `Manager.launch()` method can inject the exchange (re: #24), bind the handle for the client, and hold references to handles for clean up at the end. On clean up, the manager can shutdown all alive agents, cleanup handles, and close the launcher/exchange.

Having a single `Manager` class also means that we can bind handles to the manager rather than to individual clients, hopefully reducing the number of mailboxes/threads in the main script. This will look very similar to how handles are bound to an agent (e.g., a `.bind_to_manager` method), and needs some mechanisms in the manager to map messages back to the correct handle (maybe by ID?).

### Sample Code

_No response_

Discussion:

Solution patch:
diff --git a/aeris/agent.py b/aeris/agent.py
index dbf9d1c..ef371c0 100644
--- a/aeris/agent.py
+++ b/aeris/agent.py
@@ -7,25 +7,23 @@
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 from typing import Generic
-from typing import get_args
 from typing import TypeVar
 
 from aeris.behavior import Behavior
 from aeris.exception import BadIdentifierError
 from aeris.exception import MailboxClosedError
 from aeris.exchange import Exchange
-from aeris.handle import AgentRemoteHandle
+from aeris.handle import BoundRemoteHandle
 from aeris.handle import ClientRemoteHandle
 from aeris.handle import ProxyHandle
 from aeris.handle import UnboundRemoteHandle
 from aeris.identifier import AgentIdentifier
-from aeris.identifier import Identifier
 from aeris.message import ActionRequest
-from aeris.message import Message
 from aeris.message import PingRequest
 from aeris.message import RequestMessage
 from aeris.message import ResponseMessage
 from aeris.message import ShutdownRequest
+from aeris.multiplex import MailboxMultiplexer
 
 logger = logging.getLogger(__name__)
 
@@ -44,13 +42,13 @@ class _AgentState(enum.Enum):
 # of the Agent constructor.
 def _agent_trampoline(
     behavior: BehaviorT,
-    aid: AgentIdentifier,
+    agent_id: AgentIdentifier,
     exchange: Exchange,
     close_exchange: bool,
 ) -> Agent[BehaviorT]:
     return Agent(
         behavior,
-        aid=aid,
+        agent_id=agent_id,
         exchange=exchange,
         close_exchange=close_exchange,
     )
@@ -69,7 +67,7 @@ class Agent(Generic[BehaviorT]):
 
     Args:
         behavior: Behavior that the agent will exhibit.
-        aid: Identifier of this agent in a multi-agent system.
+        agent_id: Identifier of this agent in a multi-agent system.
         exchange: Message exchange of multi-agent system. The agent will close
             the exchange when it finished running.
         close_exchange: Close the `exchange` object when the agent is
@@ -80,11 +78,11 @@ def __init__(
         self,
         behavior: BehaviorT,
         *,
-        aid: AgentIdentifier,
+        agent_id: AgentIdentifier,
         exchange: Exchange,
         close_exchange: bool = False,
     ) -> None:
-        self.aid = aid
+        self.agent_id = agent_id
         self.behavior = behavior
         self.exchange = exchange
         self.close_exchange = close_exchange
@@ -100,10 +98,12 @@ def __init__(
         self._action_futures: dict[ActionRequest, Future[None]] = {}
         self._loop_pool: ThreadPoolExecutor | None = None
         self._loop_futures: set[Future[None]] = set()
-        # The key in bound_handles is the identifier of the remote agent
-        # the handle targets. Each running agent should only have a single
-        # handle to any given remote agent.
-        self._bound_handles: dict[Identifier, AgentRemoteHandle[Any]] = {}
+
+        self._multiplexer = MailboxMultiplexer(
+            self.agent_id,
+            self.exchange,
+            request_handler=self._request_handler,
+        )
 
     def __call__(self) -> None:
         """Alias for [run()][aeris.agent.Agent.run]."""
@@ -112,41 +112,34 @@ def __call__(self) -> None:
     def __repr__(self) -> str:
         name = type(self).__name__
         return (
-            f'{name}(aid={self.aid!r}, behavior={self.behavior!r}, '
+            f'{name}(agent_id={self.agent_id!r}, behavior={self.behavior!r}, '
             f'exchange={self.exchange!r})'
         )
 
     def __str__(self) -> str:
         name = type(self).__name__
         behavior = type(self.behavior).__name__
-        return f'{name}<{behavior}; {self.aid}>'
+        return f'{name}<{behavior}; {self.agent_id}>'
 
     def __reduce__(self) -> Any:
         return (
             _agent_trampoline,
-            (self.behavior, self.aid, self.exchange, self.close_exchange),
+            (self.behavior, self.agent_id, self.exchange, self.close_exchange),
         )
 
     def _bind_handle(
         self,
         attr: str,
-        handle: AgentRemoteHandle[Any] | UnboundRemoteHandle[Any],
+        handle: BoundRemoteHandle[Any] | UnboundRemoteHandle[Any],
     ) -> None:
-        if handle.aid in self._bound_handles:
-            raise RuntimeError(
-                f'{self} already has a handle bound to a remote agent with '
-                f'{handle.aid}. The duplicate handle should be removed from '
-                'the behavior instance.',
-            )
-        bound = handle.bind_to_agent(self.aid)
+        bound = self._multiplexer.bind(handle)
         # Replace the handle attribute on the behavior with a handle bound
         # to this agent.
         setattr(self.behavior, attr, bound)
-        self._bound_handles[bound.aid] = bound
         logger.debug(
             'Bound remote handle to %s to running agent with %s',
-            handle.aid,
-            self.aid,
+            handle.agent_id,
+            self.agent_id,
         )
 
     def _bind_handles(self) -> None:
@@ -159,8 +152,8 @@ def _bind_handles(self) -> None:
                 pass
             elif isinstance(handle, UnboundRemoteHandle):
                 self._bind_handle(attr, handle)
-            elif isinstance(handle, AgentRemoteHandle):
-                if handle.hid != self.aid:
+            elif isinstance(handle, BoundRemoteHandle):
+                if handle.mailbox_id != self.agent_id:
                     self._bind_handle(attr, handle)
             else:
                 raise AssertionError('Unreachable.')
@@ -173,7 +166,7 @@ def _send_response(self, response: ResponseMessage) -> None:
                 'Failed to send response from %s to %s. '
                 'This likely means the destination mailbox was '
                 'removed from the exchange.',
-                self.aid,
+                self.agent_id,
                 response.dest,
             )
 
@@ -186,18 +179,6 @@ def _execute_action(self, request: ActionRequest) -> None:
             response = request.response(result=result)
         self._send_response(response)
 
-    def _response_handler(self, response: ResponseMessage) -> None:
-        try:
-            handle = self._bound_handles[response.src]
-        except KeyError:
-            logger.exception(
-                'Receieved a response message from %s but no handle to '
-                'that agent is bound to this agent.',
-                response.src,
-            )
-        else:
-            handle._process_response(response)
-
     def _request_handler(self, request: RequestMessage) -> None:
         if isinstance(request, ActionRequest):
             # The _request_handler should only be called within the message
@@ -210,32 +191,13 @@ def _request_handler(self, request: RequestMessage) -> None:
                 lambda _: self._action_futures.pop(request),
             )
         elif isinstance(request, PingRequest):
-            logger.info('Ping request received by %s', self.aid)
+            logger.info('Ping request received by %s', self.agent_id)
             self._send_response(request.response())
         elif isinstance(request, ShutdownRequest):
             self.signal_shutdown()
         else:
             raise AssertionError('Unreachable.')
 
-    def _message_handler(self, message: Message) -> None:
-        if isinstance(message, get_args(ResponseMessage)):
-            self._response_handler(message)
-        elif isinstance(message, get_args(RequestMessage)):
-            self._request_handler(message)
-        else:
-            raise AssertionError('Unreachable.')
-
-    def _message_listener(self) -> None:
-        logger.info('Message listener started for %s', self.aid)
-
-        while True:
-            try:
-                message = self.exchange.recv(self.aid)
-            except MailboxClosedError:
-                break
-            else:
-                self._message_handler(message)
-
     def action(self, action: str, args: Any, kwargs: Any) -> Any:
         """Invoke an action of the agent.
 
@@ -251,7 +213,7 @@ def action(self, action: str, args: Any, kwargs: Any) -> Any:
             AttributeError: if an action with this name is not implemented by
                 the behavior of the agent.
         """
-        logger.debug('Invoking "%s" action on %s', action, self.aid)
+        logger.debug('Invoking "%s" action on %s', action, self.agent_id)
         if action not in self._actions:
             raise AttributeError(
                 f'Agent[{type(self.behavior).__name__}] does not have an '
@@ -310,12 +272,12 @@ def start(self) -> None:
                 loop_future = self._loop_pool.submit(method, self._shutdown)
                 self._loop_futures.add(loop_future)
 
-            listener_future = self._loop_pool.submit(self._message_listener)
+            listener_future = self._loop_pool.submit(self._multiplexer.listen)
             self._loop_futures.add(listener_future)
 
             self._state = _AgentState.RUNNING
 
-        logger.info('Started agent with %s', self.aid)
+        logger.info('Started agent with %s', self.agent_id)
 
     def shutdown(self) -> None:
         """Shutdown the agent.
@@ -338,7 +300,7 @@ def shutdown(self) -> None:
         if self._state is _AgentState.SHUTDOWN:
             return
 
-        logger.info('Shutdown requested for %s', self.aid)
+        logger.info('Shutdown requested for %s', self.agent_id)
         with self._state_lock:
             self._state = _AgentState.TERMINTATING
             self._shutdown.set()
@@ -347,8 +309,9 @@ def shutdown(self) -> None:
             if self._action_pool is not None:
                 self._action_pool.shutdown(wait=True, cancel_futures=True)
 
-            # Cause the message listener thread to exit by closing the mailbox.
-            self.exchange.close_mailbox(self.aid)
+            # Cause the multiplexer message listener thread to exit by closing
+            # the mailbox the multiplexer is listening to.
+            self._multiplexer.close_mailbox()
 
             # Wait on all the loops, raising any exceptions.
             for future in self._loop_futures:
@@ -362,7 +325,7 @@ def shutdown(self) -> None:
             self.behavior.shutdown()
             self._state = _AgentState.SHUTDOWN
 
-        logger.info('Shutdown agent with %s', self.aid)
+        logger.info('Shutdown agent with %s', self.agent_id)
 
     def signal_shutdown(self) -> None:
         """Signal that the agent should exit.
diff --git a/aeris/exchange/simple.py b/aeris/exchange/simple.py
index 12dce6d..814df02 100644
--- a/aeris/exchange/simple.py
+++ b/aeris/exchange/simple.py
@@ -90,7 +90,7 @@ class _BaseExchangeMessage(BaseModel):
         validate_default=True,
     )
 
-    mid: uuid.UUID = Field(default_factory=uuid.uuid4)
+    tag: uuid.UUID = Field(default_factory=uuid.uuid4)
     kind: _ExchangeMessageType = Field()
     src: Identifier = Field()
     dest: Optional[Identifier] = Field(None)  # noqa: UP007
@@ -116,7 +116,7 @@ def response(
         payload: Message | None = None,
     ) -> _ExchangeResponseMessage:
         return _ExchangeResponseMessage(
-            mid=self.mid,
+            tag=self.tag,
             kind=self.kind,
             src=self.src,
             dest=self.dest,
@@ -216,10 +216,10 @@ def _handle_message(
                 message,
             )
             if message.success:
-                self._pending[message.mid].set_result(message)
+                self._pending[message.tag].set_result(message)
             else:
                 assert message.error is not None
-                self._pending[message.mid].set_exception(message.error)
+                self._pending[message.tag].set_exception(message.error)
         else:
             logger.warning(
                 'Dropping bad message from server with type %s (exchange=%s)',
@@ -277,7 +277,7 @@ def _send_request(
         request: _ExchangeRequestMessage,
     ) -> _ExchangeResponseMessage:
         future: Future[_ExchangeResponseMessage] = Future()
-        self._pending[request.mid] = future
+        self._pending[request.tag] = future
         self._socket.send(request.model_serialize() + b'\n')
         logger.debug(
             'Sent message to server (exchange=%s, message=%r)',
@@ -285,7 +285,7 @@ def _send_request(
             request,
         )
         response = future.result(timeout=self.timeout)
-        del self._pending[request.mid]
+        del self._pending[request.tag]
         return response
 
     def close(self) -> None:
diff --git a/aeris/handle.py b/aeris/handle.py
index 2e4af3a..57843f1 100644
--- a/aeris/handle.py
+++ b/aeris/handle.py
@@ -131,17 +131,16 @@ class RemoteHandle(Generic[BehaviorT_co], abc.ABC):
 
     This is an abstract base class with three possible concrete types
     representing the three remote handle states: unbound, client bound, or
-    agent bound.
+    mailbox bound.
 
-    An unbound handle does not have an identifier for itself which means it
+    An unbound handle does not have a mailbox for itself which means it
     cannot send messages to the remote agent because the handle does not
     know the return address to use. Thus, unbound handles need to be bound
-    with `bind_as_client` or `bind_to_agent` before they can be used.
+    with `bind_as_client` or `bind_to_mailbox` before they can be used.
 
     A client bound handle has a unique handle identifier and mailbox for
-    itself. An agent bound handle shares the identifier and mailbox of the
-    running agent it is bound to (this is a different agent to the target
-    agent of the handle).
+    itself. A mailbox bound handle shares its identifier and mailbox with
+    another entity (e.g., a running agent).
 
     Note:
         When an instance is pickled and unpickled, such as when
@@ -151,21 +150,23 @@ class RemoteHandle(Generic[BehaviorT_co], abc.ABC):
 
     Args:
         exchange: Message exchange used for agent communication.
-        aid: Identifier of the target agent of this handle.
-        hid: Identifier of this handle. If unbound, this is `None`. If bound
-            as a client, this is a `ClientIdentifier`. If bound to a running
-            agent, this is the ID of that agent.
+        agent_id: Identifier of the target agent of this handle.
+        mailbox_id: Identifier of the mailbox this handle receives messages to.
+            If unbound, this is `None`.
     """
 
     def __init__(
         self,
         exchange: Exchange,
-        aid: AgentIdentifier,
-        hid: Identifier | None = None,
+        agent_id: AgentIdentifier,
+        mailbox_id: Identifier | None = None,
     ) -> None:
         self.exchange = exchange
-        self.aid = aid
-        self.hid = hid
+        self.agent_id = agent_id
+        self.mailbox_id = mailbox_id
+        # Unique identifier for each handle object; used to disambiguate
+        # messages when multiple handles are bound to the same mailbox.
+        self.handle_id = uuid.uuid4()
 
         self._futures: dict[uuid.UUID, Future[Any]] = {}
         self._closed = False
@@ -187,21 +188,21 @@ def __reduce__(
         type[UnboundRemoteHandle[Any]],
         tuple[Exchange, AgentIdentifier],
     ]:
-        return (UnboundRemoteHandle, (self.exchange, self.aid))
+        return (UnboundRemoteHandle, (self.exchange, self.agent_id))
 
     def __repr__(self) -> str:
         return (
-            f'{type(self).__name__}(aid={self.aid!r}, hid={self.hid!r}, '
-            f'exchange={self.exchange!r})'
+            f'{type(self).__name__}(agent_id={self.agent_id!r}, '
+            f'mailbox_id={self.mailbox_id!r}, exchange={self.exchange!r})'
         )
 
     def __str__(self) -> str:
         name = type(self).__name__
-        return f'{name}<{self.aid}; {self.hid}>'
+        return f'{name}<agent: {self.agent_id}; mailbox: {self.mailbox_id}>'
 
     def _process_response(self, response: ResponseMessage) -> None:
         if isinstance(response, (ActionResponse, PingResponse)):
-            future = self._futures.pop(response.mid)
+            future = self._futures.pop(response.tag)
             if response.exception is not None:
                 future.set_exception(response.exception)
             elif isinstance(response, ActionResponse):
@@ -222,7 +223,7 @@ def _send_request(self, request: RequestMessage) -> None:
     @abc.abstractmethod
     def bind_as_client(
         self,
-        cid: ClientIdentifier | None = None,
+        client_id: ClientIdentifier | None = None,
     ) -> ClientRemoteHandle[BehaviorT_co]:
         """Bind the handle as a unique client in the exchange.
 
@@ -231,8 +232,8 @@ def bind_as_client(
             different semantics.
 
         Args:
-            cid: Client identifier to be used by this handle. If `None`, a
-                new identifier will be created using the exchange.
+            client_id: Client identifier to be used by this handle. If `None`,
+                a new identifier will be created using the exchange.
 
         Returns:
             Remote handle bound to the client identifier.
@@ -240,21 +241,17 @@ def bind_as_client(
         ...
 
     @abc.abstractmethod
-    def bind_to_agent(
+    def bind_to_mailbox(
         self,
-        aid: AgentIdentifier,
-    ) -> AgentRemoteHandle[BehaviorT_co]:
-        """Bind the handle to a running agent.
-
-        Note:
-            This is an abstract method. Each remote handle variant implements
-            different semantics.
+        mailbox_id: Identifier,
+    ) -> BoundRemoteHandle[BehaviorT_co]:
+        """Bind the handle to an existing mailbox.
 
         Args:
-            aid: Identifier of the running agent to bind to.
+            mailbox_id: Identifier of the mailbox to bind to.
 
         Returns:
-            Remote handle bound to the agent identifier.
+            Remote handle bound to the identifier.
         """
         ...
 
@@ -303,29 +300,30 @@ def action(
         Raises:
             HandleClosedError: if the handle was closed.
         """
-        if self.hid is None:
+        if self.mailbox_id is None:
             # UnboundRemoteHandle overrides these methods and is the only
             # handle state variant where hid is None.
             raise AssertionError(
                 'Method should not be reachable in unbound state.',
             )
         if self._closed:
-            raise HandleClosedError(self.aid, self.hid)
+            raise HandleClosedError(self.agent_id, self.mailbox_id)
 
         request = ActionRequest(
-            src=self.hid,
-            dest=self.aid,
+            src=self.mailbox_id,
+            dest=self.agent_id,
+            label=self.handle_id,
             action=action,
             args=args,
             kwargs=kwargs,
         )
         future: Future[R] = Future()
-        self._futures[request.mid] = future
+        self._futures[request.tag] = future
         self._send_request(request)
         logger.debug(
             'Sent action request from %s to %s (action=%r)',
-            self.hid,
-            self.aid,
+            self.mailbox_id,
+            self.agent_id,
             action,
         )
         return future
@@ -347,27 +345,31 @@ def ping(self, *, timeout: float | None = None) -> float:
             HandleClosedError: if the handle was closed.
             TimeoutError: if the timeout is exceeded.
         """
-        if self.hid is None:
+        if self.mailbox_id is None:
             # UnboundRemoteHandle overrides these methods and is the only
             # handle state variant where hid is None.
             raise AssertionError(
                 'Method should not be reachable in unbound state.',
             )
         if self._closed:
-            raise HandleClosedError(self.aid, self.hid)
+            raise HandleClosedError(self.agent_id, self.mailbox_id)
 
         start = time.perf_counter()
-        request = PingRequest(src=self.hid, dest=self.aid)
+        request = PingRequest(
+            src=self.mailbox_id,
+            dest=self.agent_id,
+            label=self.handle_id,
+        )
         future: Future[None] = Future()
-        self._futures[request.mid] = future
+        self._futures[request.tag] = future
         self._send_request(request)
-        logger.debug('Sent ping from %s to %s', self.hid, self.aid)
+        logger.debug('Sent ping from %s to %s', self.mailbox_id, self.agent_id)
         future.result(timeout=timeout)
         elapsed = time.perf_counter() - start
         logger.debug(
             'Received ping from %s to %s in %.3f ms',
-            self.hid,
-            self.aid,
+            self.mailbox_id,
+            self.agent_id,
             elapsed / 1000,
         )
         return elapsed
@@ -380,18 +382,26 @@ def shutdown(self) -> None:
         Raises:
             HandleClosedError: if the handle was closed.
         """
-        if self.hid is None:
+        if self.mailbox_id is None:
             # UnboundRemoteHandle overrides these methods and is the only
             # handle state variant where hid is None.
             raise AssertionError(
                 'Method should not be reachable in unbound state.',
             )
         if self._closed:
-            raise HandleClosedError(self.aid, self.hid)
+            raise HandleClosedError(self.agent_id, self.mailbox_id)
 
-        request = ShutdownRequest(src=self.hid, dest=self.aid)
+        request = ShutdownRequest(
+            src=self.mailbox_id,
+            dest=self.agent_id,
+            label=self.handle_id,
+        )
         self._send_request(request)
-        logger.debug('Sent shutdown request from %s to %s', self.hid, self.aid)
+        logger.debug(
+            'Sent shutdown request from %s to %s',
+            self.mailbox_id,
+            self.agent_id,
+        )
 
 
 class UnboundRemoteHandle(RemoteHandle[BehaviorT_co]):
@@ -403,51 +413,53 @@ class UnboundRemoteHandle(RemoteHandle[BehaviorT_co]):
         to the remote agent.
 
     Args:
-        aid: Identifier of the agent.
         exchange: Message exchange used for agent communication.
+        agent_id: Identifier of the agent.
     """
 
-    def __init__(self, exchange: Exchange, aid: AgentIdentifier) -> None:
-        super().__init__(exchange, aid)
+    def __init__(self, exchange: Exchange, agent_id: AgentIdentifier) -> None:
+        super().__init__(exchange, agent_id=agent_id)
 
     def __repr__(self) -> str:
         name = type(self).__name__
-        return f'{name}(aid={self.aid!r}, exchange={self.exchange!r})'
+        return (
+            f'{name}(agent_id={self.agent_id!r}, exchange={self.exchange!r})'
+        )
 
     def __str__(self) -> str:
-        return f'{type(self).__name__}<{self.aid}>'
+        return f'{type(self).__name__}<agent: {self.agent_id}>'
 
     def _send_request(self, request: RequestMessage) -> None:
-        raise HandleNotBoundError(self.aid)
+        raise HandleNotBoundError(self.agent_id)
 
     def bind_as_client(
         self,
-        cid: ClientIdentifier | None = None,
+        client_id: ClientIdentifier | None = None,
     ) -> ClientRemoteHandle[BehaviorT_co]:
         """Bind the handle as a unique client in the exchange.
 
         Args:
-            cid: Client identifier to be used by this handle. If `None`, a
-                new identifier will be created using the exchange.
+            client_id: Client identifier to be used by this handle. If `None`,
+                a new identifier will be created using the exchange.
 
         Returns:
             Remote handle bound to the client identifier.
         """
-        return ClientRemoteHandle(self.exchange, self.aid, cid)
+        return ClientRemoteHandle(self.exchange, self.agent_id, client_id)
 
-    def bind_to_agent(
+    def bind_to_mailbox(
         self,
-        aid: AgentIdentifier,
-    ) -> AgentRemoteHandle[BehaviorT_co]:
-        """Bind the handle to a running agent.
+        mailbox_id: Identifier,
+    ) -> BoundRemoteHandle[BehaviorT_co]:
+        """Bind the handle to an existing mailbox.
 
         Args:
-            aid: Identifier of the running agent to bind to.
+            mailbox_id: Identifier of the mailbox to bind to.
 
         Returns:
-            Remote handle bound to the agent identifier.
+            Remote handle bound to the identifier.
         """
-        return AgentRemoteHandle(self.exchange, self.aid, aid)
+        return BoundRemoteHandle(self.exchange, self.agent_id, mailbox_id)
 
     def action(
         self,
@@ -457,78 +469,71 @@ def action(
         **kwargs: Any,
     ) -> Future[R]:
         """Raises [`HandleNotBoundError`][aeris.exception.HandleNotBoundError]."""  # noqa: E501
-        raise HandleNotBoundError(self.aid)
+        raise HandleNotBoundError(self.agent_id)
 
     def ping(self, *, timeout: float | None = None) -> float:
         """Raises [`HandleNotBoundError`][aeris.exception.HandleNotBoundError]."""  # noqa: E501
-        raise HandleNotBoundError(self.aid)
+        raise HandleNotBoundError(self.agent_id)
 
     def shutdown(self) -> None:
         """Raises [`HandleNotBoundError`][aeris.exception.HandleNotBoundError]."""  # noqa: E501
-        raise HandleNotBoundError(self.aid)
+        raise HandleNotBoundError(self.agent_id)
 
 
-class AgentRemoteHandle(RemoteHandle[BehaviorT_co]):
-    """Handle to a remote agent bound to a running agent.
-
-    Note:
-        When a handle is bound to a running agent, the running agent and the
-        handle share a mailbox. Thus, the running agent is responsible for
-        forwarding response messages it receives back to the appropriate
-        bound handle.
+class BoundRemoteHandle(RemoteHandle[BehaviorT_co]):
+    """Handle to a remote agent bound to an existing mailbox.
 
     Args:
         exchange: Message exchange used for agent communication.
-        aid: Identifier of the target agent of this handle.
-        hid: Identifier of the running agent that this handle is bound to.
+        agent_id: Identifier of the target agent of this handle.
+        mailbox_id: Identifier of the mailbox this handle receives messages to.
     """
 
     def __init__(
         self,
         exchange: Exchange,
-        aid: AgentIdentifier,
-        hid: AgentIdentifier,
+        agent_id: AgentIdentifier,
+        mailbox_id: Identifier,
     ) -> None:
-        if aid == hid:
+        if agent_id == mailbox_id:
             raise ValueError(
-                f'Cannot create handle to {aid} that is bound to itself. '
-                'Check that the values of `aid` and `hid` are different.',
+                f'Cannot create handle to {agent_id} that is bound to itself. '
+                'Check that the values of `agent_id` and `mailbox_id` '
+                'are different.',
             )
-        super().__init__(exchange, aid)
-        self.hid = hid
+        super().__init__(exchange, agent_id, mailbox_id)
 
     def bind_as_client(
         self,
-        cid: ClientIdentifier | None = None,
+        client_id: ClientIdentifier | None = None,
     ) -> ClientRemoteHandle[BehaviorT_co]:
         """Bind the handle as a unique client in the exchange.
 
         Args:
-            cid: Client identifier to be used by this handle. If `None`, a
-                new identifier will be created using the exchange.
+            client_id: Client identifier to be used by this handle. If `None`,
+                a new identifier will be created using the exchange.
 
         Returns:
             Remote handle bound to the client identifier.
         """
-        return ClientRemoteHandle(self.exchange, self.aid, cid)
+        return ClientRemoteHandle(self.exchange, self.agent_id, client_id)
 
-    def bind_to_agent(
+    def bind_to_mailbox(
         self,
-        aid: AgentIdentifier,
-    ) -> AgentRemoteHandle[BehaviorT_co]:
-        """Bind the handle to a running agent.
+        mailbox_id: Identifier,
+    ) -> BoundRemoteHandle[BehaviorT_co]:
+        """Bind the handle to an existing mailbox.
 
         Args:
-            aid: Identifier of the running agent to bind to. If this handle
-                is already bound to `aid`, `self` is returned.
+            mailbox_id: Identifier of the mailbox to bind to.
 
         Returns:
-            Remote handle bound to the agent identifier.
+            Remote handle bound to the identifier.
         """
-        if aid == self.hid:
+        if mailbox_id == self.mailbox_id:
             return self
         else:
-            return AgentRemoteHandle(self.exchange, self.aid, aid)
+            return BoundRemoteHandle(self.exchange, self.agent_id, mailbox_id)
 
     def close(
         self,
@@ -544,7 +549,11 @@ def close(
             timeout: Optional timeout used when `wait=True`.
         """
         super().close(wait_futures, timeout=timeout)
-        logger.info('Closed handle to %s with %s', self.aid, self.hid)
+        logger.info(
+            'Closed handle to %s with %s',
+            self.agent_id,
+            self.mailbox_id,
+        )
 
 
 class ClientRemoteHandle(RemoteHandle[BehaviorT_co]):
@@ -552,22 +561,21 @@ class ClientRemoteHandle(RemoteHandle[BehaviorT_co]):
 
     Args:
         exchange: Message exchange used for agent communication.
-        aid: Identifier of the target agent of this handle.
-        hid: Client identifier of this handle. If `None`, a new identifier
-            will be created using the exchange.
+        agent_id: Identifier of the target agent of this handle.
+        client_id: Client identifier of this handle. If `None`, a new
+            identifier will be created using the exchange. Note this will
+            become the `mailbox_id` attribute of the handle.
     """
 
     def __init__(
         self,
         exchange: Exchange,
-        aid: AgentIdentifier,
-        hid: ClientIdentifier | None = None,
+        agent_id: AgentIdentifier,
+        client_id: ClientIdentifier | None = None,
     ) -> None:
-        super().__init__(exchange, aid)
-
-        if hid is None:
-            hid = self.exchange.create_client()
-        self.hid = hid
+        if client_id is None:
+            client_id = exchange.create_client()
+        super().__init__(exchange, agent_id, client_id)
         self._recv_thread = threading.Thread(
             target=self._recv_responses,
             name=f'{self}-message-handler',
@@ -575,12 +583,12 @@ def __init__(
         self._recv_thread.start()
 
     def _recv_responses(self) -> None:
-        logger.debug('Started result listener thread for %s', self.hid)
-        assert self.hid is not None
+        logger.debug('Started result listener thread for %s', self.mailbox_id)
+        assert self.mailbox_id is not None
 
         while True:
             try:
-                message = self.exchange.recv(self.hid)
+                message = self.exchange.recv(self.mailbox_id)
             except MailboxClosedError:
                 break
 
@@ -590,43 +598,43 @@ def _recv_responses(self) -> None:
                 logger.error(
                     'Received invalid message response type %s from %s',
                     type(message).__name__,
-                    self.aid,
+                    self.agent_id,
                 )
 
-        logger.debug('Exiting result listener thread for %s', self.hid)
+        logger.debug('Exiting result listener thread for %s', self.mailbox_id)
 
     def bind_as_client(
         self,
-        cid: ClientIdentifier | None = None,
+        client_id: ClientIdentifier | None = None,
     ) -> ClientRemoteHandle[BehaviorT_co]:
         """Bind the handle as a unique client in the exchange.
 
         Args:
-            cid: Client identifier to be used by this handle. If `None` or
-                equal to this handle's ID, self will be returned. Otherwise, a
-                new identifier will be created using the exchange.
+            client_id: Client identifier to be used by this handle. If `None`
+                or equal to this handle's ID, self will be returned. Otherwise,
+                a new identifier will be created using the exchange.
 
         Returns:
             Remote handle bound to the client identifier.
         """
-        if cid is None or cid == self.hid:
+        if client_id is None or client_id == self.mailbox_id:
             return self
         else:
-            return ClientRemoteHandle(self.exchange, self.aid, cid)
+            return ClientRemoteHandle(self.exchange, self.agent_id, client_id)
 
-    def bind_to_agent(
+    def bind_to_mailbox(
         self,
-        aid: AgentIdentifier,
-    ) -> AgentRemoteHandle[BehaviorT_co]:
-        """Bind the handle to a running agent.
+        mailbox_id: Identifier,
+    ) -> BoundRemoteHandle[BehaviorT_co]:
+        """Bind the handle to an existing mailbox.
 
         Args:
-            aid: Identifier of the running agent to bind to.
+            mailbox_id: Identifier of the mailbox to bind to.
 
         Returns:
-            Remote handle bound to the agent identifier.
+            Remote handle bound to the identifier.
         """
-        return AgentRemoteHandle(self.exchange, self.aid, aid)
+        return BoundRemoteHandle(self.exchange, self.agent_id, mailbox_id)
 
     def close(
         self,
@@ -648,14 +656,18 @@ def close(
         """
         super().close(wait_futures, timeout=timeout)
 
-        assert isinstance(self.hid, ClientIdentifier)
+        assert isinstance(self.mailbox_id, ClientIdentifier)
         if not self._recv_thread.is_alive():
             raise RuntimeError(
-                f'Result message listener for {self.hid} is not alive. '
+                f'Result message listener for {self.mailbox_id} is not alive. '
                 'This likely means the listener thread crashed.',
             )
 
-        self.exchange.close_mailbox(self.hid)
+        self.exchange.close_mailbox(self.mailbox_id)
         self._recv_thread.join()
 
-        logger.info('Closed handle to %s with %s', self.aid, self.hid)
+        logger.info(
+            'Closed handle to %s with %s',
+            self.agent_id,
+            self.mailbox_id,
+        )
diff --git a/aeris/launcher/executor.py b/aeris/launcher/executor.py
index 7e00ec6..d694609 100644
--- a/aeris/launcher/executor.py
+++ b/aeris/launcher/executor.py
@@ -54,14 +54,14 @@ def __str__(self) -> str:
         return f'{type(self).__name__}<{self._executor}>'
 
     def _callback(self, future: Future[None]) -> None:
-        aid = self._futures.pop(future)
+        agent_id = self._futures.pop(future)
         try:
             future.result()
-            logger.info('Safely completed agent with %s', aid)
+            logger.info('Safely completed agent with %s', agent_id)
         except CancelledError:  # pragma: no cover
-            logger.warning('Cancelled agent future with %s', aid)
+            logger.warning('Cancelled agent future with %s', agent_id)
         except Exception:  # pragma: no cover
-            logger.exception('Runtime exception in agent with %s', aid)
+            logger.exception('Runtime exception in agent with %s', agent_id)
 
     def close(self) -> None:
         """Close the launcher and shutdown agents."""
@@ -83,19 +83,19 @@ def launch(
             exchange: Exchange the agent will use for messaging.
 
         Returns:
-            Mailbox used to communicate with agent.
+            Handle (unbound) used to interact with the agent.
         """
-        aid = exchange.create_agent()
+        agent_id = exchange.create_agent()
 
         agent = Agent(
             behavior,
-            aid=aid,
+            agent_id=agent_id,
             exchange=exchange,
             close_exchange=True,
         )
         future = self._executor.submit(agent)
         future.add_done_callback(self._callback)
-        self._futures[future] = aid
+        self._futures[future] = agent_id
         logger.info('Launched agent with %s', agent)
 
-        return exchange.create_handle(aid)
+        return exchange.create_handle(agent_id)
diff --git a/aeris/launcher/thread.py b/aeris/launcher/thread.py
index 011da27..41b6f94 100644
--- a/aeris/launcher/thread.py
+++ b/aeris/launcher/thread.py
@@ -82,17 +82,17 @@ def launch(
         Returns:
             Mailbox used to communicate with agent.
         """
-        aid = exchange.create_agent()
+        agent_id = exchange.create_agent()
 
         agent = Agent(
             behavior,
-            aid=aid,
+            agent_id=agent_id,
             exchange=exchange,
             close_exchange=False,
         )
-        thread = threading.Thread(target=agent, name=f'{self}-{aid}')
+        thread = threading.Thread(target=agent, name=f'{self}-{agent_id}')
         thread.start()
-        self._agents[aid] = _RunningAgent(agent, thread)
+        self._agents[agent_id] = _RunningAgent(agent, thread)
         logger.info('Launched %s', agent)
 
-        return exchange.create_handle(aid)
+        return exchange.create_handle(agent_id)
diff --git a/aeris/manager.py b/aeris/manager.py
new file mode 100644
index 0000000..12eda82
--- /dev/null
+++ b/aeris/manager.py
@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import logging
+import sys
+import threading
+from types import TracebackType
+from typing import TypeVar
+
+if sys.version_info >= (3, 11):  # pragma: >=3.11 cover
+    from typing import Self
+else:  # pragma: <3.11 cover
+    from typing_extensions import Self
+
+from aeris.behavior import Behavior
+from aeris.exchange import Exchange
+from aeris.handle import RemoteHandle
+from aeris.identifier import ClientIdentifier
+from aeris.launcher import Launcher
+from aeris.message import RequestMessage
+from aeris.multiplex import MailboxMultiplexer
+
+logger = logging.getLogger(__name__)
+
+BehaviorT = TypeVar('BehaviorT', bound=Behavior)
+
+
+class Manager:
+    """Launch and manage running agents.
+
+    The manager is provided as convenience to reduce common boilerplate code
+    for spawning agents and managing handles. Each manager registers itself
+    as a client in the exchange (i.e., each manager has its own mailbox).
+    Handles created by the manager are bound to this mailbox.
+
+    Tip:
+        This class can be used as a context manager. Upon exiting the context,
+        running agents will be shutdown, any agent handles created by the
+        manager will be closed, and the exchange and launcher will be closed.
+
+    Note:
+        The manager takes ownership of the exchange and launcher interfaces.
+        This means the manager will be responsible for closing them once the
+        manager is closed.
+
+    Args:
+        exchange: Exchange that agents and clients will use for communication.
+        launcher: Launcher used to execute agents remotely.
+    """
+
+    def __init__(self, exchange: Exchange, launcher: Launcher) -> None:
+        self._exchange = exchange
+        self._launcher = launcher
+
+        self._mailbox_id = exchange.create_client()
+        self._multiplexer = MailboxMultiplexer(
+            self.mailbox_id,
+            self._exchange,
+            self._handle_request,
+        )
+        self._listener_thread = threading.Thread(
+            target=self._multiplexer.listen,
+            name=f'multiplexer-{self.mailbox_id.uid}-listener',
+        )
+        self._listener_thread.start()
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        exc_traceback: TracebackType | None,
+    ) -> None:
+        self.close()
+
+    def __repr__(self) -> str:
+        return (
+            f'{type(self).__name__}'
+            f'(exchange={self._exchange!r}, launcher={self._launcher!r})'
+        )
+
+    def __str__(self) -> str:
+        return (
+            f'{type(self).__name__}<{self.mailbox_id}, {self._exchange}, '
+            f'{self._launcher}>'
+        )
+
+    @property
+    def exchange(self) -> Exchange:
+        """Exchange interface."""
+        return self._exchange
+
+    @property
+    def launcher(self) -> Launcher:
+        """Launcher interface."""
+        return self._launcher
+
+    @property
+    def mailbox_id(self) -> ClientIdentifier:
+        """Identifier of the mailbox used by this manager."""
+        return self._mailbox_id
+
+    def _handle_request(self, request: RequestMessage) -> None:
+        response = request.error(
+            TypeError(
+                f'Client with {self.mailbox_id} cannot fulfill requests.',
+            ),
+        )
+        self.exchange.send(response.dest, response)
+
+    def close(self) -> None:
+        """Close the manager and cleanup resources.
+
+        1. Call shutdown on all launched agents.
+        1. Close all handles created by the manager.
+        1. Close the mailbox associated with the manager.
+        1. Close the exchange.
+        1. Close the launcher.
+        """
+        for handle in self._multiplexer.bound_handles.values():
+            handle.shutdown()
+        self._multiplexer.close_bound_handles()
+        self._multiplexer.close_mailbox()
+        self._listener_thread.join()
+        self.exchange.close()
+        self.launcher.close()
+
+    def launch(self, behavior: BehaviorT) -> RemoteHandle[BehaviorT]:
+        """Launch a new agent with a specified behavior.
+
+        Note:
+            Compared to `Launcher.launch()`, this method will inject the
+            exchange and return a client-bound handle.
+
+        Args:
+            behavior: Behavior the agent should implement.
+
+        Returns:
+            Handle (client bound) used to interact with the agent.
+        """
+        unbound = self.launcher.launch(behavior, exchange=self.exchange)
+        bound = self._multiplexer.bind(unbound)
+        return bound
diff --git a/aeris/message.py b/aeris/message.py
index 5a2f233..08fe327 100644
--- a/aeris/message.py
+++ b/aeris/message.py
@@ -29,9 +29,13 @@ class BaseMessage(BaseModel):
         and message ID.
 
     Args:
-        mid: Unique message ID.
-        src: Source entity.
-        dest: Destination entity.
+        tag: Unique message tag. Used to match requests and responses.
+        src: Source mailbox identifier.
+        dest: Destination mailbox identifier.
+        label: Optional label used to disambiguate response messages when
+            multiple objects (i.e., handles) share the same mailbox.
+            Note that is is distinct from the tag which is just for matching
+            requests to responses.
     """
 
     model_config = ConfigDict(
@@ -42,12 +46,13 @@ class BaseMessage(BaseModel):
         validate_default=True,
     )
 
-    mid: uuid.UUID = Field(default_factory=uuid.uuid4)
+    tag: uuid.UUID = Field(default_factory=uuid.uuid4)
     src: Identifier
     dest: Identifier
+    label: Optional[uuid.UUID] = Field(None)  # noqa: UP007
 
     def __hash__(self) -> int:
-        return hash(type(self)) + hash(self.mid)
+        return hash(type(self)) + hash(self.tag) + hash(self.label)
 
     @classmethod
     def model_from_json(cls, data: str) -> Message:
@@ -102,9 +107,10 @@ def error(self, exception: Exception) -> ActionResponse:
             exception: Error of the action.
         """
         return ActionResponse(
-            mid=self.mid,
+            tag=self.tag,
             src=self.dest,
             dest=self.src,
+            label=self.label,
             action=self.action,
             result=None,
             exception=exception,
@@ -117,9 +123,10 @@ def response(self, result: Any) -> ActionResponse:
             result: Result of the action.
         """
         return ActionResponse(
-            mid=self.mid,
+            tag=self.tag,
             src=self.dest,
             dest=self.src,
+            label=self.label,
             action=self.action,
             result=result,
             exception=None,
@@ -165,9 +172,10 @@ def __eq__(self, other: object, /) -> bool:
         if not isinstance(other, ActionResponse):
             return False
         return (
-            self.mid == other.mid
+            self.tag == other.tag
             and self.src == other.src
             and self.dest == other.dest
+            and self.label == other.label
             and self.action == other.action
             and self.result == other.result
             # Custom __eq__ is required because exception instances need
@@ -185,9 +193,10 @@ class PingRequest(BaseMessage):
     def response(self) -> PingResponse:
         """Construct a ping response message."""
         return PingResponse(
-            mid=self.mid,
+            tag=self.tag,
             src=self.dest,
             dest=self.src,
+            label=self.label,
             exception=None,
         )
 
@@ -198,9 +207,10 @@ def error(self, exception: Exception) -> PingResponse:
             exception: Error of the action.
         """
         return PingResponse(
-            mid=self.mid,
+            tag=self.tag,
             src=self.dest,
             dest=self.src,
+            label=self.label,
             exception=exception,
         )
 
@@ -229,9 +239,10 @@ def __eq__(self, other: object, /) -> bool:
         if not isinstance(other, PingResponse):
             return False
         return (
-            self.mid == other.mid
+            self.tag == other.tag
             and self.src == other.src
             and self.dest == other.dest
+            and self.label == other.label
             # Custom __eq__ is required because exception instances need
             # to be compared by type. I.e., Exception() == Exception() is
             # always False.
@@ -257,9 +268,10 @@ def _validate_agent(cls, dest: Identifier) -> Identifier:
     def response(self) -> ShutdownResponse:
         """Construct a shutdown response message."""
         return ShutdownResponse(
-            mid=self.mid,
+            tag=self.tag,
             src=self.dest,
             dest=self.src,
+            label=self.label,
             exception=None,
         )
 
@@ -270,9 +282,10 @@ def error(self, exception: Exception) -> ShutdownResponse:
             exception: Error of the action.
         """
         return ShutdownResponse(
-            mid=self.mid,
+            tag=self.tag,
             src=self.dest,
             dest=self.src,
+            label=self.label,
             exception=exception,
         )
 
@@ -301,7 +314,7 @@ def __eq__(self, other: object, /) -> bool:
         if not isinstance(other, ShutdownResponse):
             return False
         return (
-            self.mid == other.mid
+            self.tag == other.tag
             and self.src == other.src
             and self.dest == other.dest
             # Custom __eq__ is required because exception instances need
diff --git a/aeris/multiplex.py b/aeris/multiplex.py
new file mode 100644
index 0000000..1ee8df8
--- /dev/null
+++ b/aeris/multiplex.py
@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import logging
+import sys
+import uuid
+from types import TracebackType
+from typing import Any
+from typing import Callable
+from typing import get_args
+from typing import TypeVar
+
+if sys.version_info >= (3, 11):  # pragma: >=3.11 cover
+    from typing import Self
+else:  # pragma: <3.11 cover
+    from typing_extensions import Self
+
+from aeris.behavior import Behavior
+from aeris.exception import MailboxClosedError
+from aeris.exchange import Exchange
+from aeris.handle import BoundRemoteHandle
+from aeris.handle import RemoteHandle
+from aeris.identifier import Identifier
+from aeris.message import Message
+from aeris.message import RequestMessage
+from aeris.message import ResponseMessage
+
+logger = logging.getLogger(__name__)
+
+BehaviorT_co = TypeVar('BehaviorT_co', bound=Behavior, covariant=True)
+
+
+class MailboxMultiplexer:
+    """Multiplex a single mailbox across many consumers.
+
+    A mailbox represents a recipient entity. In many cases, there may be
+    many entities within a single process that want to send and receive
+    messages. For example, a running agent may have multiple handles to other
+    agents. A naive approach would be for the agent and each handle to have
+    their own mailbox, but this requires a listening thread in the process
+    for each mailbox. This does not scale well. The multiplexer lets
+    multiple entities (e.g., an agent and its handles) share the a single
+    mailbox so their is one listening thread and messages are dispatched
+    to the appropriate entity (i.e., object) within the process.
+
+    Note:
+        This class should not be considered as a part of the public API. It
+        is used internally by other components, such as the
+        [`Agent`][aeris.agent.Agent] and [`Manager`][aeris.manager.Manager],
+        which use multiple handles concurrently.
+
+    Args:
+        mailbox_id: Identifier of the mailbox to multiplex. For example, the
+            identifier of an agent.
+        exchange: The exchange interface managing the mailbox.
+        request_handler: A callable invoked when the request message is
+            received to the inbox.
+    """
+
+    def __init__(
+        self,
+        mailbox_id: Identifier,
+        exchange: Exchange,
+        request_handler: Callable[[RequestMessage], None],
+    ) -> None:
+        self.mailbox_id = mailbox_id
+        self.exchange = exchange
+        self.request_handler = request_handler
+        self.bound_handles: dict[uuid.UUID, BoundRemoteHandle[Any]] = {}
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        exc_traceback: TracebackType | None,
+    ) -> None:
+        self.close()
+
+    def __repr__(self) -> str:
+        name = type(self).__name__
+        return (
+            f'{name}(mailbox_id={self.mailbox_id!r}, '
+            f'exchange={self.exchange!r})'
+        )
+
+    def __str__(self) -> str:
+        return f'{type(self).__name__}<{self.mailbox_id}; {self.exchange}>'
+
+    def _message_handler(self, message: Message) -> None:
+        if isinstance(message, get_args(RequestMessage)):
+            self.request_handler(message)
+        elif isinstance(message, get_args(ResponseMessage)):
+            try:
+                handle = self.bound_handles[message.label]
+            except KeyError:
+                logger.exception(
+                    'Receieved a response message from %s but no handle to '
+                    'that agent is bound to this multiplexer.',
+                    message.src,
+                )
+            else:
+                handle._process_response(message)
+        else:
+            raise AssertionError('Unreachable.')
+
+    def bind(
+        self,
+        handle: RemoteHandle[BehaviorT_co],
+    ) -> BoundRemoteHandle[BehaviorT_co]:
+        """Bind a handle to this mailbox.
+
+        Args:
+            handle: Remote handle to bind.
+
+        Returns:
+            Remote handle bound to this mailbox.
+        """
+        bound = handle.bind_to_mailbox(self.mailbox_id)
+        self.bound_handles[bound.handle_id] = bound
+        logger.debug(
+            'Bound remote handle to %s to client multiplexer with %s',
+            handle.agent_id,
+            self.mailbox_id,
+        )
+        return bound
+
+    def close(self) -> None:
+        """Close the multiplexer.
+
+        Closes all handles bound to this mailbox and then closes the mailbox.
+        """
+        self.close_bound_handles()
+        self.close_mailbox()
+
+    def close_bound_handles(self) -> None:
+        """Close all handles bound to this mailbox."""
+        for key in tuple(self.bound_handles):
+            handle = self.bound_handles.pop(key)
+            handle.close(wait_futures=False)
+
+    def close_mailbox(self) -> None:
+        """Close the mailbox."""
+        self.exchange.close_mailbox(self.mailbox_id)
+
+    def listen(self) -> None:
+        """Listen for new messages in the mailbox and process them.
+
+        Request messages are processed via the `request_handler`, and response
+        messages are dispatched to the handle that created the corresponding
+        request.
+
+        Warning:
+            This method loops forever, until the mailbox is closed. Thus this
+            method is typically run inside of a thread.
+
+        Note:
+            Response messages intended for a handle that does not exist
+            will be logged and discarded.
+        """
+        logger.info('Message listener started for %s', self.mailbox_id)
+
+        while True:
+            try:
+                message = self.exchange.recv(self.mailbox_id)
+            except MailboxClosedError:
+                break
+            else:
+                self._message_handler(message)
diff --git a/examples/01-actor-client/run-01.py b/examples/01-actor-client/run-01.py
index 127b139..ea23888 100644
--- a/examples/01-actor-client/run-01.py
+++ b/examples/01-actor-client/run-01.py
@@ -8,6 +8,7 @@
 from aeris.exchange.thread import ThreadExchange
 from aeris.launcher.thread import ThreadLauncher
 from aeris.logging import init_logging
+from aeris.manager import Manager
 
 
 class Counter(Behavior):
@@ -28,9 +29,12 @@ def get_count(self) -> int:
 def main() -> int:
     init_logging(logging.DEBUG)
 
-    with ThreadExchange() as exchange, ThreadLauncher() as launcher:
+    with Manager(
+        exchange=ThreadExchange(),
+        launcher=ThreadLauncher(),
+    ) as manager:
         behavior = Counter()
-        agent = launcher.launch(behavior, exchange).bind_as_client()
+        agent = manager.launch(behavior)
 
         future: Future[int] = agent.action('get_count')
         assert future.result() == 0
@@ -40,8 +44,6 @@ def main() -> int:
         future = agent.action('get_count')
         assert future.result() == 1
 
-        agent.close()
-
     return 0
 
 
diff --git a/examples/02-agent-loop/run-02.py b/examples/02-agent-loop/run-02.py
index 7f07790..0f4044b 100644
--- a/examples/02-agent-loop/run-02.py
+++ b/examples/02-agent-loop/run-02.py
@@ -11,6 +11,7 @@
 from aeris.exchange.thread import ThreadExchange
 from aeris.launcher.thread import ThreadLauncher
 from aeris.logging import init_logging
+from aeris.manager import Manager
 
 
 class Counter(Behavior):
@@ -33,17 +34,18 @@ def get_count(self) -> int:
 def main() -> int:
     init_logging(logging.DEBUG)
 
-    with ThreadExchange() as exchange, ThreadLauncher() as launcher:
+    with Manager(
+        exchange=ThreadExchange(),
+        launcher=ThreadLauncher(),
+    ) as manager:
         behavior = Counter()
-        agent = launcher.launch(behavior, exchange).bind_as_client()
+        agent = manager.launch(behavior)
 
         time.sleep(2)
 
         future: Future[int] = agent.action('get_count')
         assert future.result() >= 1
 
-        agent.close()
-
     return 0
 
 
diff --git a/examples/03-agent-agent/run-03.py b/examples/03-agent-agent/run-03.py
index 823a77e..3d9a66c 100644
--- a/examples/03-agent-agent/run-03.py
+++ b/examples/03-agent-agent/run-03.py
@@ -9,6 +9,7 @@
 from aeris.handle import Handle
 from aeris.launcher.thread import ThreadLauncher
 from aeris.logging import init_logging
+from aeris.manager import Manager
 
 
 class Coordinator(Behavior):
@@ -42,13 +43,13 @@ def reverse(self, text: str) -> str:
 def main() -> int:
     init_logging(logging.DEBUG)
 
-    with ThreadExchange() as exchange, ThreadLauncher() as launcher:
-        lowerer = launcher.launch(Lowerer(), exchange)
-        reverser = launcher.launch(Reverser(), exchange)
-        coordinator = launcher.launch(
-            Coordinator(lowerer, reverser),
-            exchange,
-        ).bind_as_client()
+    with Manager(
+        exchange=ThreadExchange(),
+        launcher=ThreadLauncher(),
+    ) as manager:
+        lowerer = manager.launch(Lowerer())
+        reverser = manager.launch(Reverser())
+        coordinator = manager.launch(Coordinator(lowerer, reverser))
 
         text = 'DEADBEEF'
         expected = 'feebdaed'
@@ -56,10 +57,6 @@ def main() -> int:
         future: Future[str] = coordinator.action('process', text)
         assert future.result() == expected
 
-        coordinator.close()
-        reverser.close()
-        lowerer.close()
-
     return 0
 
 
diff --git a/examples/04-execution/run-04.py b/examples/04-execution/run-04.py
index e3a056f..29f38f4 100644
--- a/examples/04-execution/run-04.py
+++ b/examples/04-execution/run-04.py
@@ -1,18 +1,16 @@
 from __future__ import annotations
 
-import asyncio
 import logging
 from concurrent.futures import Future
 from concurrent.futures import ProcessPoolExecutor
 
 from aeris.behavior import action
 from aeris.behavior import Behavior
-from aeris.exchange.simple import serve_forever
-from aeris.exchange.simple import SimpleServer
 from aeris.exchange.simple import spawn_simple_exchange
 from aeris.handle import Handle
 from aeris.launcher.executor import ExecutorLauncher
 from aeris.logging import init_logging
+from aeris.manager import Manager
 
 EXCHANGE_PORT = 5346
 logger = logging.getLogger(__name__)
@@ -46,29 +44,22 @@ def reverse(self, text: str) -> str:
         return text[::-1]
 
 
-def run_exchange_server() -> None:
-    server = SimpleServer('localhost', EXCHANGE_PORT)
-    asyncio.run(serve_forever(server))
-
-
 def main() -> int:
     init_logging(logging.INFO)
 
     with spawn_simple_exchange('localhost', EXCHANGE_PORT) as exchange:
-        # Agents are launched using a Launcher. The ExecutorLauncher can use
-        # any concurrent.futures.Executor (here, a ProcessPoolExecutor) to
-        # execute agents. The launcher is initialized to use the simple
-        # exchange that we just set up for agent and client communication.
-        with ExecutorLauncher(ProcessPoolExecutor(max_workers=3)) as launcher:
-            # Initialize and launch each of the three agents. The
-            # bind_as_client method returns a client handle to the agent that
-            # can be used to send commands to that agent.
-            lowerer = launcher.launch(Lowerer(), exchange).bind_as_client()
-            reverser = launcher.launch(Reverser(), exchange).bind_as_client()
-            coordinator = launcher.launch(
-                Coordinator(lowerer, reverser),
-                exchange,
-            ).bind_as_client()
+        with Manager(
+            exchange=exchange,
+            # Agents are launched using a Launcher. The ExecutorLauncher can
+            # use any concurrent.futures.Executor (here, a ProcessPoolExecutor)
+            # to execute agents.
+            launcher=ExecutorLauncher(ProcessPoolExecutor(max_workers=3)),
+        ) as manager:
+            # Initialize and launch each of the three agents. The returned
+            # type is a handle to that agent used to invoke actions.
+            lowerer = manager.launch(Lowerer())
+            reverser = manager.launch(Reverser())
+            coordinator = manager.launch(Coordinator(lowerer, reverser))
 
             text = 'DEADBEEF'
             expected = 'feebdaed'
@@ -76,15 +67,8 @@ def main() -> int:
             future: Future[str] = coordinator.action('process', text)
             assert future.result() == expected
 
-            # Instruct all agents to shutdown
-            lowerer.shutdown()
-            reverser.shutdown()
-            coordinator.shutdown()
-
-            # Close client handles to all agents
-            lowerer.close()
-            reverser.close()
-            coordinator.close()
+        # Upon exit, the Manager context will instruct each agent to shutdown
+        # and then close the handles, exchange, and launcher interfaces.
 
     return 0
 


Test patch:
diff --git a/tests/agent_test.py b/tests/agent_test.py
index ca4dd03..3a583fc 100644
--- a/tests/agent_test.py
+++ b/tests/agent_test.py
@@ -10,11 +10,10 @@
 from aeris.behavior import Behavior
 from aeris.behavior import loop
 from aeris.exchange import Exchange
-from aeris.handle import AgentRemoteHandle
+from aeris.handle import BoundRemoteHandle
 from aeris.handle import ClientRemoteHandle
 from aeris.handle import Handle
 from aeris.handle import UnboundRemoteHandle
-from aeris.identifier import AgentIdentifier
 from aeris.message import ActionRequest
 from aeris.message import ActionResponse
 from aeris.message import PingRequest
@@ -50,8 +49,8 @@ def setter(self, shutdown: threading.Event) -> None:
 
 
 def test_agent_start_shutdown(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    agent = Agent(SignalingBehavior(), aid=aid, exchange=exchange)
+    agent_id = exchange.create_agent()
+    agent = Agent(SignalingBehavior(), agent_id=agent_id, exchange=exchange)
 
     agent.start()
     agent.start()  # Idempotency check.
@@ -65,9 +64,19 @@ def test_agent_start_shutdown(exchange: Exchange) -> None:
     assert agent.behavior.shutdown_event.is_set()
 
 
+def test_agent_shutdown_without_start(exchange: Exchange) -> None:
+    agent_id = exchange.create_agent()
+    agent = Agent(SignalingBehavior(), agent_id=agent_id, exchange=exchange)
+
+    agent.shutdown()
+
+    assert not agent.behavior.setup_event.is_set()
+    assert agent.behavior.shutdown_event.is_set()
+
+
 def test_agent_run_in_thread(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    agent = Agent(SignalingBehavior(), aid=aid, exchange=exchange)
+    agent_id = exchange.create_agent()
+    agent = Agent(SignalingBehavior(), agent_id=agent_id, exchange=exchange)
     assert isinstance(repr(agent), str)
     assert isinstance(str(agent), str)
 
@@ -83,134 +92,129 @@ def test_agent_run_in_thread(exchange: Exchange) -> None:
 
 
 def test_agent_shutdown_message(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    cid = exchange.create_client()
+    agent_id = exchange.create_agent()
+    client_id = exchange.create_client()
 
-    agent = Agent(EmptyBehavior(), aid=aid, exchange=exchange)
+    agent = Agent(EmptyBehavior(), agent_id=agent_id, exchange=exchange)
     thread = threading.Thread(target=agent)
     thread.start()
 
-    shutdown = ShutdownRequest(src=cid, dest=aid)
-    exchange.send(aid, shutdown)
+    shutdown = ShutdownRequest(src=client_id, dest=agent_id)
+    exchange.send(agent_id, shutdown)
 
     thread.join(timeout=TEST_THREAD_JOIN_TIMEOUT)
     assert not thread.is_alive()
 
 
 def test_agent_ping_message(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    cid = exchange.create_client()
+    agent_id = exchange.create_agent()
+    client_id = exchange.create_client()
 
-    agent = Agent(EmptyBehavior(), aid=aid, exchange=exchange)
+    agent = Agent(EmptyBehavior(), agent_id=agent_id, exchange=exchange)
     assert isinstance(repr(agent), str)
     assert isinstance(str(agent), str)
 
     thread = threading.Thread(target=agent)
     thread.start()
 
-    ping = PingRequest(src=cid, dest=aid)
-    exchange.send(aid, ping)
-    message = exchange.recv(cid)
+    ping = PingRequest(src=client_id, dest=agent_id)
+    exchange.send(agent_id, ping)
+    message = exchange.recv(client_id)
     assert isinstance(message, PingResponse)
 
-    shutdown = ShutdownRequest(src=cid, dest=aid)
-    exchange.send(aid, shutdown)
+    shutdown = ShutdownRequest(src=client_id, dest=agent_id)
+    exchange.send(agent_id, shutdown)
 
     thread.join(timeout=TEST_THREAD_JOIN_TIMEOUT)
     assert not thread.is_alive()
 
 
 def test_agent_action_message(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    cid = exchange.create_client()
+    agent_id = exchange.create_agent()
+    client_id = exchange.create_client()
 
-    agent = Agent(CounterBehavior(), aid=aid, exchange=exchange)
+    agent = Agent(CounterBehavior(), agent_id=agent_id, exchange=exchange)
     thread = threading.Thread(target=agent)
     thread.start()
 
     value = 42
-    request = ActionRequest(src=cid, dest=aid, action='add', args=(value,))
-    exchange.send(aid, request)
-    message = exchange.recv(cid)
+    request = ActionRequest(
+        src=client_id,
+        dest=agent_id,
+        action='add',
+        args=(value,),
+    )
+    exchange.send(agent_id, request)
+    message = exchange.recv(client_id)
     assert isinstance(message, ActionResponse)
     assert message.exception is None
     assert message.result is None
 
-    request = ActionRequest(src=cid, dest=aid, action='count')
-    exchange.send(aid, request)
-    message = exchange.recv(cid)
+    request = ActionRequest(src=client_id, dest=agent_id, action='count')
+    exchange.send(agent_id, request)
+    message = exchange.recv(client_id)
     assert isinstance(message, ActionResponse)
     assert message.exception is None
     assert message.result == value
 
-    shutdown = ShutdownRequest(src=cid, dest=aid)
-    exchange.send(aid, shutdown)
+    shutdown = ShutdownRequest(src=client_id, dest=agent_id)
+    exchange.send(agent_id, shutdown)
 
     thread.join(timeout=TEST_THREAD_JOIN_TIMEOUT)
     assert not thread.is_alive()
 
 
 def test_agent_action_message_error(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    cid = exchange.create_client()
+    agent_id = exchange.create_agent()
+    client_id = exchange.create_client()
 
-    agent = Agent(ErrorBehavior(), aid=aid, exchange=exchange)
+    agent = Agent(ErrorBehavior(), agent_id=agent_id, exchange=exchange)
     thread = threading.Thread(target=agent)
     thread.start()
 
-    request = ActionRequest(src=cid, dest=aid, action='fails')
-    exchange.send(aid, request)
-    message = exchange.recv(cid)
+    request = ActionRequest(src=client_id, dest=agent_id, action='fails')
+    exchange.send(agent_id, request)
+    message = exchange.recv(client_id)
     assert isinstance(message, ActionResponse)
     assert isinstance(message.exception, RuntimeError)
     assert 'This action always fails.' in str(message.exception)
 
-    shutdown = ShutdownRequest(src=cid, dest=aid)
-    exchange.send(aid, shutdown)
+    shutdown = ShutdownRequest(src=client_id, dest=agent_id)
+    exchange.send(agent_id, shutdown)
 
     thread.join(timeout=TEST_THREAD_JOIN_TIMEOUT)
     assert not thread.is_alive()
 
 
 def test_agent_action_message_unknown(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    cid = exchange.create_client()
+    agent_id = exchange.create_agent()
+    client_id = exchange.create_client()
 
-    agent = Agent(EmptyBehavior(), aid=aid, exchange=exchange)
+    agent = Agent(EmptyBehavior(), agent_id=agent_id, exchange=exchange)
     thread = threading.Thread(target=agent)
     thread.start()
 
-    request = ActionRequest(src=cid, dest=aid, action='null')
-    exchange.send(aid, request)
-    message = exchange.recv(cid)
+    request = ActionRequest(src=client_id, dest=agent_id, action='null')
+    exchange.send(agent_id, request)
+    message = exchange.recv(client_id)
     assert isinstance(message, ActionResponse)
     assert isinstance(message.exception, AttributeError)
     assert 'null' in str(message.exception)
 
-    shutdown = ShutdownRequest(src=cid, dest=aid)
-    exchange.send(aid, shutdown)
+    shutdown = ShutdownRequest(src=client_id, dest=agent_id)
+    exchange.send(agent_id, shutdown)
 
     thread.join(timeout=TEST_THREAD_JOIN_TIMEOUT)
     assert not thread.is_alive()
 
 
-def test_agent_log_bad_response(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    agent = Agent(EmptyBehavior(), aid=aid, exchange=exchange)
-    response = PingResponse(
-        src=AgentIdentifier.new(),
-        dest=AgentIdentifier.new(),
-    )
-    agent._response_handler(response)
-
-
 class HandleBindingBehavior(Behavior):
     def __init__(
         self,
         unbound: UnboundRemoteHandle[EmptyBehavior],
         client_bound: ClientRemoteHandle[EmptyBehavior],
-        agent_bound: AgentRemoteHandle[EmptyBehavior],
-        self_bound: AgentRemoteHandle[EmptyBehavior],
+        agent_bound: BoundRemoteHandle[EmptyBehavior],
+        self_bound: BoundRemoteHandle[EmptyBehavior],
     ) -> None:
         self.unbound = unbound
         self.client_bound = client_bound
@@ -218,13 +222,17 @@ def __init__(
         self.self_bound = self_bound
 
     def setup(self) -> None:
-        assert isinstance(self.unbound, AgentRemoteHandle)
+        assert isinstance(self.unbound, BoundRemoteHandle)
         assert isinstance(self.client_bound, ClientRemoteHandle)
-        assert isinstance(self.agent_bound, AgentRemoteHandle)
-        assert isinstance(self.self_bound, AgentRemoteHandle)
+        assert isinstance(self.agent_bound, BoundRemoteHandle)
+        assert isinstance(self.self_bound, BoundRemoteHandle)
 
-        assert self.unbound.hid is not None
-        assert self.unbound.hid == self.agent_bound.hid == self.self_bound.hid
+        assert self.unbound.mailbox_id is not None
+        assert (
+            self.unbound.mailbox_id
+            == self.agent_bound.mailbox_id
+            == self.self_bound.mailbox_id
+        )
 
     def shutdown(self) -> None:
         self.unbound.close()
@@ -234,56 +242,29 @@ def shutdown(self) -> None:
 
 
 def test_agent_run_bind_handles(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
+    agent_id = exchange.create_agent()
     behavior = HandleBindingBehavior(
         unbound=UnboundRemoteHandle(exchange, exchange.create_agent()),
         client_bound=ClientRemoteHandle(exchange, exchange.create_agent()),
-        agent_bound=AgentRemoteHandle(
+        agent_bound=BoundRemoteHandle(
             exchange,
             exchange.create_agent(),
             exchange.create_agent(),
         ),
-        self_bound=AgentRemoteHandle(
+        self_bound=BoundRemoteHandle(
             exchange,
             exchange.create_agent(),
-            aid,
+            agent_id,
         ),
     )
-    agent = Agent(behavior, aid=aid, exchange=exchange)
+    agent = Agent(behavior, agent_id=agent_id, exchange=exchange)
 
     agent._bind_handles()
     agent._bind_handles()  # Idempotency check
     agent.behavior.setup()
     agent.behavior.shutdown()
     # The client-bound and self-bound remote handles should be ignored.
-    assert len(agent._bound_handles) == 2  # noqa: PLR2004
-
-
-class DuplicateBindingsBehavior(Behavior):
-    def __init__(
-        self,
-        handle1: UnboundRemoteHandle[EmptyBehavior],
-        handle2: UnboundRemoteHandle[EmptyBehavior],
-    ) -> None:
-        self.handle1 = handle1
-        self.handle2 = handle2
-
-
-def test_agent_run_duplicate_handles_error(exchange: Exchange) -> None:
-    self_aid = exchange.create_agent()
-    remote_aid = exchange.create_agent()
-    behavior = DuplicateBindingsBehavior(
-        handle1=UnboundRemoteHandle(exchange, remote_aid),
-        handle2=UnboundRemoteHandle(exchange, remote_aid),
-    )
-    agent = Agent(behavior, aid=self_aid, exchange=exchange)
-
-    error = (
-        f'{agent} already has a handle bound to a remote agent '
-        f'with {remote_aid}.'
-    )
-    with pytest.raises(RuntimeError, match=error):
-        agent.run()
+    assert len(agent._multiplexer.bound_handles) == 2  # noqa: PLR2004
 
 
 class RunBehavior(Behavior):
@@ -291,7 +272,7 @@ def __init__(self, doubler: Handle[DoubleBehavior]) -> None:
         self.doubler = doubler
 
     def shutdown(self) -> None:
-        assert isinstance(self.doubler, AgentRemoteHandle)
+        assert isinstance(self.doubler, BoundRemoteHandle)
         self.doubler.shutdown()
 
     @action
@@ -319,8 +300,16 @@ def test_agent_to_handle_handles(exchange: Exchange) -> None:
     runner_behavior = RunBehavior(doubler_handle)
     doubler_behavior = DoubleBehavior()
 
-    runner_agent = Agent(runner_behavior, aid=runner_id, exchange=exchange)
-    doubler_agent = Agent(doubler_behavior, aid=doubler_id, exchange=exchange)
+    runner_agent = Agent(
+        runner_behavior,
+        agent_id=runner_id,
+        exchange=exchange,
+    )
+    doubler_agent = Agent(
+        doubler_behavior,
+        agent_id=doubler_id,
+        exchange=exchange,
+    )
 
     runner_thread = threading.Thread(target=runner_agent)
     doubler_thread = threading.Thread(target=doubler_agent)
diff --git a/tests/handle_test.py b/tests/handle_test.py
index ae1f062..e0a0dda 100644
--- a/tests/handle_test.py
+++ b/tests/handle_test.py
@@ -10,7 +10,7 @@
 from aeris.exception import HandleNotBoundError
 from aeris.exchange import Exchange
 from aeris.exchange.thread import ThreadExchange
-from aeris.handle import AgentRemoteHandle
+from aeris.handle import BoundRemoteHandle
 from aeris.handle import ClientRemoteHandle
 from aeris.handle import Handle
 from aeris.handle import ProxyHandle
@@ -50,9 +50,9 @@ def test_proxy_handle_errors() -> None:
 
 
 def test_unbound_remote_handle_serialize(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
+    agent_id = exchange.create_agent()
     handle: UnboundRemoteHandle[Any]
-    with UnboundRemoteHandle(exchange, aid) as handle:
+    with UnboundRemoteHandle(exchange, agent_id) as handle:
         # Note: don't call pickle.dumps here because ThreadExchange
         # is not pickleable so we test __reduce__ directly.
         class_, args = handle.__reduce__()
@@ -63,22 +63,22 @@ def test_unbound_remote_handle_serialize(exchange: Exchange) -> None:
 
 
 def test_unbound_remote_handle_bind(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
+    agent_id = exchange.create_agent()
     handle: UnboundRemoteHandle[Any]
-    with UnboundRemoteHandle(exchange, aid) as handle:
+    with UnboundRemoteHandle(exchange, agent_id) as handle:
         client_bound: ClientRemoteHandle[Any]
         with handle.bind_as_client() as client_bound:
             assert isinstance(client_bound, ClientRemoteHandle)
-        agent_bound: AgentRemoteHandle[Any]
-        with handle.bind_to_agent(AgentIdentifier.new()) as agent_bound:
-            assert isinstance(agent_bound, AgentRemoteHandle)
+        agent_bound: BoundRemoteHandle[Any]
+        with handle.bind_to_mailbox(AgentIdentifier.new()) as agent_bound:
+            assert isinstance(agent_bound, BoundRemoteHandle)
 
 
 def test_unbound_remote_handle_errors(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
+    agent_id = exchange.create_agent()
     handle: UnboundRemoteHandle[Any]
-    with UnboundRemoteHandle(exchange, aid) as handle:
-        request = PingRequest(src=ClientIdentifier.new(), dest=aid)
+    with UnboundRemoteHandle(exchange, agent_id) as handle:
+        request = PingRequest(src=ClientIdentifier.new(), dest=agent_id)
         with pytest.raises(HandleNotBoundError):
             handle._send_request(request)
         with pytest.raises(HandleNotBoundError):
@@ -90,14 +90,14 @@ def test_unbound_remote_handle_errors(exchange: Exchange) -> None:
 
 
 def test_remote_handle_closed_error(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
+    agent_id = exchange.create_agent()
     handles: list[RemoteHandle[Any]] = [
-        AgentRemoteHandle(exchange, aid, exchange.create_agent()),
-        ClientRemoteHandle(exchange, aid, exchange.create_client()),
+        BoundRemoteHandle(exchange, agent_id, exchange.create_agent()),
+        ClientRemoteHandle(exchange, agent_id, exchange.create_client()),
     ]
     for handle in handles:
         handle.close()
-        assert handle.hid is not None
+        assert handle.mailbox_id is not None
         with pytest.raises(HandleClosedError):
             handle.action('foo')
         with pytest.raises(HandleClosedError):
@@ -107,10 +107,10 @@ def test_remote_handle_closed_error(exchange: Exchange) -> None:
 
 
 def test_agent_remote_handle_serialize(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    hid = exchange.create_agent()
-    handle: AgentRemoteHandle[Any]
-    with AgentRemoteHandle(exchange, aid, hid) as handle:
+    agent_id = exchange.create_agent()
+    mailbox_id = exchange.create_agent()
+    handle: BoundRemoteHandle[Any]
+    with BoundRemoteHandle(exchange, agent_id, mailbox_id) as handle:
         # Note: don't call pickle.dumps here because ThreadExchange
         # is not pickleable so we test __reduce__ directly.
         class_, args = handle.__reduce__()
@@ -118,36 +118,36 @@ def test_agent_remote_handle_serialize(exchange: Exchange) -> None:
             assert isinstance(reconstructed, UnboundRemoteHandle)
             assert str(reconstructed) != str(handle)
             assert repr(reconstructed) != repr(handle)
-            assert reconstructed.aid == handle.aid
+            assert reconstructed.agent_id == handle.agent_id
 
 
 def test_agent_remote_handle_bind(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    hid = exchange.create_agent()
-    handle: AgentRemoteHandle[Any]
-    with AgentRemoteHandle(exchange, aid, hid) as handle:
-        assert isinstance(handle.hid, AgentIdentifier)
+    agent_id = exchange.create_agent()
+    mailbox_id = exchange.create_agent()
+    handle: BoundRemoteHandle[Any]
+    with BoundRemoteHandle(exchange, agent_id, mailbox_id) as handle:
+        assert isinstance(handle.mailbox_id, AgentIdentifier)
         client_bound: ClientRemoteHandle[Any]
         with handle.bind_as_client() as client_bound:
             assert isinstance(client_bound, ClientRemoteHandle)
         with pytest.raises(
             ValueError,
-            match=f'Cannot create handle to {handle.aid}',
+            match=f'Cannot create handle to {handle.agent_id}',
         ):
-            handle.bind_to_agent(handle.aid)
-        agent_bound: AgentRemoteHandle[Any]
-        with handle.bind_to_agent(handle.hid) as agent_bound:
+            handle.bind_to_mailbox(handle.agent_id)
+        agent_bound: BoundRemoteHandle[Any]
+        with handle.bind_to_mailbox(handle.mailbox_id) as agent_bound:
             assert agent_bound is handle
-        with handle.bind_to_agent(AgentIdentifier.new()) as agent_bound:
+        with handle.bind_to_mailbox(AgentIdentifier.new()) as agent_bound:
             assert agent_bound is not handle
-            assert isinstance(agent_bound, AgentRemoteHandle)
+            assert isinstance(agent_bound, BoundRemoteHandle)
 
 
 def test_client_remote_handle_serialize(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    hid = exchange.create_client()
+    agent_id = exchange.create_agent()
+    mailbox_id = exchange.create_client()
     handle: ClientRemoteHandle[Any]
-    with ClientRemoteHandle(exchange, aid, hid) as handle:
+    with ClientRemoteHandle(exchange, agent_id, mailbox_id) as handle:
         # Note: don't call pickle.dumps here because ThreadExchange
         # is not pickleable so we test __reduce__ directly.
         class_, args = handle.__reduce__()
@@ -155,22 +155,22 @@ def test_client_remote_handle_serialize(exchange: Exchange) -> None:
             assert isinstance(reconstructed, UnboundRemoteHandle)
             assert str(reconstructed) != str(handle)
             assert repr(reconstructed) != repr(handle)
-            assert reconstructed.aid == handle.aid
+            assert reconstructed.agent_id == handle.agent_id
 
 
 def test_client_remote_handle_bind(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
-    hid = exchange.create_client()
+    agent_id = exchange.create_agent()
+    mailbox_id = exchange.create_client()
     handle: ClientRemoteHandle[Any]
-    with ClientRemoteHandle(exchange, aid, hid) as handle:
+    with ClientRemoteHandle(exchange, agent_id, mailbox_id) as handle:
         assert handle.bind_as_client() is handle
         client_bound: ClientRemoteHandle[Any]
         with handle.bind_as_client(exchange.create_client()) as client_bound:
             assert client_bound is not handle
             assert isinstance(client_bound, ClientRemoteHandle)
-        agent_bound: AgentRemoteHandle[Any]
-        with handle.bind_to_agent(AgentIdentifier.new()) as agent_bound:
-            assert isinstance(agent_bound, AgentRemoteHandle)
+        agent_bound: BoundRemoteHandle[Any]
+        with handle.bind_to_mailbox(AgentIdentifier.new()) as agent_bound:
+            assert isinstance(agent_bound, BoundRemoteHandle)
 
 
 def test_client_remote_handle_log_bad_response(
@@ -181,11 +181,11 @@ def test_client_remote_handle_log_bad_response(
     handle: RemoteHandle[Any]
     with launcher.launch(behavior, exchange) as handle:
         client = handle.bind_as_client()
-        assert client.hid is not None
+        assert client.mailbox_id is not None
         # Should log but not crash
         client.exchange.send(
-            client.hid,
-            PingRequest(src=client.aid, dest=client.hid),
+            client.mailbox_id,
+            PingRequest(src=client.agent_id, dest=client.mailbox_id),
         )
         assert client.ping() > 0
 
@@ -194,13 +194,16 @@ def test_client_remote_handle_log_bad_response(
     'ignore:.*:pytest.PytestUnhandledThreadExceptionWarning',
 )
 def test_client_remote_handle_recv_thread_crash(exchange: Exchange) -> None:
-    aid = exchange.create_agent()
+    agent_id = exchange.create_agent()
 
     with mock.patch(
         'aeris.handle.ClientRemoteHandle._recv_responses',
         side_effect=Exception(),
     ):
-        handle: ClientRemoteHandle[Any] = ClientRemoteHandle(exchange, aid)
+        handle: ClientRemoteHandle[Any] = ClientRemoteHandle(
+            exchange,
+            agent_id,
+        )
 
     with pytest.raises(
         RuntimeError,
diff --git a/tests/manager_test.py b/tests/manager_test.py
new file mode 100644
index 0000000..c4dbe7c
--- /dev/null
+++ b/tests/manager_test.py
@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import time
+
+from aeris.exchange.thread import ThreadExchange
+from aeris.launcher.thread import ThreadLauncher
+from aeris.manager import Manager
+from aeris.message import PingRequest
+from aeris.message import PingResponse
+from testing.behavior import SleepBehavior
+from testing.constant import TEST_LOOP_SLEEP
+
+
+def test_protocol() -> None:
+    with Manager(
+        exchange=ThreadExchange(),
+        launcher=ThreadLauncher(),
+    ) as manager:
+        assert isinstance(repr(manager), str)
+        assert isinstance(str(manager), str)
+
+
+def test_basic_usage() -> None:
+    behavior = SleepBehavior(TEST_LOOP_SLEEP)
+    with Manager(
+        exchange=ThreadExchange(),
+        launcher=ThreadLauncher(),
+    ) as manager:
+        manager.launch(behavior)
+        manager.launch(behavior)
+
+        time.sleep(5 * TEST_LOOP_SLEEP)
+
+
+def test_reply_to_requests_with_error() -> None:
+    with Manager(
+        exchange=ThreadExchange(),
+        launcher=ThreadLauncher(),
+    ) as manager:
+        client_id = manager.exchange.create_client()
+        request = PingRequest(src=client_id, dest=manager.mailbox_id)
+        manager.exchange.send(request.dest, request)
+        response = manager.exchange.recv(client_id)
+        assert isinstance(response, PingResponse)
+        assert isinstance(response.exception, TypeError)
diff --git a/tests/multiplex_test.py b/tests/multiplex_test.py
new file mode 100644
index 0000000..b4e4dcf
--- /dev/null
+++ b/tests/multiplex_test.py
@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+from typing import Any
+from unittest import mock
+
+import pytest
+
+from aeris.exception import HandleClosedError
+from aeris.exception import MailboxClosedError
+from aeris.exchange.thread import ThreadExchange
+from aeris.handle import BoundRemoteHandle
+from aeris.handle import UnboundRemoteHandle
+from aeris.identifier import ClientIdentifier
+from aeris.message import PingRequest
+from aeris.message import PingResponse
+from aeris.multiplex import MailboxMultiplexer
+
+
+def test_protocol(exchange: ThreadExchange) -> None:
+    uid = exchange.create_client()
+    with MailboxMultiplexer(
+        uid,
+        exchange,
+        request_handler=lambda _: None,
+    ) as mutliplexer:
+        assert isinstance(repr(mutliplexer), str)
+        assert isinstance(str(mutliplexer), str)
+
+
+def test_listen_exit_on_mailbox_close(exchange: ThreadExchange) -> None:
+    uid = exchange.create_client()
+    exchange.close_mailbox(uid)
+    with MailboxMultiplexer(
+        uid,
+        exchange,
+        request_handler=lambda _: None,
+    ) as multiplexer:
+        # Should immediately return because the mailbox is closed.
+        multiplexer.listen()
+
+
+def test_bind_handle(exchange: ThreadExchange) -> None:
+    uid = exchange.create_client()
+    aid = exchange.create_agent()
+    unbound: UnboundRemoteHandle[Any] = UnboundRemoteHandle(exchange, aid)
+    with MailboxMultiplexer(
+        uid,
+        exchange,
+        request_handler=lambda _: None,
+    ) as multiplexer:
+        bound = multiplexer.bind(unbound)
+        assert isinstance(bound, BoundRemoteHandle)
+        multiplexer.close_bound_handles()
+        with pytest.raises(HandleClosedError):
+            bound.ping()
+
+
+def test_bind_duplicate_handle(exchange: ThreadExchange) -> None:
+    uid = exchange.create_client()
+    aid = exchange.create_agent()
+    unbound: UnboundRemoteHandle[Any] = UnboundRemoteHandle(exchange, aid)
+    with MailboxMultiplexer(
+        uid,
+        exchange,
+        request_handler=lambda _: None,
+    ) as multiplexer:
+        multiplexer.bind(unbound)
+        multiplexer.bind(unbound)
+
+
+def test_request_message_handler(exchange: ThreadExchange) -> None:
+    uid = exchange.create_client()
+    handler = mock.MagicMock()
+    request = PingRequest(src=ClientIdentifier.new(), dest=uid)
+
+    with mock.patch.object(
+        exchange,
+        'recv',
+        side_effect=(request, MailboxClosedError(uid)),
+    ):
+        with MailboxMultiplexer(
+            uid,
+            exchange,
+            request_handler=handler,
+        ) as multiplexer:
+            multiplexer.listen()
+
+        handler.assert_called_once()
+
+
+def test_response_message_handler(exchange: ThreadExchange) -> None:
+    uid = exchange.create_client()
+    aid = exchange.create_agent()
+    unbound: UnboundRemoteHandle[Any] = UnboundRemoteHandle(exchange, aid)
+
+    with MailboxMultiplexer(
+        uid,
+        exchange,
+        request_handler=lambda _: None,
+    ) as multiplexer:
+        bound = multiplexer.bind(unbound)
+        response = PingResponse(src=aid, dest=uid, label=bound.handle_id)
+        with mock.patch.object(bound, '_process_response') as mocked:
+            with mock.patch.object(
+                exchange,
+                'recv',
+                side_effect=(response, MailboxClosedError(uid)),
+            ):
+                multiplexer.listen()
+            mocked.assert_called_once()
+
+
+def test_response_message_handler_bad_src(exchange: ThreadExchange) -> None:
+    uid = exchange.create_client()
+    aid = exchange.create_agent()
+    unbound: UnboundRemoteHandle[Any] = UnboundRemoteHandle(exchange, aid)
+    response = PingResponse(src=ClientIdentifier.new(), dest=uid)
+
+    with mock.patch.object(
+        exchange,
+        'recv',
+        side_effect=(response, MailboxClosedError(uid)),
+    ):
+        with MailboxMultiplexer(
+            uid,
+            exchange,
+            request_handler=lambda _: None,
+        ) as multiplexer:
+            bound = multiplexer.bind(unbound)
+            with mock.patch.object(bound, '_process_response') as mocked:
+                multiplexer.listen()
+                # Message handler will not have a valid handle to pass
+                # the message to so it just logs an exception and moves on.
+                mocked.assert_not_called()