academy.handle

academy/handle.py

Handle

Handle(
    agent_id: AgentId[AgentT_co] | None = None,
    *,
    exchange: ExchangeClient[Any] | None = None,
    ignore_context: bool = False,
    request_serializer: SerializationStrategy | None = None,
    result_serializer: SerializationStrategy | None = None,
    exception_serializer: (
        SerializationStrategy | None
    ) = None,
    polling_interval: float = 60,
    reject_on_inactive: bool = False
)

Bases: Generic[AgentT_co]

Handle to a remote agent.

Internally, handles use an ExchangeClient to send requests to and receive responses from the remote agent. By default the correct exchange client is inferred from the context using a context variable (specifically, the academy.handle.exchange_context variable). This allows the same handle to be used in different contexts, automatically using the correct client to send messages.

When a handle is used in contexts that have not configured the exchange client (such as outside of an agent runtime or Manager), a default exchange can be provided via the exchange argument. For advanced usage, the ignore_context flag will cause the handle to only use the exchange argument no matter what the current context is.

Note

The exchange argument will not be included when a handle is pickled. Thus, unpickled handles must be used in a context that configures an exchange client.

Parameters:

• agent_id (AgentId[AgentT_co] | None, default: None ) –

  ID of the remote agent, or None to defer binding (used by batch.queue()).

• exchange (ExchangeClient[Any] | None, default: None ) –

  A default exchange client to be used if an exchange client is not configured in the current context.

• ignore_context (bool, default: False ) –

  Ignore the current context and force use of exchange for communication.

• request_serializer (SerializationStrategy | None, default: None ) –

  Strategy used to serialize arguments. If None, use the academy.serialize.default_serializer

• result_serializer (SerializationStrategy | None, default: None ) –

  Strategy used to serialize results. If false-y, use the same strategy as the request serializer.

• exception_serializer (SerializationStrategy | None, default: None ) –

  Strategy used to serialize results. If false-y, use the same strategy as the result serializer.

• polling_interval (float, default: 60 ) –

  Interval to poll exchange to see if agent is active.

• reject_on_inactive (bool, default: False ) –

  Return an error when target agent is inactive.

Raises:

• ValueError –

  If ignore_context=True but exchange is not provided.

Source code in academy/handle.py

def __init__(  # noqa: PLR0913
    self,
    agent_id: AgentId[AgentT_co] | None = None,
    *,
    exchange: ExchangeClient[Any] | None = None,
    ignore_context: bool = False,
    request_serializer: SerializationStrategy | None = None,
    result_serializer: SerializationStrategy | None = None,
    exception_serializer: SerializationStrategy | None = None,
    polling_interval: float = 60,
    reject_on_inactive: bool = False,
) -> None:
    self._agent_id: AgentId[AgentT_co] | None = agent_id
    self._exchange = exchange
    self.request_serializer = request_serializer
    self.result_serializer = result_serializer
    self.exception_serializer = exception_serializer
    self._registered_exchanges: WeakSet[ExchangeClient[Any]] = WeakSet()
    self.ignore_context = ignore_context

    if ignore_context and not exchange:
        raise ValueError(
            'Cannot initialize handle with ignore_context=True '
            'and no explicit exchange.',
        )

    # 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._pending_response_futures: dict[
        uuid.UUID,
        asyncio.Future[Any],
    ] = {}
    self._pending_actions: set[uuid.UUID] = set()
    self._pending_actions_lock = asyncio.Lock()

    if self._exchange is not None:
        self._register_with_exchange(self._exchange)

    # _agent_status is the private variable to keep track of the mailbox
    # status in the polling loop --- it is not always current (based on
    # when the last poll was), and is only updated if reject_on_inactive
    # is true. For live mailbox status use `agent_status`
    self._agent_status: MailboxStatus | None = None
    self.polling_interval = polling_interval
    self._reject_on_inactive: bool = False
    self.reject_on_inactive = reject_on_inactive

