Skip to content

academy.exchange.client

ExchangeClient 

ExchangeClient(transport: ExchangeTransportT)

Bases: ABC, Generic[ExchangeTransportT]

Base exchange client.

Warning

Exchange clients should only be created via ExchangeFactory.create_agent_client() or ExchangeFactory.create_user_client()!

Parameters:

Source code in academy/exchange/client.py 
66
67
68
69
70
71
72
73
74
75
def __init__(
    self,
    transport: ExchangeTransportT,
) -> None:
    self._transport = transport
    self._handles: WeakValueDictionary[uuid.UUID, Handle[Any]] = (
        WeakValueDictionary()
    )
    self._close_lock = asyncio.Lock()
    self._closed = False

client_id abstractmethod property 

client_id: EntityId

Client ID as registered with the exchange.

close abstractmethod async 

close() -> None

Close the transport.

Source code in academy/exchange/client.py 
102
103
104
105
@abc.abstractmethod
async def close(self) -> None:
    """Close the transport."""
    ...

discover async 

discover(
    agent: type[Agent], *, allow_subclasses: bool = True
) -> tuple[AgentId[Any], ...]

Discover peer agents with a given agent.

Parameters:

  • agent (type[Agent]) –

    Agent type of interest.

  • allow_subclasses (bool, default: True ) –

    Return agents implementing subclasses of the agent.

Returns:

Source code in academy/exchange/client.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def discover(
    self,
    agent: type[Agent],
    *,
    allow_subclasses: bool = True,
) -> tuple[AgentId[Any], ...]:
    """Discover peer agents with a given agent.

    Args:
        agent: Agent type of interest.
        allow_subclasses: Return agents implementing subclasses of the
            agent.

    Returns:
        Tuple of agent IDs implementing the agent.
    """
    return await self._transport.discover(
        agent,
        allow_subclasses=allow_subclasses,
    )

factory 

factory() -> ExchangeFactory[ExchangeTransportT]

Get an exchange factory.

Source code in academy/exchange/client.py 
128
129
130
def factory(self) -> ExchangeFactory[ExchangeTransportT]:
    """Get an exchange factory."""
    return self._transport.factory()

register_handle 

register_handle(handle: Handle[AgentT]) -> None

Register an existing handle to receive messages.

Parameters:

Source code in academy/exchange/client.py 
132
133
134
135
136
137
138
def register_handle(self, handle: Handle[AgentT]) -> None:
    """Register an existing handle to receive messages.

    Args:
        handle: Handle to register.
    """
    self._handles[handle.handle_id] = handle

register_agent async 

register_agent(
    agent: type[AgentT], *, name: str | None = None
) -> AgentRegistration[AgentT]

Register a new agent and associated mailbox with the exchange.

Parameters:

  • agent (type[AgentT]) –

    Agent type of the agent.

  • name (str | None, default: None ) –

    Optional display name for the agent.

Returns:

Source code in academy/exchange/client.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def register_agent(
    self,
    agent: type[AgentT],
    *,
    name: str | None = None,
) -> AgentRegistration[AgentT]:
    """Register a new agent and associated mailbox with the exchange.

    Args:
        agent: Agent type of the agent.
        name: Optional display name for the agent.

    Returns:
        Agent registration info.
    """
    registration = await self._transport.register_agent(
        agent,
        name=name,
    )
    logger.info('Registered %s in exchange', registration.agent_id)
    return registration

send async 

send(message: Message[Any]) -> None

Send a message to a mailbox.

Parameters:

Raises:

  • BadEntityIdError –

    If a mailbox for message.dest does not exist.

  • MailboxTerminatedError –

    If the mailbox was closed.

Source code in academy/exchange/client.py 
162
163
164
165
166
167
168
169
170
171
172
173
async def send(self, message: Message[Any]) -> None:
    """Send a message to a mailbox.

    Args:
        message: Message to send.

    Raises:
        BadEntityIdError: If a mailbox for `message.dest` does not exist.
        MailboxTerminatedError: If the mailbox was closed.
    """
    await self._transport.send(message)
    logger.debug('Sent %s to %s', type(message).__name__, message.dest)

status async 

status(uid: EntityId) -> MailboxStatus

Check the status of a mailbox in the exchange.

Parameters:

  • uid (EntityId) –

    Entity identifier of the mailbox to check.

