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 
68
69
70
71
72
73
74
75
76
77
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 
 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
125
126
127
128
129
130
131
132
133
134
135
136
137
@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
    p = client.ping()
    assert isinstance(p, Awaitable), (
        'ping should be awaitable from an async redis instance'
    )
    await p

    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,
            extra={'academy.mailbox_id': 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 
284
285
286
287
288
289
290
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[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,
    )