agent_id property writable

agent_id: AgentId[AgentT_co]

ID of the remote agent.

Raises:

• RuntimeError –

  If the handle is not bound.

reject_on_inactive property writable

reject_on_inactive: bool

Reject actions when agent is inactive.

exchange property

exchange: ExchangeClient[Any]

Exchange client used to send messages.

Returns:

• ExchangeClient[Any] –

  Exchange client.

Raises:

• ExchangeClientNotFoundError –

  If no exchange client is set in the current context nor was one provided to the handle.

__setstate__

__setstate__(state: dict[str, Any]) -> None

Set state.

This is necessary for unpickling to not treat set state as a remote action.

Source code in academy/handle.py

def __setstate__(self, state: dict[str, Any]) -> None:
    """Set state.

    This is necessary for unpickling to not treat set state as a
    remote action.
    """
    reject_on_inactive: bool = state.pop('reject_on_inactive', False)
    self.__dict__.update(state)
    self.reject_on_inactive = reject_on_inactive

agent_stats async

agent_stats() -> AgentStats

Return live exchange-level metrics for the remote agent.

Source code in academy/handle.py

async def agent_stats(self) -> AgentStats:
    """Return live exchange-level metrics for the remote agent."""
    return await self.exchange.agent_stats(self.agent_id)

agent_status async

agent_status() -> MailboxStatus

Return live status of the agent mailbox.

Source code in academy/handle.py

async def agent_status(self) -> MailboxStatus:
    """Return live status of the agent mailbox."""
    status = await self.exchange.status(self.agent_id)
    if self._reject_on_inactive:
        # To avoid a situation where agent_status shows returns
        # active, and actions are rejected with a stale status
        await self._update_agent_status(status)

    return status

action async

action(action: str, /, *args: Any, **kwargs: Any) -> R

Invoke an action on the agent.

Parameters:

• action (str) –

  Action to invoke.

• args (Any, default: () ) –

  Positional arguments for the action.

• kwargs (Any, default: {} ) –

  Keywords arguments for the action.

Returns:

• R –

  Result of the action.

Raises:

• AgentTerminatedError –

  If the agent's mailbox was closed. This typically indicates the agent shutdown for another reason (it self terminated or via another handle).

• AgentInactiveError –

  If the agent is inactive (missed heartbeats) and handle is configured to reject actions when agent is inactive (reject_on_inactive=True)

• Exception –

  Any exception raised by the action.

Source code in academy/handle.py

