academy.exchange.cloud

academy/exchange/cloud/__init__.py

HttpExchangeFactory

HttpExchangeFactory(
    url: str = DEFAULT_EXCHANGE_URL,
    auth_method: Literal["globus"] | None = None,
    additional_headers: dict[str, str] | None = None,
    request_timeout_s: float = 60,
    ssl_verify: bool | None = None,
)

Bases: ExchangeFactory[HttpExchangeTransport]

Http exchange client factory.

Parameters:

• url (str, default: DEFAULT_EXCHANGE_URL ) –

  Address of HTTP exchange. Defaults to the Academy-hosted exchange

• auth_method (Literal['globus'] | None, default: None ) –

  Method to get authorization headers

• additional_headers (dict[str, str] | None, default: None ) –

  Any other information necessary to communicate with the exchange. Used for passing the Globus bearer token

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

  Same as requests.Session.verify. If the server's TLS certificate should be validated. Should be true if using HTTPS Only set to false for testing or local development.

Source code in academy/exchange/cloud/client.py

def __init__(
    self,
    url: str = DEFAULT_EXCHANGE_URL,
    auth_method: Literal['globus'] | None = None,
    additional_headers: dict[str, str] | None = None,
    request_timeout_s: float = 60,
    ssl_verify: bool | None = None,
) -> None:
    if additional_headers is None:
        additional_headers = {}

    if (
        url == DEFAULT_EXCHANGE_URL
        and 'Authorization' not in additional_headers
    ):
        auth_method = 'globus'

    additional_headers |= get_auth_headers(auth_method)

    self._info = _HttpConnectionInfo(
        url=url,
        additional_headers=additional_headers,
        ssl_verify=ssl_verify,
        request_timeout_s=request_timeout_s,
    )

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:

• AgentExchangeClient[AgentT, ExchangeTransportT] –

  Agent exchange client.

Raises:

• BadEntityIdError –

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

Source code in academy/exchange/factory.py

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:

• UserExchangeClient[ExchangeTransportT] –

  User exchange client.

Source code in academy/exchange/factory.py

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,
    )

HttpExchangeTransport

HttpExchangeTransport(
    mailbox_id: EntityId,
    session: ClientSession,
    connection_info: _HttpConnectionInfo,
)

Bases: ExchangeTransportMixin, NoPickleMixin

Http exchange client.

Parameters:

• mailbox_id (EntityId) –

  Identifier of the mailbox on the exchange. If there is not an id provided, the exchange will create a new client mailbox.

• session (ClientSession) –

  Http session.

• connection_info (_HttpConnectionInfo) –

  Exchange connection info.

Source code in academy/exchange/cloud/client.py

def __init__(
    self,
    mailbox_id: EntityId,
    session: aiohttp.ClientSession,
    connection_info: _HttpConnectionInfo,
) -> None:
    self._mailbox_id = mailbox_id
    self._session = session
    self._info = connection_info
    self._retry_time_ms: float = 1000
    self._last_event_id: int | None = None

    base_url = self._info.url
    self._mailbox_url = f'{base_url}/mailbox'
    self._message_url = f'{base_url}/message'
    self._discover_url = f'{base_url}/discover'
    self._listen_url = f'{base_url}/mailbox/listen'
    self._heartbeat_url = f'{base_url}/mailbox/heartbeat'

new async classmethod

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

Instantiate a new transport.

Parameters:

• connection_info (_HttpConnectionInfo) –

  Exchange connection information.

• 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 registered entity if mailbox_id is None.

Returns:

• Self –

  An instantiated transport bound to a specific mailbox.

Source code in academy/exchange/cloud/client.py

