orchestrata_ultimate.cxp.transport

Reference › orchestrata_ultimate › cxp › transport

transport is the construction point for the CXP client and server. Code that needs to talk to the kernel calls create_cxp_client or create_cxp_server here and does not import a concrete transport. Its importers include the gateway server (which uses create_cxp_client to reach the kernel), the agent runtime, the kernel CLI and dispatcher, the knowledge-graph server and the test-services script. See cxp.

CXPServerLike and CXPClientLike are protocols naming what callers rely on. The server side offers servicer, start, stop and serve_forever. The client side offers connect, disconnect, send_packet, send_packets, receive_packets, subscribe, acknowledge, health_check and get_status. cxp_transport returns the selected transport, which is always cxpt, the only transport: it reads CXP_TRANSPORT and raises ValueError if the variable still names the retired grpc transport or any unknown one. Both factories call it first, so a stale setting fails before anything is built, and then return DirectCXPClient or DirectCXPServer from direct_cxp_adapter.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPServerLike (protocol)

Bases: Protocol · Line 41

Structural type for a CXP server (currently the cxpt direct transport).

CXPClientLike (protocol)

Bases: Protocol · Line 54

Structural type for a CXP client (currently the cxpt direct transport).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/cxp/transport.py

"""CXP transport construction.

The CXP seam runs on ``cxpt`` -- the vendored free-threaded direct transport.
``grpc.aio`` was the historical alternative and was retired on 2026-08-07
(SPEC-023: its EventEngine threads re-enable the GIL, and its vendored
free-threaded wheel could not be resolved by a fresh clone). See
``plans/grpc-retirement-design.md``.

Call sites still construct through :func:`create_cxp_server` /
:func:`create_cxp_client` rather than naming a concrete class, and depend on
the :class:`CXPServerLike` / :class:`CXPClientLike` structural types. That
seam is why the retirement was a deletion rather than a migration -- and it
stays, so a future transport is again a construction-site change.

``CXP_TRANSPORT`` is still read, but only to REJECT a stale ``grpc`` value:
a deployment that still asks for grpc must fail loudly at construction
rather than silently receive a different transport than it configured.
"""

from __future__ import annotations

import os
from collections.abc import Awaitable, Callable, Mapping
from typing import TYPE_CHECKING, Any, Protocol

if TYPE_CHECKING:
    from orchestrata_ultimate.core.packet import CXPPacket as CXPPacketModel
    from orchestrata_ultimate.cxp.proto_codec import CXPClientConfig
    from orchestrata_ultimate.cxp.servicer import CXPServerConfig, CXPServicer

# cxpt is the only transport. The env var survives solely so a stale
# `CXP_TRANSPORT=grpc` is rejected instead of silently ignored.
_TRANSPORT_ENV_VAR = "CXP_TRANSPORT"
_GRPC = "grpc"
_CXPT = "cxpt"

# packet_handler shape shared by CXPServer / DirectCXPServer.
PacketHandler = Callable[[dict[str, Any]], "dict[str, Any] | Awaitable[dict[str, Any]]"]


class CXPServerLike(Protocol):
    """Structural type for a CXP server (currently the cxpt direct transport)."""

    @property
    def servicer(self) -> CXPServicer:
        """The CXPServicer the transport wraps (reply sink seam lives here)."""
        ...

    async def start(self) -> None: ...
    async def stop(self, grace: int = ...) -> None: ...
    async def serve_forever(self) -> None: ...


class CXPClientLike(Protocol):
    """Structural type for a CXP client (currently the cxpt direct transport)."""

    async def connect(self) -> None: ...
    async def disconnect(self) -> None: ...
    async def send_packet(self, packet: CXPPacketModel) -> dict[str, Any]: ...
    async def send_packets(
        self, packets: list[CXPPacketModel]
    ) -> list[dict[str, Any]]: ...
    async def receive_packets(
        self, agent_id: str, timeout_seconds: int = ..., max_messages: int = ...
    ) -> list[dict[str, Any]]: ...
    async def subscribe(
        self,
        agent_id: str,
        subjects: list[str] | None = ...,
        from_agents: list[str] | None = ...,
    ) -> list[dict[str, Any]]: ...
    async def acknowledge(
        self, message_id: str, agent_id: str, status: str = ...
    ) -> bool: ...
    async def health_check(self) -> dict[str, Any]: ...
    async def get_status(self, requester: str = ...) -> list[dict[str, Any]]: ...


def cxp_transport(env: Mapping[str, str] | None = None) -> str:
    """Return the selected CXP transport.

    Always ``"cxpt"`` -- it is the only transport. Raises if the environment
    still asks for the retired ``grpc``.

    Args:
        env: Optional explicit environment mapping (tests inject one instead of
            mutating ``os.environ``); defaults to the process environment.

    Raises:
        ValueError: If ``CXP_TRANSPORT`` names a transport that no longer
            exists.
    """
    environ = os.environ if env is None else env
    requested = environ.get(_TRANSPORT_ENV_VAR, _CXPT).strip().lower()
    if requested == _GRPC:
        raise ValueError(
            f"{_TRANSPORT_ENV_VAR}={_GRPC!r} but the grpc.aio CXP transport was "
            "retired on 2026-08-07 (SPEC-023). Remove the setting or set it to "
            f"{_CXPT!r}; see plans/grpc-retirement-design.md."
        )
    if requested != _CXPT:
        raise ValueError(
            f"{_TRANSPORT_ENV_VAR}={requested!r} is not a known CXP transport "
            f"(expected {_CXPT!r})"
        )
    return requested


def create_cxp_server(
    config: CXPServerConfig | None = None,
    packet_handler: PacketHandler | None = None,
    *,
    env: Mapping[str, str] | None = None,
) -> CXPServerLike:
    """Construct the CXP server.

    Raises:
        ValueError: If ``CXP_TRANSPORT`` names a retired/unknown transport.
    """
    cxp_transport(env)  # reject a stale CXP_TRANSPORT before constructing
    from orchestrata_ultimate.cxp.direct_cxp_adapter import DirectCXPServer

    return DirectCXPServer(config, packet_handler)


def create_cxp_client(
    config: CXPClientConfig | None = None,
    *,
    env: Mapping[str, str] | None = None,
) -> CXPClientLike:
    """Construct the CXP client.

    Raises:
        ValueError: If ``CXP_TRANSPORT`` names a retired/unknown transport.
    """
    cxp_transport(env)  # reject a stale CXP_TRANSPORT before constructing
    from orchestrata_ultimate.cxp.direct_cxp_adapter import DirectCXPClient

    return DirectCXPClient(config)