async def action(self, action: str, /, *args: Any, **kwargs: Any) -> R:
    """Invoke an action on the agent.

    Args:
        action: Action to invoke.
        args: Positional arguments for the action.
        kwargs: Keywords arguments for the action.

    Returns:
        Result of the action.

    Raises:
        AgentTerminatedError: If the agent's mailbox was closed. This
            typically indicates the agent shutdown for another reason
            (it self terminated or via another handle).
        AgentInactiveError: If the agent is inactive (missed heartbeats)
            and handle is configured to reject actions when agent is
            inactive (reject_on_inactive=True)
        Exception: Any exception raised by the action.
    """
    tag_id = uuid.uuid4()
    invocation_extra = {
        'academy.action': action,
        'academy.action_tag': tag_id,
    }

    logger.debug(
        'Invoking action %s with tag id %s',
        action,
        tag_id,
        extra=invocation_extra
        | {
            'academy.action_state': 'start',
            'academy.action_args': args,
            'academy.action_kwargs': kwargs,
            'academy.agent_id': self.agent_id,
        },
    )
    if self.reject_on_inactive:
        await self._check_status()

    exchange = self.exchange
    self._register_with_exchange(exchange)
    serialization = self.request_serializer or default_serializer.get()

    request = Message.create(
        src=exchange.client_id,
        dest=self.agent_id,
        label=self.handle_id,
        tag=tag_id,
        body=ActionRequest(
            action=action,
            pargs=args,
            kargs=kwargs,
            serialization=serialization,
            result_serialization=self.result_serializer,
            exception_serialization=self.exception_serializer,
        ),
    )
    loop = asyncio.get_running_loop()
    future: asyncio.Future[Message[Response]] = loop.create_future()
    self._pending_response_futures[request.tag] = future
    async with self._pending_actions_lock:
        self._pending_actions.add(request.tag)

    try:
        logger.debug(
            'Sending action request from %s to %s (action=%r)',
            exchange.client_id,
            self.agent_id,
            action,
            extra=request.log_extra()
            | invocation_extra
            | {'academy.action_state': 'sending'},
        )
        await self.exchange.send(request)
        logger.debug(
            'Waiting for result of action %s with tag id %s',
            action,
            tag_id,
            extra=invocation_extra
            | {
                'academy.action_state': 'waiting',
            },
        )
        await future
    except asyncio.CancelledError:
        cancel_request = Message.create(
            src=exchange.client_id,
            dest=self.agent_id,
            label=self.handle_id,
            body=CancelRequest(target_tag=request.tag),
        )
        logger.debug(
            'Cancelling action %s with tag id %s',
            action,
            tag_id,
            extra=cancel_request.log_extra()
            | invocation_extra
            | {
                'academy.action_state': 'cancelled',
            },
        )
        cancel_future: asyncio.Future[None] = loop.create_future()
        self._pending_response_futures[cancel_request.tag] = cancel_future
        await self.exchange.send(cancel_request)
        raise

    assert future.done()
    assert future.exception() is None
    message = future.result()
    body = message.get_body()

    if isinstance(body, ActionResponse):
        result = body.get_result()
        logger.debug(
            'Successfully completed action %s with tag id %s',
            action,
            tag_id,
            extra=invocation_extra
            | {
                'academy.action_state': 'success',
                'academy.result': result,
                'academy.agent_id': self.agent_id,
            },
        )
        return result
    elif isinstance(body, ErrorResponse):
        exception = body.get_exception()
        logger.debug(
            'Completed action %s with tag id %s with exception',
            action,
            tag_id,
            extra=invocation_extra
            | {
                'academy.action_state': 'exception',
            },
        )
        raise exception

    raise RuntimeError(
        'Invalid response received from action request.',
    )  # pragma: no cover

ping async

ping(*, timeout: float | None = None) -> float

Ping the agent.

Ping the agent and wait to get a response.

Parameters:

• timeout (float | None, default: None ) –

  Optional timeout in seconds to wait for the response.

Returns:

• float –

  Round-trip time in seconds.

Raises:

• AgentTerminatedError –

  If the agent's mailbox was closed. This typically indicates the agent shutdown for another reason (it self terminated or via another handle).

• TimeoutError –

  If the timeout is exceeded.

Source code in academy/handle.py

async def ping(self, *, timeout: float | None = None) -> float:
    """Ping the agent.

    Ping the agent and wait to get a response.

    Args:
        timeout: Optional timeout in seconds to wait for the response.

    Returns:
        Round-trip time in seconds.

    Raises:
        AgentTerminatedError: If the agent's mailbox was closed. This
            typically indicates the agent shutdown for another reason
            (it self terminated or via another handle).
        TimeoutError: If the timeout is exceeded.
    """
    exchange = self.exchange
    self._register_with_exchange(exchange)

    request = Message.create(
        src=exchange.client_id,
        dest=self.agent_id,
        label=self.handle_id,
        body=PingRequest(),
    )
    loop = asyncio.get_running_loop()
    future: asyncio.Future[Message[Response]] = loop.create_future()
    self._pending_response_futures[request.tag] = future
    start = time.perf_counter()
    await self.exchange.send(request)
    logger.debug(
        'Sent ping from %s to %s',
        exchange.client_id,
        self.agent_id,
        extra=request.log_extra(),
    )

    await asyncio.wait_for(future, timeout)

    assert future.done()
    message = future.result()
    body = message.get_body()

    if isinstance(body, ErrorResponse):
        raise body.get_exception()

    if self.reject_on_inactive:
        # A successful ping indicates active regardless of
        # heartbeat, so we can use ping to wait for an agent
        # to become active.
        await self._update_agent_status(MailboxStatus.ACTIVE)

    elapsed = time.perf_counter() - start
    logger.debug(
        'Received ping from %s to %s in %.1f ms',
        exchange.client_id,
        self.agent_id,
        elapsed * 1000,
        extra=request.log_extra()
        | {
            'academy.ping_time_s': elapsed,
        },
    )
    return elapsed