@classmethod
async def new(
    cls,
    *,
    connection_info: _HttpConnectionInfo,
    mailbox_id: EntityId | None = None,
    name: str | None = None,
) -> Self:
    """Instantiate a new transport.

    Args:
        connection_info: Exchange connection information.
        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 registered entity if `mailbox_id` is
            `None`.

    Returns:
        An instantiated transport bound to a specific mailbox.
    """
    ssl_verify = connection_info.ssl_verify
    if ssl_verify is None:  # pragma: no branch
        scheme = urlparse(connection_info.url).scheme
        ssl_verify = scheme == 'https'

    session = aiohttp.ClientSession(
        connector=aiohttp.TCPConnector(ssl=ssl_verify),
        headers=connection_info.additional_headers,
        trust_env=True,
    )

    if mailbox_id is None:
        mailbox_id = UserId.new(name=name)
        async with session.post(
            f'{connection_info.url}/mailbox',
            json={'mailbox': mailbox_id.model_dump_json()},
        ) as response:
            _raise_for_status(response, mailbox_id)
        logger.info(
            'Registered %s in exchange',
            mailbox_id,
            extra={'academy.mailbox_id': mailbox_id},
        )

    return cls(mailbox_id, session, connection_info)

GlobusExchangeFactory

GlobusExchangeFactory(
    project_id: UUID,
    client_params: dict[str, Any] | None = None,
    request_timeout_s: float = 60,
)

Bases: ExchangeFactory[GlobusExchangeTransport]

Globus exchange client factory.

Parameters:

• project_id (UUID) –

  Project to create new clients under. Must be able to authenticate as a administrator.

• client_params (dict[str, Any] | None, default: None ) –

  Additional parameters for globus client.

• request_timeout_s (float, default: 60 ) –

  Maximum length of receive/listen requests to the exchange.

Source code in academy/exchange/cloud/globus.py

def __init__(
    self,
    project_id: uuid.UUID,
    client_params: dict[str, Any] | None = None,
    request_timeout_s: float = 60,
) -> None:
    self.info = _AcademyConnectionInfo(
        project_id=project_id,
        client_params=client_params,
        request_timeout_s=request_timeout_s,
    )

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:

• AgentExchangeClient[AgentT, ExchangeTransportT] –

  Agent exchange client.

Raises:

• BadEntityIdError –

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

Source code in academy/exchange/factory.py

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:

• UserExchangeClient[ExchangeTransportT] –

  User exchange client.

Source code in academy/exchange/factory.py

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,
    )

GlobusExchangeTransport

GlobusExchangeTransport(
    mailbox_id: EntityId,
    *,
    connection_info: _AcademyConnectionInfo,
    app: GlobusApp | None = None,
    authorizer: GlobusAuthorizer | None = None
)

Bases: ExchangeTransportMixin, NoPickleMixin

Globus exchange client.

Parameters:

• mailbox_id (EntityId) –

  Identifier of the mailbox on the exchange. If there is not an id provided, the exchange will create a new client mailbox.

• connection_info (_AcademyConnectionInfo) –

  Project id, client parameters and other information about the connection to the service.

• app (GlobusApp | None, default: None ) –

  For user authorization through token retrieval.

• authorizer (GlobusAuthorizer | None, default: None ) –

  For service authorization through token retrieval.

Source code in academy/exchange/cloud/globus.py

def __init__(
    self,
    mailbox_id: EntityId,
    *,
    connection_info: _AcademyConnectionInfo,
    app: GlobusApp | None = None,
    authorizer: GlobusAuthorizer | None = None,
) -> None:
    self._mailbox_id = mailbox_id
    self.project = connection_info.project_id
    self.child_clients: list[uuid.UUID] = []
    self.client_params = connection_info.client_params or {}
    self.request_timeout_s = connection_info.request_timeout_s

    self.login_time = datetime.min
    self._app = app
    self._authorizer = authorizer
    self._auth_lock = threading.Lock()
    self._local_data = threading.local()
    self.executor = ThreadPoolExecutor(
        thread_name_prefix='exchange-globus-thread',
    )

exchange_client property

exchange_client: AcademyGlobusClient

A thread local copy of the Globus AuthClient.