Source code in academy/exchange/client.py 
175
176
177
178
179
180
181
async def status(self, uid: EntityId) -> MailboxStatus:
    """Check the status of a mailbox in the exchange.

    Args:
        uid: Entity identifier of the mailbox to check.
    """
    return await self._transport.status(uid)

terminate async 

terminate(uid: EntityId) -> None

Terminate a mailbox in the exchange.

Terminating a mailbox means that the corresponding entity will no longer be able to receive messages.

Note

This method is a no-op if the mailbox does not exist.

Parameters:

  • uid (EntityId) –

    Entity identifier of the mailbox to close.

Source code in academy/exchange/client.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
async def terminate(self, uid: EntityId) -> None:
    """Terminate a mailbox in the exchange.

    Terminating a mailbox means that the corresponding entity will no
    longer be able to receive messages.

    Note:
        This method is a no-op if the mailbox does not exist.

    Args:
        uid: Entity identifier of the mailbox to close.
    """
    await self._transport.terminate(uid)

AgentExchangeClient 

AgentExchangeClient(
    agent_id: AgentId[AgentT],
    transport: ExchangeTransportT,
    request_handler: RequestHandler[RequestT_co],
)

Bases: ExchangeClient[ExchangeTransportT], Generic[AgentT, ExchangeTransportT]

Agent exchange client.

Warning

Agent exchange clients should only be created via ExchangeFactory.create_agent_client()!

Parameters:

  • agent_id (AgentId[AgentT]) –

    Agent ID.

  • transport (ExchangeTransportT) –

    Exchange transport bound to agent_id.

  • request_handler (RequestHandler[RequestT_co]) –

    Request handler of the agent that will be called for each message received to this agent's mailbox. start_listener: Start a message listener thread.

Source code in academy/exchange/client.py 
233
234
235
236
237
238
239
240
241
def __init__(
    self,
    agent_id: AgentId[AgentT],
    transport: ExchangeTransportT,
    request_handler: RequestHandler[RequestT_co],
) -> None:
    super().__init__(transport)
    self._agent_id = agent_id
    self._request_handler = request_handler

client_id property 

client_id: AgentId[AgentT]

Agent ID of the client.

close async 

close() -> None

Close the user client.

This closes the underlying exchange transport and all handles created by this client. The agent's mailbox will not be terminated so the agent can be started again later.

Source code in academy/exchange/client.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
async def close(self) -> None:
    """Close the user client.

    This closes the underlying exchange transport and all handles created
    by this client. The agent's mailbox will not be terminated so the agent
    can be started again later.
    """
    async with self._close_lock:
        if self._closed:
            return

        await self._transport.close()
        self._closed = True
        logger.info('Closed exchange client for %s', self.client_id)

discover async 

discover(
    agent: type[Agent], *, allow_subclasses: bool = True
) -> tuple[AgentId[Any], ...]

Discover peer agents with a given agent.

Parameters:

  • agent (type[Agent]) –

    Agent type of interest.

  • allow_subclasses (bool, default: True ) –

    Return agents implementing subclasses of the agent.

Returns:

Source code in academy/exchange/client.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def discover(
    self,
    agent: type[Agent],
    *,
    allow_subclasses: bool = True,
) -> tuple[AgentId[Any], ...]:
    """Discover peer agents with a given agent.

    Args:
        agent: Agent type of interest.
        allow_subclasses: Return agents implementing subclasses of the
            agent.

    Returns:
        Tuple of agent IDs implementing the agent.
    """
    return await self._transport.discover(
        agent,
        allow_subclasses=allow_subclasses,
    )

factory 

factory() -> ExchangeFactory[ExchangeTransportT]

Get an exchange factory.

Source code in academy/exchange/client.py 
128
129
130
def factory(self) -> ExchangeFactory[ExchangeTransportT]:
    """Get an exchange factory."""
    return self._transport.factory()

register_handle 

register_handle(handle: Handle[AgentT]) -> None

Register an existing handle to receive messages.

Parameters:

Source code in academy/exchange/client.py 
132
133
134
135
136
137
138
def register_handle(self, handle: Handle[AgentT]) -> None:
    """Register an existing handle to receive messages.

    Args:
        handle: Handle to register.
    """
    self._handles[handle.handle_id] = handle

register_agent async 