shutdown async

shutdown(*, terminate: bool | None = None) -> None

Instruct the agent to shutdown.

This is non-blocking and will only send the message.

Parameters:

• terminate (bool | None, default: None ) –

  Override the termination behavior of the agent defined in the RuntimeConfig.

Raises:

• AgentTerminatedError –

  If the agent's mailbox was closed. This typically indicates the agent shutdown for another reason (it self terminated or via another handle).

Source code in academy/handle.py

async def shutdown(self, *, terminate: bool | None = None) -> None:
    """Instruct the agent to shutdown.

    This is non-blocking and will only send the message.

    Args:
        terminate: Override the termination behavior of the agent defined
            in the [`RuntimeConfig`][academy.runtime.RuntimeConfig].

    Raises:
        AgentTerminatedError: If the agent's mailbox was closed. This
            typically indicates the agent shutdown for another reason
            (it self terminated or via another handle).
    """
    exchange = self.exchange
    self._register_with_exchange(exchange)

    request = Message.create(
        src=exchange.client_id,
        dest=self.agent_id,
        label=self.handle_id,
        body=ShutdownRequest(terminate=terminate),
    )

    loop = asyncio.get_running_loop()
    future: asyncio.Future[Message[Response]] = loop.create_future()
    self._pending_response_futures[request.tag] = future
    await self.exchange.send(request)

    logger.debug(
        'Sent shutdown request from %s to %s',
        exchange.client_id,
        self.agent_id,
        extra=request.log_extra(),
    )

    future.add_done_callback(self._shutdown_callback)

ProxyHandle

ProxyHandle(agent: AgentT)

Bases: Handle[AgentT]

Proxy handle.

A proxy handle is thin wrapper around an Agent instance that is useful for testing agents that are initialized with a handle to another agent without needing to spawn agents. This wrapper invokes actions synchronously.

Source code in academy/handle.py

def __init__(self, agent: AgentT) -> None:
    self.agent = agent
    self._agent_id: AgentId[AgentT] = AgentId.new()
    self._agent_closed = False
    self._reject_on_inactive = False

reject_on_inactive property writable

reject_on_inactive: bool

Reject actions when agent is inactive.

agent_id property writable

agent_id: AgentId[AgentT_co]

ID of the remote agent.

Raises:

• RuntimeError –

  If the handle is not bound.

exchange property

exchange: ExchangeClient[Any]

Exchange client used to send messages.

Returns:

• ExchangeClient[Any] –

  Exchange client.

Raises:

• ExchangeClientNotFoundError –

  If no exchange client is set in the current context nor was one provided to the handle.

action async

action(action: str, /, *args: Any, **kwargs: Any) -> R

Invoke an action on the agent.

Parameters:

• action (str) –

  Action to invoke.

• args (Any, default: () ) –

  Positional arguments for the action.

• kwargs (Any, default: {} ) –

  Keywords arguments for the action.

Returns:

• R –

  Result of the action.

Raises:

• AgentTerminatedError –

  If the agent's mailbox was closed. This typically indicates the agent shutdown for another reason (it self terminated or via another handle).

• Exception –

  Any exception raised by the action.

Source code in academy/handle.py