auth_client property

auth_client: AuthClient

A thread local copy of the Globus AuthClient.

new async classmethod

new(
    *,
    connection_info: _AcademyConnectionInfo,
    app: GlobusApp | None = None,
    authorizer: GlobusAuthorizer | None = None,
    mailbox_id: EntityId | None = None,
    name: str | None = None
) -> Self

Instantiate a new transport.

Parameters:

• connection_info (_AcademyConnectionInfo) –

  Project id, client parameters and other information about the connection to the service.

• app (GlobusApp | None, default: None ) –

  For user authorization through token retrieval

• authorizer (GlobusAuthorizer | None, default: None ) –

  For service authorization through token retrieval

• 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 registered entity if mailbox_id is None.

Returns:

• Self –

  An instantiated transport bound to a specific mailbox.

Source code in academy/exchange/cloud/globus.py

@classmethod
async def new(
    cls,
    *,
    connection_info: _AcademyConnectionInfo,
    app: GlobusApp | None = None,
    authorizer: GlobusAuthorizer | None = None,
    mailbox_id: EntityId | None = None,
    name: str | None = None,
) -> Self:
    """Instantiate a new transport.

    Args:
        connection_info: Project id, client parameters and other
            information about the connection to the service.
        app: For user authorization through token retrieval
        authorizer: For service authorization through token retrieval
        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 registered entity if `mailbox_id` is
            `None`.

    Returns:
        An instantiated transport bound to a specific mailbox.
    """
    loop = asyncio.get_running_loop()

    if mailbox_id is None:
        mailbox_id = UserId.new(name=name)
        client = cls(
            mailbox_id,
            connection_info=connection_info,
            app=app,
            authorizer=authorizer,
        )
        await loop.run_in_executor(
            client.executor,
            client._register_client,
        )
        logger.info(
            'Registered %s in exchange',
            mailbox_id,
            extra={'academy.mailbox_id': mailbox_id},
        )
        return client

    return cls(
        mailbox_id,
        connection_info=connection_info,
        app=app,
        authorizer=authorizer,
    )

register_agents async

register_agents(
    agents: list[tuple[type[AgentT], str | None]],
) -> list[GlobusAgentRegistration[AgentT]]

Register multiple agents with a single auth prompt.

Parameters:

• agents (list[tuple[type[AgentT], str | None]]) –

  List of (agent_type, name) pairs to register.

Returns:

• list[GlobusAgentRegistration[AgentT]] –

  List of agent registrations in order of input.

Source code in academy/exchange/cloud/globus.py

async def register_agents(
    self,
    agents: list[tuple[type[AgentT], str | None]],
) -> list[GlobusAgentRegistration[AgentT]]:
    """Register multiple agents with a single auth prompt.

    Args:
        agents: List of (agent_type, name) pairs to register.

    Returns:
        List of agent registrations in order of input.
    """
    loop = asyncio.get_running_loop()

    # Phase 1: create auth entities concurrently.
    # _prepare_registration no longer mutates shared GlobusApp
    # state, and the auth_client property serializes interactive
    # login via _auth_lock, so parallel dispatch is safe.
    aids: list[AgentId[AgentT]] = [
        AgentId.new(name=name) for _, name in agents
    ]
    prepared: list[_PendingRegistration[AgentT]] = list(
        await asyncio.gather(
            *(
                loop.run_in_executor(
                    self.executor,
                    self._prepare_registration,
                    aid,
                )
                for aid in aids
            ),
        ),
    )
    pending = list(
        zip(
            prepared,
            [at for at, _ in agents],
            strict=True,
        ),
    )

    # Accumulate scope requirements, then do a single login
    # that satisfies all of them at once.
    assert self._app is not None
    for p, _ in pending:
        self._app.add_scope_requirements(
            {str(p.client_id): [p.scope]},
        )

    if self._app.login_required():
        await loop.run_in_executor(
            self.executor,
            self._app.login,
        )

    # Phase 2: extract delegated tokens and create mailboxes.
    # Safe to parallelize — no shared state mutation, and the
    # app already holds valid tokens for all scopes.
    regs: list[GlobusAgentRegistration[AgentT]] = list(
        await asyncio.gather(
            *(
                loop.run_in_executor(
                    self.executor,
                    self._finalize_registration,
                    p,
                )
                for p, _ in pending
            ),
        ),
    )

    await asyncio.gather(
        *(
            loop.run_in_executor(
                self.executor,
                self._create_mailbox,
                reg.agent_id,
                agent_type,
            )
            for reg, (_, agent_type) in zip(
                regs,
                pending,
                strict=True,
            )
        ),
    )

    for reg in regs:
        self.child_clients.append(reg.client_id)

    return regs