register_agent(
    agent: type[AgentT], *, name: str | None = None
) -> AgentRegistration[AgentT]

Register a new agent and associated mailbox with the exchange.

Parameters:

  • agent (type[AgentT]) –

    Agent type of the agent.

  • name (str | None, default: None ) –

    Optional display name for the agent.

Returns:

Source code in academy/exchange/client.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def register_agent(
    self,
    agent: type[AgentT],
    *,
    name: str | None = None,
) -> AgentRegistration[AgentT]:
    """Register a new agent and associated mailbox with the exchange.

    Args:
        agent: Agent type of the agent.
        name: Optional display name for the agent.

    Returns:
        Agent registration info.
    """
    registration = await self._transport.register_agent(
        agent,
        name=name,
    )
    logger.info('Registered %s in exchange', registration.agent_id)
    return registration

send async 

send(message: Message[Any]) -> None

Send a message to a mailbox.

Parameters:

Raises:

  • BadEntityIdError –

    If a mailbox for message.dest does not exist.

  • MailboxTerminatedError –

    If the mailbox was closed.

Source code in academy/exchange/client.py 
162
163
164
165
166
167
168
169
170
171
172
173
async def send(self, message: Message[Any]) -> None:
    """Send a message to a mailbox.

    Args:
        message: Message to send.

    Raises:
        BadEntityIdError: If a mailbox for `message.dest` does not exist.
        MailboxTerminatedError: If the mailbox was closed.
    """
    await self._transport.send(message)
    logger.debug('Sent %s to %s', type(message).__name__, message.dest)

status async 

status(uid: EntityId) -> MailboxStatus

Check the status of a mailbox in the exchange.

Parameters:

  • uid (EntityId) –

    Entity identifier of the mailbox to check.

Source code in academy/exchange/client.py 
175
176
177
178
179
180
181
async def status(self, uid: EntityId) -> MailboxStatus:
    """Check the status of a mailbox in the exchange.

    Args:
        uid: Entity identifier of the mailbox to check.
    """
    return await self._transport.status(uid)

terminate async 

terminate(uid: EntityId) -> None

Terminate a mailbox in the exchange.

Terminating a mailbox means that the corresponding entity will no longer be able to receive messages.

Note

This method is a no-op if the mailbox does not exist.

Parameters:

  • uid (EntityId) –

    Entity identifier of the mailbox to close.

Source code in academy/exchange/client.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
async def terminate(self, uid: EntityId) -> None:
    """Terminate a mailbox in the exchange.

    Terminating a mailbox means that the corresponding entity will no
    longer be able to receive messages.

    Note:
        This method is a no-op if the mailbox does not exist.

    Args:
        uid: Entity identifier of the mailbox to close.
    """
    await self._transport.terminate(uid)

UserExchangeClient 

UserExchangeClient(
    user_id: UserId,
    transport: ExchangeTransportT,
    *,
    start_listener: bool = True
)

Bases: ExchangeClient[ExchangeTransportT]

User exchange client.

Warning

User exchange clients should only be created via ExchangeFactory.create_user_client()!

Parameters:

  • user_id (UserId) –

    User ID.

  • transport (ExchangeTransportT) –

    Exchange transport bound to user_id.

  • start_listener (bool, default: True ) –

    Start a message listener thread.

Source code in academy/exchange/client.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def __init__(
    self,
    user_id: UserId,
    transport: ExchangeTransportT,
    *,
    start_listener: bool = True,
) -> None:
    super().__init__(transport)
    self._user_id = user_id
    self._listener_task: asyncio.Task[None] | None = None
    if start_listener:
        self._listener_task = asyncio.create_task(
            self._listen_for_messages(),
            name=f'user-exchange-listener-{self.client_id}',
        )

client_id property 

client_id: UserId

User ID of the client.

close async 

close() -> None

Close the user client.

This terminates the user's mailbox, closes the underlying exchange transport.

Source code in academy/exchange/client.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
async def close(self) -> None:
    """Close the user client.

    This terminates the user's mailbox, closes the underlying exchange
    transport.
    """
    async with self._close_lock:
        if self._closed:
            return

        await self._transport.terminate(self.client_id)
        logger.info(f'Terminated mailbox for {self.client_id}')
        await self._stop_listener_task()
        await self._transport.close()
        self._closed = True
        logger.info('Closed exchange client for %s', self.client_id)