async def action(self, action: str, /, *args: Any, **kwargs: Any) -> R:
    """Invoke an action on the agent.

    Args:
        action: Action to invoke.
        args: Positional arguments for the action.
        kwargs: Keywords arguments for the action.

    Returns:
        Result of the action.

    Raises:
        AgentTerminatedError: If the agent's mailbox was closed. This
            typically indicates the agent shutdown for another reason
            (it self terminated or via another handle).
        Exception: Any exception raised by the action.
    """
    if self._agent_closed:
        raise AgentTerminatedError(self.agent_id)

    method = getattr(self.agent, action)
    return await method(*args, **kwargs)

ping async

ping(*, timeout: float | None = None) -> float

Ping the agent.

This is a no-op for proxy handles and returns 0 latency.

Parameters:

• timeout (float | None, default: None ) –

  Optional timeout in seconds to wait for the response.

Returns:

• float –

  Round-trip time in seconds.

Raises:

• AgentTerminatedError –

  If the agent's mailbox was closed. This typically indicates the agent shutdown for another reason (it self terminated or via another handle).

• TimeoutError –

  If the timeout is exceeded.

Source code in academy/handle.py

async def ping(self, *, timeout: float | None = None) -> float:
    """Ping the agent.

    This is a no-op for proxy handles and returns 0 latency.

    Args:
        timeout: Optional timeout in seconds to wait for the response.

    Returns:
        Round-trip time in seconds.

    Raises:
        AgentTerminatedError: If the agent's mailbox was closed. This
            typically indicates the agent shutdown for another reason
            (it self terminated or via another handle).
        TimeoutError: If the timeout is exceeded.
    """
    if self._agent_closed:
        raise AgentTerminatedError(self.agent_id)
    return 0

shutdown async

shutdown(*, terminate: bool | None = None) -> None

Instruct the agent to shutdown.

This is non-blocking and will only send the message.

Parameters:

• terminate (bool | None, default: None ) –

  Override the termination behavior of the agent defined in the RuntimeConfig.

Raises:

• AgentTerminatedError –

  If the agent's mailbox was closed. This typically indicates the agent shutdown for another reason (it self terminated or via another handle).

Source code in academy/handle.py

async def shutdown(self, *, terminate: bool | None = None) -> None:
    """Instruct the agent to shutdown.

    This is non-blocking and will only send the message.

    Args:
        terminate: Override the termination behavior of the agent defined
            in the [`RuntimeConfig`][academy.runtime.RuntimeConfig].

    Raises:
        AgentTerminatedError: If the agent's mailbox was closed. This
            typically indicates the agent shutdown for another reason
            (it self terminated or via another handle).
    """
    if self._agent_closed:
        raise AgentTerminatedError(self.agent_id)
    self._agent_closed = True if terminate is None else terminate

__setstate__

__setstate__(state: dict[str, Any]) -> None

Set state.

This is necessary for unpickling to not treat set state as a remote action.

Source code in academy/handle.py

def __setstate__(self, state: dict[str, Any]) -> None:
    """Set state.

    This is necessary for unpickling to not treat set state as a
    remote action.
    """
    reject_on_inactive: bool = state.pop('reject_on_inactive', False)
    self.__dict__.update(state)
    self.reject_on_inactive = reject_on_inactive

agent_stats async

agent_stats() -> AgentStats

Return live exchange-level metrics for the remote agent.

Source code in academy/handle.py

async def agent_stats(self) -> AgentStats:
    """Return live exchange-level metrics for the remote agent."""
    return await self.exchange.agent_stats(self.agent_id)

agent_status async

agent_status() -> MailboxStatus

Return live status of the agent mailbox.

Source code in academy/handle.py

async def agent_status(self) -> MailboxStatus:
    """Return live status of the agent mailbox."""
    status = await self.exchange.status(self.agent_id)
    if self._reject_on_inactive:
        # To avoid a situation where agent_status shows returns
        # active, and actions are rejected with a stale status
        await self._update_agent_status(status)

    return status