spawn_http_exchange

spawn_http_exchange(
    host: str = "0.0.0.0",
    port: int = 5463,
    *,
    level: int | str = WARNING,
    timeout: float | None = None
) -> Generator[HttpExchangeFactory]

Context manager that spawns an HTTP exchange in a subprocess.

This function spawns a new process (rather than forking) and wait to return until a connection with the exchange has been established. When exiting the context manager, SIGINT will be sent to the exchange process. If the process does not exit within 5 seconds, it will be killed.

Warning

The exclusion of authentication and ssl configuration is intentional. This method should only be used for temporary exchanges in trusted environments (such as a fully firewalled individual workstation with no other user access).

Parameters:

• host (str, default: '0.0.0.0' ) –

  Host the exchange should listen on.

• port (int, default: 5463 ) –

  Port the exchange should listen on.

• level (int | str, default: WARNING ) –

  Logging level.

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

  Connection timeout when waiting for exchange to start.

Returns:

• Generator[HttpExchangeFactory] –

  Exchange interface connected to the spawned exchange.

Source code in academy/exchange/cloud/client.py

@contextlib.contextmanager
def spawn_http_exchange(
    host: str = '0.0.0.0',
    port: int = 5463,
    *,
    level: int | str = logging.WARNING,
    timeout: float | None = None,
) -> Generator[HttpExchangeFactory]:
    """Context manager that spawns an HTTP exchange in a subprocess.

    This function spawns a new process (rather than forking) and wait to
    return until a connection with the exchange has been established.
    When exiting the context manager, `SIGINT` will be sent to the exchange
    process. If the process does not exit within 5 seconds, it will be
    killed.

    Warning:
        The exclusion of authentication and ssl configuration is
        intentional. This method should only be used for temporary exchanges
        in trusted environments (such as a fully firewalled individual
        workstation with no other user access).

    Args:
        host: Host the exchange should listen on.
        port: Port the exchange should listen on.
        level: Logging level.
        timeout: Connection timeout when waiting for exchange to start.

    Returns:
        Exchange interface connected to the spawned exchange.
    """
    config = ExchangeServingConfig(
        host=host,
        port=port,
        logger=LogConfig(level=level),
    )

    # Fork is not safe in multi-threaded context.
    context = multiprocessing.get_context('spawn')
    exchange_process = context.Process(target=_run, args=(config,))
    exchange_process.start()

    logger.info('Starting exchange server...')
    wait_connection(host, port, timeout=timeout)
    logger.info('Started exchange server')

    base_url = f'http://{host}:{port}'
    factory = HttpExchangeFactory(base_url)
    try:
        yield factory
    finally:
        logger.info('Terminating exchange server...')
        wait = 5
        exchange_process.terminate()
        exchange_process.join(timeout=wait)
        if exchange_process.exitcode is None:  # pragma: no cover
            logger.info(
                'Killing exchange server after waiting %s seconds',
                wait,
                extra={'academy.delay': wait},
            )
            exchange_process.kill()
            exchange_process.join()
        else:
            logger.info('Terminated exchange server')
        exchange_process.close()