Skip to content

academy.exchange.local

LocalAgentRegistration dataclass 

LocalAgentRegistration(agent_id: AgentId[AgentT])

Bases: Generic[AgentT]

Agent registration for thread exchanges.

agent_id instance-attribute 

agent_id: AgentId[AgentT]

Unique identifier for the agent created by the exchange.

LocalExchangeTransport 

LocalExchangeTransport(
    mailbox_id: EntityId, state: _LocalExchangeState
)

Bases: ExchangeTransportMixin, NoPickleMixin

Local exchange client bound to a specific mailbox.

Source code in academy/exchange/local.py 
70
71
72
73
74
75
76
def __init__(
    self,
    mailbox_id: EntityId,
    state: _LocalExchangeState,
) -> None:
    self._mailbox_id = mailbox_id
    self._state = state

new classmethod 

new(
    mailbox_id: EntityId | None = None,
    *,
    name: str | None = None,
    state: _LocalExchangeState
) -> Self

Instantiate a new transport.

Parameters:

  • mailbox_id (EntityId | None, default: None ) –

    Bind the transport to the specific mailbox. If None, a new user entity will be registered and the transport will be bound to that mailbox.

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

    Display name of the redistered entity if mailbox_id is None.

  • state (_LocalExchangeState) –

    Shared state among exchange clients.

Returns:

  • Self –

    An instantiated transport bound to a specific mailbox.

Source code in academy/exchange/local.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@classmethod
def new(
    cls,
    mailbox_id: EntityId | None = None,
    *,
    name: str | None = None,
    state: _LocalExchangeState,
) -> Self:
    """Instantiate a new transport.

    Args:
        mailbox_id: Bind the transport to the specific mailbox. If `None`,
            a new user entity will be registered and the transport will be
            bound to that mailbox.
        name: Display name of the redistered entity if `mailbox_id` is
            `None`.
        state: Shared state among exchange clients.

    Returns:
        An instantiated transport bound to a specific mailbox.
    """
    if mailbox_id is None:
        mailbox_id = UserId.new(name=name)
        state.queues[mailbox_id] = Queue().async_q
        state.locks[mailbox_id] = Lock()
        logger.info(
            'Registered %s in exchange',
            mailbox_id,
            extra={'academy.mailbox_id': mailbox_id},
        )
    return cls(mailbox_id, state)

LocalExchangeFactory 

LocalExchangeFactory(
    *, _state: _LocalExchangeState | None = None
)

Bases: ExchangeFactory[LocalExchangeTransport], NoPickleMixin

Local exchange client factory.

A thread exchange can be used to pass messages between agents running in separate threads of a single process.

Source code in academy/exchange/local.py 
206
207
208
209
210
211
def __init__(
    self,
    *,
    _state: _LocalExchangeState | None = None,
):
    self._state = _LocalExchangeState() if _state is None else _state

create_agent_client async 

create_agent_client(
    registration: AgentRegistration[AgentT],
    request_handler: RequestHandler[RequestT_co],
) -> AgentExchangeClient[AgentT, ExchangeTransportT]

Create a new agent exchange client.

An agent must be registered with the exchange before an exchange client can be created. For example: 

factory = ExchangeFactory(...)
user_client = factory.create_user_client()
registration = user_client.register_agent(...)
agent_client = factory.create_agent_client(registration, ...)

Parameters:

  • registration (AgentRegistration[AgentT]) –

    Registration information returned by the exchange.

  • request_handler (RequestHandler[RequestT_co]) –

    Agent request message handler.

Returns:

Raises:

  • BadEntityIdError –

    If an agent with registration.agent_id is not already registered with the exchange.

Source code in academy/exchange/factory.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def create_agent_client(
    self,
    registration: AgentRegistration[AgentT],
    request_handler: RequestHandler[RequestT_co],
) -> AgentExchangeClient[AgentT, ExchangeTransportT]:
    """Create a new agent exchange client.

    An agent must be registered with the exchange before an exchange
    client can be created. For example:
    ```python
    factory = ExchangeFactory(...)
    user_client = factory.create_user_client()
    registration = user_client.register_agent(...)
    agent_client = factory.create_agent_client(registration, ...)
    ```

    Args:
        registration: Registration information returned by the exchange.
        request_handler: Agent request message handler.

    Returns:
        Agent exchange client.

    Raises:
        BadEntityIdError: If an agent with `registration.agent_id` is not
            already registered with the exchange.
    """
    agent_id: AgentId[AgentT] = registration.agent_id
    transport = await self._create_transport(
        mailbox_id=agent_id,
        registration=registration,
    )
    assert transport.mailbox_id == agent_id
    status = await transport.status(agent_id)
    if status != MailboxStatus.ACTIVE:
        await transport.close()
        raise BadEntityIdError(agent_id)
    return AgentExchangeClient(
        agent_id,
        transport,
        request_handler=request_handler,
    )

create_user_client async 

create_user_client(
    *, name: str | None = None, start_listener: bool = True
) -> UserExchangeClient[ExchangeTransportT]

Create a new user in the exchange and associated client.

Parameters:

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

    Display name of the client on the exchange.

  • start_listener (bool, default: True ) –

    Start a message listener thread.

Returns:

Source code in academy/exchange/factory.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def create_user_client(
    self,
    *,
    name: str | None = None,
    start_listener: bool = True,
) -> UserExchangeClient[ExchangeTransportT]:
    """Create a new user in the exchange and associated client.

    Args:
        name: Display name of the client on the exchange.
        start_listener: Start a message listener thread.

    Returns:
        User exchange client.
    """
    transport = await self._create_transport(mailbox_id=None, name=name)
    user_id = transport.mailbox_id
    assert isinstance(user_id, UserId)
    return UserExchangeClient(
        user_id,
        transport,
        start_listener=start_listener,
    )