orchestrata_ultimate.cxp.endpoint_config

Reference › orchestrata_ultimate › cxp › endpoint_config

endpoint_config is the single place that resolves where the CXP kernel endpoint is, from environment variables read at call time. It has no internal imports and is used by proto_codec, servicer, the gateway server, and the kernel, agent runtime, harness and knowledge-graph server modules. Every function accepts an optional env mapping, which tests use in place of os.environ.

cxp_grpc_port returns the port from CXP_GRPC_PORT or else the port embedded in CXP_GRPC_ENDPOINT, and raises CXPConfigError if neither is set. require_port has no default: an unset, empty, non-integer or out-of-range value raises CXPConfigError. cxp_grpc_host reads CXP_GRPC_HOST and defaults to localhost. cxp_grpc_endpoint returns CXP_GRPC_ENDPOINT when set, which lets a deployment point at a remote kernel host, and otherwise builds one from the host and port. _explicit_endpoint treats an unset value, an empty value or an unsubstituted CHANGEME_ template value as not configured. The names keep the grpc wording although the transport is now the cxpt direct transport built in transport. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPConfigError

Bases: RuntimeError · Line 19

A required CXP/gRPC environment variable is missing or malformed.

Functions

Source

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

"""Single source of truth for CXP gRPC endpoint configuration.

Ports are read from the environment (ultimately from ``.env``) at call time,
with **no hardcoded numeric default anywhere** — a missing port raises rather
than silently falling back, so the port lives in exactly one place and cannot
drift between the kernel, the dispatcher, the harness, and the deployment
manifests.

Host values are *not* ports and may keep a sensible default (e.g. ``localhost``
for connecting, ``0.0.0.0`` for binding); only ports are env-mandatory.
"""

from __future__ import annotations

import os
from collections.abc import Mapping


class CXPConfigError(RuntimeError):
    """A required CXP/gRPC environment variable is missing or malformed."""


def _get_env(env: Mapping[str, str] | None) -> Mapping[str, str]:
    """Return *env* if provided, otherwise the process environment."""
    return os.environ if env is None else env


def require_port(env_var: str, env: Mapping[str, str] | None = None) -> int:
    """Return the integer port named by *env_var*, read at call time.

    No default: if the variable is unset or empty, raise — the value must be
    defined in the environment (``.env``).

    Args:
        env_var: Name of the environment variable to read.
        env: Optional explicit environment mapping. When provided, values are
            read from this mapping instead of ``os.environ``.
    """
    environ = _get_env(env)
    raw = environ.get(env_var)
    if not raw:
        raise CXPConfigError(
            f"{env_var} is not set; define it in .env (no hardcoded default)."
        )
    try:
        port = int(raw)
    except ValueError as exc:
        raise CXPConfigError(f"{env_var}={raw!r} is not an integer") from exc
    if not 1 <= port <= 65535:
        raise CXPConfigError(f"{env_var}={port} is outside the 1..65535 range")
    return port


def _port_from_endpoint(endpoint: str) -> int:
    cleaned = endpoint.replace("grpc://", "").replace("cxp://", "")
    if ":" not in cleaned:
        raise CXPConfigError(f"CXP_GRPC_ENDPOINT={endpoint!r} has no ':port'")
    _host, port_str = cleaned.rsplit(":", 1)
    try:
        port = int(port_str)
    except ValueError as exc:
        raise CXPConfigError(
            f"CXP_GRPC_ENDPOINT={endpoint!r} port {port_str!r} is not an integer"
        ) from exc
    return port


def _explicit_endpoint(environ: Mapping[str, str]) -> str | None:
    """Return ``CXP_GRPC_ENDPOINT``, or ``None`` when unset/empty/sentinel.

    An unsubstituted .env template sentinel (``CHANGEME_*``) means "not
    configured" — same convention as ``kernel/cli_main._load_config``'s
    ``ORCHESTRATA_CONFIG`` guard.
    """
    endpoint = environ.get("CXP_GRPC_ENDPOINT")
    if not endpoint or endpoint.startswith("CHANGEME_"):
        return None
    return endpoint


def cxp_grpc_port(env: Mapping[str, str] | None = None) -> int:
    """The CXP gRPC kernel port.

    ``CXP_GRPC_PORT`` if set, else the port embedded in ``CXP_GRPC_ENDPOINT``;
    raises ``CXPConfigError`` if neither is defined.

    Args:
        env: Optional explicit environment mapping. When provided, values are
            read from this mapping instead of ``os.environ``.
    """
    environ = _get_env(env)
    if environ.get("CXP_GRPC_PORT"):
        return require_port("CXP_GRPC_PORT", env=environ)
    endpoint = _explicit_endpoint(environ)
    if endpoint:
        return _port_from_endpoint(endpoint)
    raise CXPConfigError(
        "Neither CXP_GRPC_PORT nor CXP_GRPC_ENDPOINT is set; define one in .env."
    )


def cxp_grpc_host(env: Mapping[str, str] | None = None) -> str:
    """Host to connect to the CXP kernel (host is not a port; default localhost).

    Args:
        env: Optional explicit environment mapping. When provided, values are
            read from this mapping instead of ``os.environ``.
    """
    return _get_env(env).get("CXP_GRPC_HOST", "localhost")


def cxp_grpc_endpoint(env: Mapping[str, str] | None = None) -> str:
    """The CXP gRPC endpoint string.

    ``CXP_GRPC_ENDPOINT`` if set (lets a deployment point at a remote kernel
    host, e.g. ``grpc://orchestrata-agent-runtime:50053``), otherwise derived as
    ``grpc://<CXP_GRPC_HOST>:<CXP_GRPC_PORT>`` (the port is env-mandatory).

    Args:
        env: Optional explicit environment mapping. When provided, values are
            read from this mapping instead of ``os.environ``.
    """
    environ = _get_env(env)
    explicit = _explicit_endpoint(environ)
    if explicit:
        return explicit
    return f"grpc://{cxp_grpc_host(environ)}:{cxp_grpc_port(environ)}"