Skip to content

academy.exchange.redis

RedisAgentRegistration dataclass 

RedisAgentRegistration(agent_id: AgentId[AgentT])

Bases: Generic[AgentT]

Agent registration for redis exchanges.

agent_id instance-attribute 

agent_id: AgentId[AgentT]

Unique identifier for the agent created by the exchange.

RedisExchangeTransport 

RedisExchangeTransport(
    mailbox_id: EntityId,
    redis_client: Redis,
    *,
    redis_info: _RedisConnectionInfo
)

Bases: ExchangeTransportMixin, NoPickleMixin

Redis exchange transport bound to a specific mailbox.

Source code in academy/exchange/redis.py 
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    mailbox_id: EntityId,
    redis_client: redis.asyncio.Redis,
    *,
    redis_info: _RedisConnectionInfo,
) -> None:
    self._mailbox_id = mailbox_id
    self._client = redis_client
    self._redis_info = redis_info

new async classmethod 

new(
    *,
    mailbox_id: EntityId | None = None,
    name: str | None = None,
    redis_info: _RedisConnectionInfo
) -> 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.

  • redis_info (_RedisConnectionInfo) –

    Redis connection information.

Returns:

  • Self –

    An instantiated transport bound to a specific mailbox.

Raises:

Source code in academy/exchange/redis.py 
 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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@classmethod
async def new(
    cls,
    *,
    mailbox_id: EntityId | None = None,
    name: str | None = None,
    redis_info: _RedisConnectionInfo,
) -> 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`.
        redis_info: Redis connection information.

    Returns:
        An instantiated transport bound to a specific mailbox.

    Raises:
        redis.exceptions.ConnectionError: If the Redis server is not
            reachable.
    """
    client = redis.asyncio.Redis(
        host=redis_info.hostname,
        port=redis_info.port,
        decode_responses=False,
        **redis_info.kwargs,
    )
    # Ensure the redis server is reachable else fail early
    await client.ping()

    if mailbox_id is None:
        mailbox_id = UserId.new(name=name)
        await client.set(
            f'active:{mailbox_id.uid}',
            _MailboxState.ACTIVE.value,
        )
        logger.info('Registered %s in exchange', mailbox_id)
    return cls(mailbox_id, client, redis_info=redis_info)

RedisExchangeFactory 

RedisExchangeFactory(
    hostname: str, port: int, **redis_kwargs: Any
)

Bases: ExchangeFactory[RedisExchangeTransport]

Redis exchange client factory.

Parameters:

  • hostname (str) –

    Redis server hostname.

  • port (int) –

    Redis server port.

  • redis_kwargs (Any, default: {} ) –

    Extra keyword arguments to pass to redis.Redis().

Source code in academy/exchange/redis.py 
264
265
266
267
268
269
270
def __init__(
    self,
    hostname: str,
    port: int,
    **redis_kwargs: Any,
) -> None:
    self.redis_info = _RedisConnectionInfo(hostname, port, redis_kwargs)

create_agent_client async 

create_agent_client(
    registration: AgentRegistration[AgentT],
    request_handler: RequestHandler,
) -> 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) –

    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/__init__.py 
 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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def create_agent_client(
    self,
    registration: AgentRegistration[AgentT],
    request_handler: RequestHandler,
) -> 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/__init__.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
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,
    )