discover async 

discover(
    agent: type[Agent], *, allow_subclasses: bool = True
) -> tuple[AgentId[Any], ...]

Discover peer agents with a given agent.

Parameters:

  • agent (type[Agent]) –

    Agent type of interest.

  • allow_subclasses (bool, default: True ) –

    Return agents implementing subclasses of the agent.

Returns:

Source code in academy/exchange/client.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def discover(
    self,
    agent: type[Agent],
    *,
    allow_subclasses: bool = True,
) -> tuple[AgentId[Any], ...]:
    """Discover peer agents with a given agent.

    Args:
        agent: Agent type of interest.
        allow_subclasses: Return agents implementing subclasses of the
            agent.

    Returns:
        Tuple of agent IDs implementing the agent.
    """
    return await self._transport.discover(
        agent,
        allow_subclasses=allow_subclasses,
    )

factory 

factory() -> ExchangeFactory[ExchangeTransportT]

Get an exchange factory.

Source code in academy/exchange/client.py 
128
129
130
def factory(self) -> ExchangeFactory[ExchangeTransportT]:
    """Get an exchange factory."""
    return self._transport.factory()

register_handle 

register_handle(handle: Handle[AgentT]) -> None

Register an existing handle to receive messages.

Parameters:

Source code in academy/exchange/client.py 
132
133
134
135
136
137
138
def register_handle(self, handle: Handle[AgentT]) -> None:
    """Register an existing handle to receive messages.

    Args:
        handle: Handle to register.
    """
    self._handles[handle.handle_id] = handle

register_agent async 

register_agent(
    agent: type[AgentT], *, name: str | None = None
) -> AgentRegistration[AgentT]

Register a new agent and associated mailbox with the exchange.

Parameters:

  • agent (type[AgentT]) –

    Agent type of the agent.

  • name (str | None, default: None ) –

    Optional display name for the agent.

Returns:

Source code in academy/exchange/client.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def register_agent(
    self,
    agent: type[AgentT],
    *,
    name: str | None = None,
) -> AgentRegistration[AgentT]:
    """Register a new agent and associated mailbox with the exchange.

    Args:
        agent: Agent type of the agent.
        name: Optional display name for the agent.

    Returns:
        Agent registration info.
    """
    registration = await self._transport.register_agent(
        agent,
        name=name,
    )
    logger.info('Registered %s in exchange', registration.agent_id)
    return registration

send async 

send(message: Message[Any]) -> None

Send a message to a mailbox.

Parameters:

Raises:

  • BadEntityIdError –

    If a mailbox for message.dest does not exist.

  • MailboxTerminatedError –

    If the mailbox was closed.

Source code in academy/exchange/client.py 
162
163
164
165
166
167
168
169
170
171
172
173
async def send(self, message: Message[Any]) -> None:
    """Send a message to a mailbox.

    Args:
        message: Message to send.

    Raises:
        BadEntityIdError: If a mailbox for `message.dest` does not exist.
        MailboxTerminatedError: If the mailbox was closed.
    """
    await self._transport.send(message)
    logger.debug('Sent %s to %s', type(message).__name__, message.dest)

status async 

status(uid: EntityId) -> MailboxStatus

Check the status of a mailbox in the exchange.

Parameters:

  • uid (EntityId) –

    Entity identifier of the mailbox to check.

Source code in academy/exchange/client.py 
175
176
177
178
179
180
181
async def status(self, uid: EntityId) -> MailboxStatus:
    """Check the status of a mailbox in the exchange.

    Args:
        uid: Entity identifier of the mailbox to check.
    """
    return await self._transport.status(uid)

terminate async 

terminate(uid: EntityId) -> None

Terminate a mailbox in the exchange.

Terminating a mailbox means that the corresponding entity will no longer be able to receive messages.

Note

This method is a no-op if the mailbox does not exist.

Parameters:

  • uid (EntityId) –

    Entity identifier of the mailbox to close.

Source code in academy/exchange/client.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
async def terminate(self, uid: EntityId) -> None:
    """Terminate a mailbox in the exchange.

    Terminating a mailbox means that the corresponding entity will no
    longer be able to receive messages.

    Note:
        This method is a no-op if the mailbox does not exist.

    Args:
        uid: Entity identifier of the mailbox to close.
    """
    await self._transport.terminate(uid)