datarobot_genai.dragent.agent_card_registry

agent_card_registry

Client for the central DataRobot agent card registry.

The central registry provides a tenant-scoped list of agent cards that requires only standard DataRobot API-token authentication (DATAROBOT_API_TOKEN). This avoids the chicken-and-egg problem where an individual agent's card endpoint is behind per-agent AuthN/AuthZ.

The :class:AgentCardRegistry supports batch fetching so that many function groups sharing the same workflow can resolve all their cards in a minimum number of HTTP round-trips instead of N+1 individual requests.

Lookups can be registered before the first fetch so that get() automatically triggers a single batch prefetch for all pending IDs on its first invocation — no explicit prefetch() call required.

.. note:: The registry issues one HTTP call per ID kind — deployment, external and workload IDs are never mixed in a single request. Sending deploymentIds together with workloadIds is rejected by the API with HTTP 400 (the two are mutually exclusive), and combining either with externalIds AND-matches, returning empty results.

.. note:: The API caps each ID parameter at 20 values, so :meth:AgentCardRegistry._fetch splits longer ID lists into chunks of 20 and accumulates the entries before parsing them as one result set.

DataRobotRegistrySettings

Bases: DataRobotAppFrameworkBaseSettings

DataRobot endpoint setting for the central agent card registry.

Loads DATAROBOT_ENDPOINT from env vars (including Runtime Parameters), .env, file secrets, or Pulumi config using the standard :class:DataRobotAppFrameworkBaseSettings priority chain.

Deliberately not read through resolve_config().resolve_datarobot_endpoint(): that resolver substitutes the public app.datarobot.com default when nothing is configured, and this lookup sends the API token to whatever host it resolves, so an unset endpoint has to stay distinguishable from a configured one. The API token itself does come from the global config; see :func:_resolve_settings.

Source code in datarobot_genai/dragent/agent_card_registry.py

class DataRobotRegistrySettings(DataRobotAppFrameworkBaseSettings):
    """DataRobot endpoint setting for the central agent card registry.

    Loads ``DATAROBOT_ENDPOINT`` from env vars (including Runtime Parameters),
    ``.env``, file secrets, or Pulumi config using the standard
    :class:`DataRobotAppFrameworkBaseSettings` priority chain.

    Deliberately *not* read through ``resolve_config().resolve_datarobot_endpoint()``:
    that resolver substitutes the public ``app.datarobot.com`` default when nothing
    is configured, and this lookup sends the API token to whatever host it resolves,
    so an unset endpoint has to stay distinguishable from a configured one. The API
    token itself does come from the global config; see :func:`_resolve_settings`.
    """

    datarobot_endpoint: str | None = None

AgentCardRegistryConfig

Bases: DataRobotAppFrameworkBaseSettings

Configuration for the agent card registry cache.

Controllable via environment variables (prefix-free, following the standard :class:DataRobotAppFrameworkBaseSettings resolution chain).

Set AGENT_CARD_REGISTRY_CACHE_TTL=0 to disable caching entirely (every get() triggers a fresh HTTP fetch).

AGENT_CARD_REGISTRY_SOFT_CACHE_TTL controls when cached cards are considered fresh (skip registry fetch). AGENT_CARD_REGISTRY_CACHE_TTL is the hard bound for stale-if-error when the registry is unreachable. When soft TTL is unset it defaults to the hard TTL.

Source code in datarobot_genai/dragent/agent_card_registry.py

class AgentCardRegistryConfig(DataRobotAppFrameworkBaseSettings):
    """Configuration for the agent card registry cache.

    Controllable via environment variables (prefix-free, following the
    standard :class:`DataRobotAppFrameworkBaseSettings` resolution chain).

    Set ``AGENT_CARD_REGISTRY_CACHE_TTL=0`` to disable caching entirely
    (every ``get()`` triggers a fresh HTTP fetch).

    ``AGENT_CARD_REGISTRY_SOFT_CACHE_TTL`` controls when cached cards are
    considered fresh (skip registry fetch). ``AGENT_CARD_REGISTRY_CACHE_TTL``
    is the hard bound for stale-if-error when the registry is unreachable.
    When soft TTL is unset it defaults to the hard TTL.
    """

    agent_card_registry_cache_ttl: int = Field(
        default=_DEFAULT_CACHE_TTL_SECONDS,
        ge=0,
        description=(
            "Hard time-to-live for cached agent cards in seconds. Entries older "
            "than this are not served under stale-if-error. Set to 0 to disable "
            "caching (every get() triggers a fresh fetch). Default: 86400 (24 hours)."
        ),
    )

    agent_card_registry_soft_cache_ttl: int | None = Field(
        default=None,
        ge=0,
        description=(
            "Soft time-to-live for cached agent cards in seconds. Entries within "
            "this age are returned without contacting the registry; older entries "
            "are refreshed on demand and by the background refresh loop. Must not "
            "exceed agent_card_registry_cache_ttl. Defaults to the hard TTL when unset."
        ),
    )

    agent_card_registry_timeout: float = Field(
        default=_DEFAULT_TIMEOUT_SECONDS,
        gt=0,
        description="HTTP timeout in seconds for registry API requests. Default: 30.",
    )

    agent_card_registry_on_duplicate: DuplicateStrategy = Field(
        default="first",
        description=(
            "Strategy when the registry returns multiple agent cards for the "
            "same external ID.  The registry API returns cards sorted by "
            "creation time (ascending), so 'first' keeps the earliest "
            "registered card, 'last' keeps the most recently registered card, "
            "and 'error' raises AgentCardRegistryError.  Default: 'first'."
        ),
    )

    @model_validator(mode="after")
    def _resolve_soft_cache_ttl(self) -> AgentCardRegistryConfig:
        """Default unset soft TTL to hard TTL and reject soft > hard."""
        hard = self.agent_card_registry_cache_ttl
        if hard == 0:
            self.agent_card_registry_soft_cache_ttl = 0
            return self
        soft = self.agent_card_registry_soft_cache_ttl
        if soft is None:
            self.agent_card_registry_soft_cache_ttl = hard
            return self
        if soft > hard:
            raise ValueError(
                "agent_card_registry_soft_cache_ttl cannot exceed agent_card_registry_cache_ttl"
            )
        return self

AgentCardRegistryError

Bases: RuntimeError

Raised when the central agent card registry lookup fails.

Source code in datarobot_genai/dragent/agent_card_registry.py

class AgentCardRegistryError(RuntimeError):
    """Raised when the central agent card registry lookup fails."""

ParsedRegistryCards

Bases: NamedTuple

Parsed registry response with lookup key types for cache indexing.

Source code in datarobot_genai/dragent/agent_card_registry.py

class ParsedRegistryCards(NamedTuple):
    """Parsed registry response with lookup key types for cache indexing."""

    cards: dict[str, AgentCard]
    key_types: dict[str, LookupKeyType]
    registry_ids: dict[str, RegistryIds]

AgentCardRegistry

Batch-capable, TTL-cached client for the central agent card registry.

IDs are register()-ed synchronously at config-parse time (no I/O). The first get() flushes all pending IDs in ≤3 HTTP calls — one per ID kind (deployment, external, workload), never mixed in a single request, plus one extra call per 20 IDs of the same kind (the API's per-parameter cap). Subsequent get() calls hit the in-memory cache until the soft TTL (AGENT_CARD_REGISTRY_SOFT_CACHE_TTL, defaulting to the hard TTL) expires. When a refresh fails, a cached card may still be returned while it remains within AGENT_CARD_REGISTRY_CACHE_TTL (the hard TTL / stale-if-error bound).

Source code in datarobot_genai/dragent/agent_card_registry.py

class AgentCardRegistry:
    """Batch-capable, TTL-cached client for the central agent card registry.

    IDs are ``register()``-ed synchronously at config-parse time (no I/O).
    The first ``get()`` flushes all pending IDs in ≤3 HTTP calls — one per ID
    kind (deployment, external, workload), never mixed in a single request, plus
    one extra call per 20 IDs of the same kind (the API's per-parameter cap).
    Subsequent ``get()`` calls hit the in-memory cache until the soft TTL
    (``AGENT_CARD_REGISTRY_SOFT_CACHE_TTL``, defaulting to the hard TTL) expires.
    When a refresh fails, a cached card may still be returned while it remains
    within ``AGENT_CARD_REGISTRY_CACHE_TTL`` (the hard TTL / stale-if-error bound).
    """

    def __init__(
        self,
        *,
        config: AgentCardRegistryConfig,
        api_token: str,
        endpoint: str,
    ) -> None:
        self._lock = asyncio.Lock()

        # Pending registrations (filled synchronously, flushed on first get)
        self._pending_deployment_ids: set[str] = set()
        self._pending_external_ids: set[str] = set()
        self._pending_workload_ids: set[str] = set()

        # All IDs registered at config-parse time (used for background refresh)
        self._registered_deployment_ids: set[str] = set()
        self._registered_external_ids: set[str] = set()
        self._registered_workload_ids: set[str] = set()

        self._timeout = config.agent_card_registry_timeout
        self._cache_ttl = config.agent_card_registry_cache_ttl
        soft_cache_ttl = config.agent_card_registry_soft_cache_ttl
        assert soft_cache_ttl is not None  # normalized by _resolve_soft_cache_ttl
        self._soft_cache_ttl: int = soft_cache_ttl
        self._on_duplicate = config.agent_card_registry_on_duplicate
        self._api_token = api_token
        self._endpoint = endpoint
        self._backend = create_agent_card_cache_backend(self._cache_ttl)
        # Monotonic deadline while the registry is treated as unhealthy.
        self._unhealthy_until: float = 0.0

        logger.info(
            "AgentCardRegistry created (cache_ttl=%ds, soft_cache_ttl=%ds, l2=%s)",
            self._cache_ttl,
            self._soft_cache_ttl,
            isinstance(self._backend, LayeredAgentCardCacheBackend),
        )

    @property
    def soft_cache_ttl(self) -> int:
        """Soft TTL in seconds."""
        return self._soft_cache_ttl

    def _cooldown_active(self) -> bool:
        """Return whether a process-local failure cooldown is still in effect."""
        return time.monotonic() < self._unhealthy_until

    def _mark_registry_unhealthy(self) -> None:
        """Start a short jittered cooldown after a registry HTTP/transport failure."""
        cooldown = random.uniform(
            _FAILURE_COOLDOWN_MIN_SECONDS,
            _FAILURE_COOLDOWN_MAX_SECONDS,
        )
        self._unhealthy_until = time.monotonic() + cooldown
        logger.warning(
            "Agent card registry failure cooldown started (%.0fs); "
            "subsequent lookups will skip HTTP until it expires.",
            cooldown,
        )

    def _raise_if_cooldown_active(self) -> None:
        """Raise when the process-local failure cooldown has not expired."""
        if not self._cooldown_active():
            return
        remaining = max(0.0, self._unhealthy_until - time.monotonic())
        raise AgentCardRegistryError(
            f"Agent card registry is in a failure cooldown "
            f"({remaining:.0f}s remaining); skipping HTTP request."
        )

    # ------------------------------------------------------------------
    # Registration (synchronous — called at config-parse time)
    # ------------------------------------------------------------------

    def register(
        self,
        *,
        deployment_id: str | None = None,
        external_id: str | None = None,
        workload_id: str | None = None,
    ) -> None:
        """Declare intent to look up an agent card.

        Call this at config-parse/validation time so that the first
        :meth:`get` can batch all pending IDs into a single prefetch.

        Exactly one of ``deployment_id``, ``external_id`` or ``workload_id``
        must be given.
        """
        if deployment_id:
            self._pending_deployment_ids.add(deployment_id)
            self._registered_deployment_ids.add(deployment_id)
        elif external_id:
            self._pending_external_ids.add(external_id)
            self._registered_external_ids.add(external_id)
        elif workload_id:
            self._pending_workload_ids.add(workload_id)
            self._registered_workload_ids.add(workload_id)

    def has_registered_lookups(self) -> bool:
        """Return whether any registry lookup IDs were registered."""
        return bool(
            self._registered_deployment_ids
            or self._registered_external_ids
            or self._registered_workload_ids
        )

    # ------------------------------------------------------------------
    # Internal HTTP
    # ------------------------------------------------------------------

    async def _fetch(
        self, params: dict[str, str], *, auth_headers: dict[str, str] | None = None
    ) -> ParsedRegistryCards:
        """Fetch and parse all agent cards matching *params*.

        Splits ID lists longer than the API's per-parameter cap
        (:data:`_MAX_IDS_PER_REQUEST`) into several requests, accumulating the
        raw entries so that the ``on_duplicate`` strategy is applied once
        across the full result set — the same way pagination is accumulated
        inside :meth:`_fetch_entries`.

        *auth_headers* override the owner-key ``Authorization`` header.
        """
        all_entries: list[dict[str, Any]] = []
        for chunk_params in _chunk_id_params(params):
            all_entries.extend(await self._fetch_entries(chunk_params, auth_headers=auth_headers))

        parsed = _parse_registry_response({"data": all_entries}, on_duplicate=self._on_duplicate)
        logger.info("Fetched %d agent card(s) from registry.", len(parsed.cards))
        return parsed

    async def _fetch_entries(
        self, params: dict[str, str], *, auth_headers: dict[str, str] | None = None
    ) -> list[dict[str, Any]]:
        """Execute a single registry HTTP GET with pagination, returning raw entries.

        Requests the maximum page size (100) to minimise round-trips, then
        follows ``next`` links until all pages are consumed.

        After a transport or HTTP failure on the owner-key path a short
        process-local cooldown skips further HTTP attempts so soft-expired
        on-demand lookups do not stampede Control Hub while it is down.
        Caller-credential failures do not start the cooldown: those fall back
        to the owner key, which must still be allowed to try.
        """
        self._raise_if_cooldown_active()

        registry_url = build_agent_cards_registry_url(self._endpoint)
        headers = auth_headers or {"Authorization": f"Bearer {self._api_token}"}
        using_owner_key = auth_headers is None
        params_with_limit = {"limit": str(_MAX_PAGE_SIZE), **params}

        logger.info(
            "Fetching agent cards from registry: %s (params=%s, credential=%s)",
            registry_url,
            params,
            "caller" if auth_headers else "owner",
        )

        all_entries: list[dict[str, Any]] = []
        pages_fetched = 0

        try:
            async with httpx.AsyncClient(timeout=httpx.Timeout(self._timeout)) as client:
                response = await client.get(registry_url, params=params_with_limit, headers=headers)
                response.raise_for_status()
                body = response.json()
                all_entries.extend(body.get("data", []))

                next_url = body.get("next")
                pages_fetched = 1
                while next_url:
                    if pages_fetched >= _MAX_PAGES:
                        logger.warning(
                            "Pagination safety limit reached (%d pages). "
                            "Some agent cards may not have been fetched.",
                            _MAX_PAGES,
                        )
                        break
                    logger.debug("Following pagination link (page %d).", pages_fetched + 1)
                    response = await client.get(next_url, headers=headers)
                    response.raise_for_status()
                    body = response.json()
                    all_entries.extend(body.get("data", []))
                    next_url = body.get("next")
                    pages_fetched += 1

        except httpx.HTTPStatusError as exc:
            if using_owner_key:
                self._mark_registry_unhealthy()
            raise AgentCardRegistryError(
                f"Agent card registry request failed with HTTP "
                f"{exc.response.status_code}. Verify your API token and that "
                f"the agents are registered."
            ) from exc
        except httpx.HTTPError as exc:
            if using_owner_key:
                self._mark_registry_unhealthy()
            raise AgentCardRegistryError(f"Agent card registry request failed: {exc}") from exc

        logger.debug(
            "Fetched %d registry entry(ies) over %d page(s).",
            len(all_entries),
            pages_fetched,
        )
        return all_entries

    async def _is_fresh(self, key: str, *, key_type: LookupKeyType) -> bool:
        """Return True if *key* is cached and within the soft TTL."""
        record = await self._backend.get_fresh(
            key,
            cache_ttl=self._soft_cache_ttl,
            key_type=key_type,
        )
        return record is not None

    async def _try_get_stale(self, key: str, *, key_type: LookupKeyType) -> AgentCard | None:
        """Return a stale cached card when refresh failed."""
        record = await self._backend.get_stale(
            key,
            max_staleness_seconds=self._cache_ttl,
            key_type=key_type,
        )
        if record is None:
            return None
        logger.warning(
            "Central agent card registry unreachable; serving cached agent card for "
            "%s_id='%s' (age=%.0fs)",
            key_type,
            key,
            record.age_seconds(),
        )
        return record.card

    async def _store_cards(self, parsed: ParsedRegistryCards) -> None:
        if not parsed.cards:
            return
        await self._backend.store(
            parsed.cards,
            key_types=parsed.key_types,
            registry_ids=parsed.registry_ids,
        )

    def _age_cache_entry_for_test(self, lookup_key: str, seconds: float) -> None:
        """Shift a cached entry's fetch time backward (tests only)."""
        backend = self._backend
        if isinstance(backend, MemoryAgentCardCacheBackend):
            backend.age_entry_for_test(lookup_key, seconds)
            return

        if isinstance(backend, LayeredAgentCardCacheBackend):
            backend.memory.age_entry_for_test(lookup_key, seconds)

    async def _uncached(self, ids: Iterable[str], *, key_type: LookupKeyType) -> list[str]:
        """Return the *ids* of *key_type* that are not cached within the soft TTL."""
        return [i for i in ids if not await self._is_fresh(i, key_type=key_type)]

    async def _flush_pending(self) -> None:
        """Batch-fetch all registered-but-uncached IDs.  Must be called under ``_lock``."""
        missing: dict[LookupKeyType, list[str]] = {
            "deployment": await self._uncached(self._pending_deployment_ids, key_type="deployment"),
            "external": await self._uncached(self._pending_external_ids, key_type="external"),
            "workload": await self._uncached(self._pending_workload_ids, key_type="workload"),
        }

        # Clear pending sets regardless — they've been processed
        self._pending_deployment_ids.clear()
        self._pending_external_ids.clear()
        self._pending_workload_ids.clear()

        await self._fetch_and_store_by_kind(missing)

    async def _fetch_and_store_by_kind(
        self,
        ids_by_kind: dict[LookupKeyType, list[str]],
    ) -> None:
        """Fetch and cache each ID kind in its **own** request.

        Kinds are never combined: ``deploymentIds`` + ``workloadIds`` is an HTTP
        400, and either mixed with ``externalIds`` AND-matches to nothing.
        """
        for key_type, ids in ids_by_kind.items():
            if not ids:
                continue
            parsed = await self._fetch({_id_param_for(key_type): ",".join(ids)})
            await self._store_cards(parsed)

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    async def prefetch(
        self,
        *,
        deployment_ids: list[str] | None = None,
        external_ids: list[str] | None = None,
        workload_ids: list[str] | None = None,
    ) -> None:
        """Batch-fetch and cache agent cards in a minimum number of HTTP calls.

        Issues **separate** requests per ID kind: ``deploymentIds`` together
        with ``workloadIds`` is rejected with HTTP 400, and either combined with
        ``externalIds`` AND-matches and returns nothing.

        Already-cached (non-expired) IDs are skipped.

        Parameters
        ----------
        deployment_ids:
            Deployment IDs to prefetch.
        external_ids:
            External IDs to prefetch.
        workload_ids:
            Workload IDs to prefetch.
        """
        async with self._lock:
            missing: dict[LookupKeyType, list[str]] = {
                "deployment": await self._uncached(deployment_ids or [], key_type="deployment"),
                "external": await self._uncached(external_ids or [], key_type="external"),
                "workload": await self._uncached(workload_ids or [], key_type="workload"),
            }

            if not any(missing.values()):
                logger.info(
                    "Agent card registry cache: all requested IDs satisfied from cache; "
                    "skipping central registry fetch."
                )
                return

            await self._fetch_and_store_by_kind(missing)

    async def refresh_all_registered(self) -> None:
        """Re-fetch registered IDs whose cache entries are past the soft TTL.

        Existing cache entries are left in place on failure so stale-if-error
        can continue serving them during registry outages.  Errors propagate so
        the background refresh loop can log once and back off.
        """
        if not self.has_registered_lookups():
            logger.debug("No registered agent card IDs; skipping background refresh.")
            return

        deployment_ids = sorted(self._registered_deployment_ids)
        external_ids = sorted(self._registered_external_ids)
        workload_ids = sorted(self._registered_workload_ids)
        logger.debug(
            "Refreshing registered agent cards "
            "(deployment_ids=%s, external_ids=%s, workload_ids=%s)",
            deployment_ids,
            external_ids,
            workload_ids,
        )
        await self.prefetch(
            deployment_ids=deployment_ids or None,
            external_ids=external_ids or None,
            workload_ids=workload_ids or None,
        )

    async def get(
        self,
        *,
        deployment_id: str | None = None,
        external_id: str | None = None,
        workload_id: str | None = None,
        auth_headers: dict[str, str] | None = None,
    ) -> AgentCard:
        """Return a single agent card, using the cache or fetching on demand.

        On the first call, all IDs previously passed to :meth:`register`
        are batch-fetched in a single prefetch (one HTTP call per ID kind).
        Subsequent calls for already-cached, non-expired cards are instant.

        Exactly one of ``deployment_id``, ``external_id`` or ``workload_id``
        must be provided.

        Pass *auth_headers* to fetch a miss as the caller; the owner key is the
        fallback.

        Raises
        ------
        AgentCardRegistryError
            If the card cannot be found or the request fails.
        """
        provided = [i for i in (deployment_id, external_id, workload_id) if i]
        if len(provided) != 1:
            raise AgentCardRegistryError(
                "Specify exactly one of 'deployment_id', 'external_id' or 'workload_id'."
            )

        lookup_key: str = provided[0]
        lookup_key_type: LookupKeyType = (
            "deployment" if deployment_id else "external" if external_id else "workload"
        )

        # Fast path — fresh cache hit
        if fresh := await self._backend.get_fresh(
            lookup_key,
            cache_ttl=self._soft_cache_ttl,
            key_type=lookup_key_type,
        ):
            return fresh.card

        if auth_headers:
            return await self._get_with_caller_auth(lookup_key, lookup_key_type, auth_headers)
        return await self._get_with_owner_key(lookup_key, lookup_key_type)

    async def _get_with_caller_auth(
        self, lookup_key: str, lookup_key_type: LookupKeyType, auth_headers: dict[str, str]
    ) -> AgentCard:
        """Fetch a card as the caller; fall back to the owner key on any failure."""
        params = {_id_param_for(lookup_key_type): lookup_key}
        try:
            parsed = await self._fetch(params, auth_headers=auth_headers)
        except (AgentCardRegistryError, UnicodeEncodeError) as exc:
            # Log the type only: the error may carry the caller's headers.
            reason = type(exc.__cause__ or exc).__name__
        else:
            await self._store_cards(parsed)
            if lookup_key in parsed.cards:
                return parsed.cards[lookup_key]
            reason = "not found"

        logger.warning(
            "Caller-token registry lookup failed for %s_id='%s' (%s); using owner key.",
            lookup_key_type,
            lookup_key,
            reason,
        )
        return await self._get_with_owner_key(lookup_key, lookup_key_type)

    async def _get_with_owner_key(
        self, lookup_key: str, lookup_key_type: LookupKeyType
    ) -> AgentCard:
        async with self._lock:
            # Double-check after acquiring lock
            if fresh := await self._backend.get_fresh(
                lookup_key,
                cache_ttl=self._soft_cache_ttl,
                key_type=lookup_key_type,
            ):
                return fresh.card

            try:
                # Flush all pending registrations in a batch
                if (
                    self._pending_deployment_ids
                    or self._pending_external_ids
                    or self._pending_workload_ids
                ):
                    await self._flush_pending()
                    if fresh := await self._backend.get_fresh(
                        lookup_key,
                        cache_ttl=self._soft_cache_ttl,
                        key_type=lookup_key_type,
                    ):
                        return fresh.card

                # Still not fresh — fetch individually
                params: dict[str, str] = {_id_param_for(lookup_key_type): lookup_key}
                parsed = await self._fetch(params)
                await self._store_cards(parsed)

                if lookup_key in parsed.cards:
                    return parsed.cards[lookup_key]

                if fresh := await self._backend.get_fresh(
                    lookup_key,
                    cache_ttl=self._soft_cache_ttl,
                    key_type=lookup_key_type,
                ):
                    return fresh.card

                # Successful miss — evict stale entry so stale-if-error cannot
                # resurrect a deregistered agent on a later fetch failure.
                await self._backend.evict(lookup_key, key_type=lookup_key_type)
            except AgentCardRegistryError:
                if stale_card := await self._try_get_stale(lookup_key, key_type=lookup_key_type):
                    return stale_card
                raise

        # Fetch succeeded but the requested key was absent from the response.
        raise AgentCardRegistryError(
            f"No agent card found in the central registry for "
            f"{lookup_key_type}_id='{lookup_key}'. Verify that the agent exists and is "
            "registered in your organisation."
        )

soft_cache_ttl property

soft_cache_ttl: int

Soft TTL in seconds.

register

register(*, deployment_id: str | None = None, external_id: str | None = None, workload_id: str | None = None) -> None

Declare intent to look up an agent card.

Call this at config-parse/validation time so that the first :meth:get can batch all pending IDs into a single prefetch.

Exactly one of deployment_id, external_id or workload_id must be given.

Source code in datarobot_genai/dragent/agent_card_registry.py

def register(
    self,
    *,
    deployment_id: str | None = None,
    external_id: str | None = None,
    workload_id: str | None = None,
) -> None:
    """Declare intent to look up an agent card.

    Call this at config-parse/validation time so that the first
    :meth:`get` can batch all pending IDs into a single prefetch.

    Exactly one of ``deployment_id``, ``external_id`` or ``workload_id``
    must be given.
    """
    if deployment_id:
        self._pending_deployment_ids.add(deployment_id)
        self._registered_deployment_ids.add(deployment_id)
    elif external_id:
        self._pending_external_ids.add(external_id)
        self._registered_external_ids.add(external_id)
    elif workload_id:
        self._pending_workload_ids.add(workload_id)
        self._registered_workload_ids.add(workload_id)

has_registered_lookups

has_registered_lookups() -> bool

Return whether any registry lookup IDs were registered.

Source code in datarobot_genai/dragent/agent_card_registry.py

def has_registered_lookups(self) -> bool:
    """Return whether any registry lookup IDs were registered."""
    return bool(
        self._registered_deployment_ids
        or self._registered_external_ids
        or self._registered_workload_ids
    )

prefetch async

prefetch(*, deployment_ids: list[str] | None = None, external_ids: list[str] | None = None, workload_ids: list[str] | None = None) -> None

Batch-fetch and cache agent cards in a minimum number of HTTP calls.

Issues separate requests per ID kind: deploymentIds together with workloadIds is rejected with HTTP 400, and either combined with externalIds AND-matches and returns nothing.

Already-cached (non-expired) IDs are skipped.

Parameters

deployment_ids: Deployment IDs to prefetch. external_ids: External IDs to prefetch. workload_ids: Workload IDs to prefetch.

Source code in datarobot_genai/dragent/agent_card_registry.py

async def prefetch(
    self,
    *,
    deployment_ids: list[str] | None = None,
    external_ids: list[str] | None = None,
    workload_ids: list[str] | None = None,
) -> None:
    """Batch-fetch and cache agent cards in a minimum number of HTTP calls.

    Issues **separate** requests per ID kind: ``deploymentIds`` together
    with ``workloadIds`` is rejected with HTTP 400, and either combined with
    ``externalIds`` AND-matches and returns nothing.

    Already-cached (non-expired) IDs are skipped.

    Parameters
    ----------
    deployment_ids:
        Deployment IDs to prefetch.
    external_ids:
        External IDs to prefetch.
    workload_ids:
        Workload IDs to prefetch.
    """
    async with self._lock:
        missing: dict[LookupKeyType, list[str]] = {
            "deployment": await self._uncached(deployment_ids or [], key_type="deployment"),
            "external": await self._uncached(external_ids or [], key_type="external"),
            "workload": await self._uncached(workload_ids or [], key_type="workload"),
        }

        if not any(missing.values()):
            logger.info(
                "Agent card registry cache: all requested IDs satisfied from cache; "
                "skipping central registry fetch."
            )
            return

        await self._fetch_and_store_by_kind(missing)

refresh_all_registered async

refresh_all_registered() -> None

Re-fetch registered IDs whose cache entries are past the soft TTL.

Existing cache entries are left in place on failure so stale-if-error can continue serving them during registry outages. Errors propagate so the background refresh loop can log once and back off.

Source code in datarobot_genai/dragent/agent_card_registry.py

async def refresh_all_registered(self) -> None:
    """Re-fetch registered IDs whose cache entries are past the soft TTL.

    Existing cache entries are left in place on failure so stale-if-error
    can continue serving them during registry outages.  Errors propagate so
    the background refresh loop can log once and back off.
    """
    if not self.has_registered_lookups():
        logger.debug("No registered agent card IDs; skipping background refresh.")
        return

    deployment_ids = sorted(self._registered_deployment_ids)
    external_ids = sorted(self._registered_external_ids)
    workload_ids = sorted(self._registered_workload_ids)
    logger.debug(
        "Refreshing registered agent cards "
        "(deployment_ids=%s, external_ids=%s, workload_ids=%s)",
        deployment_ids,
        external_ids,
        workload_ids,
    )
    await self.prefetch(
        deployment_ids=deployment_ids or None,
        external_ids=external_ids or None,
        workload_ids=workload_ids or None,
    )

get async

get(*, deployment_id: str | None = None, external_id: str | None = None, workload_id: str | None = None, auth_headers: dict[str, str] | None = None) -> AgentCard

Return a single agent card, using the cache or fetching on demand.

On the first call, all IDs previously passed to :meth:register are batch-fetched in a single prefetch (one HTTP call per ID kind). Subsequent calls for already-cached, non-expired cards are instant.

Exactly one of deployment_id, external_id or workload_id must be provided.

Pass auth_headers to fetch a miss as the caller; the owner key is the fallback.

Raises

AgentCardRegistryError If the card cannot be found or the request fails.

Source code in datarobot_genai/dragent/agent_card_registry.py

async def get(
    self,
    *,
    deployment_id: str | None = None,
    external_id: str | None = None,
    workload_id: str | None = None,
    auth_headers: dict[str, str] | None = None,
) -> AgentCard:
    """Return a single agent card, using the cache or fetching on demand.

    On the first call, all IDs previously passed to :meth:`register`
    are batch-fetched in a single prefetch (one HTTP call per ID kind).
    Subsequent calls for already-cached, non-expired cards are instant.

    Exactly one of ``deployment_id``, ``external_id`` or ``workload_id``
    must be provided.

    Pass *auth_headers* to fetch a miss as the caller; the owner key is the
    fallback.

    Raises
    ------
    AgentCardRegistryError
        If the card cannot be found or the request fails.
    """
    provided = [i for i in (deployment_id, external_id, workload_id) if i]
    if len(provided) != 1:
        raise AgentCardRegistryError(
            "Specify exactly one of 'deployment_id', 'external_id' or 'workload_id'."
        )

    lookup_key: str = provided[0]
    lookup_key_type: LookupKeyType = (
        "deployment" if deployment_id else "external" if external_id else "workload"
    )

    # Fast path — fresh cache hit
    if fresh := await self._backend.get_fresh(
        lookup_key,
        cache_ttl=self._soft_cache_ttl,
        key_type=lookup_key_type,
    ):
        return fresh.card

    if auth_headers:
        return await self._get_with_caller_auth(lookup_key, lookup_key_type, auth_headers)
    return await self._get_with_owner_key(lookup_key, lookup_key_type)

get_default_registry async

get_default_registry() -> AgentCardRegistry

Return the module-level :class:AgentCardRegistry singleton.

Created lazily on first access. Credentials are resolved from :class:DataRobotRegistrySettings at instantiation time. On enclave workloads, L2 MemorySpace provisioning runs on this event loop before the singleton is constructed so the cache backend is not stuck on L1-only.

Source code in datarobot_genai/dragent/agent_card_registry.py

async def get_default_registry() -> AgentCardRegistry:
    """Return the module-level :class:`AgentCardRegistry` singleton.

    Created lazily on first access.  Credentials are resolved from
    :class:`DataRobotRegistrySettings` at instantiation time.  On enclave
    workloads, L2 MemorySpace provisioning runs on this event loop before the
    singleton is constructed so the cache backend is not stuck on L1-only.
    """
    if _RegistryHolder.instance is None:
        async with _RegistryHolder.lock:
            if _RegistryHolder.instance is None:
                # Provision L2 on this loop before __init__ builds the cache
                # backend. The sync helper cannot nest asyncio.run here.
                await try_resolve_memory_space_id_async()
                api_token, endpoint = _resolve_settings()
                _RegistryHolder.instance = AgentCardRegistry(
                    config=AgentCardRegistryConfig(),
                    api_token=api_token,
                    endpoint=endpoint,
                )
    return _RegistryHolder.instance

get_default_registry_sync

get_default_registry_sync() -> AgentCardRegistry

Return the singleton, creating it if needed (synchronous).

Safe to call from pydantic validators and other sync contexts (e.g. config-parse time) because :class:AgentCardRegistry.__init__ does no I/O. If import-time bootstrap already provisioned the L2 MemorySpace, the constructor attaches write-behind even when this is called on a running event loop.

Source code in datarobot_genai/dragent/agent_card_registry.py

def get_default_registry_sync() -> AgentCardRegistry:
    """Return the singleton, creating it if needed (synchronous).

    Safe to call from pydantic validators and other sync contexts
    (e.g. config-parse time) because :class:`AgentCardRegistry.__init__`
    does no I/O. If import-time bootstrap already provisioned the L2
    MemorySpace, the constructor attaches write-behind even when this is
    called on a running event loop.
    """
    if _RegistryHolder.instance is None:
        api_token, endpoint = _resolve_settings()
        _RegistryHolder.instance = AgentCardRegistry(
            config=AgentCardRegistryConfig(),
            api_token=api_token,
            endpoint=endpoint,
        )
    return _RegistryHolder.instance

reset_default_registry

reset_default_registry() -> None

Reset the module-level singleton (for testing).

Source code in datarobot_genai/dragent/agent_card_registry.py

def reset_default_registry() -> None:
    """Reset the module-level singleton (for testing)."""
    _RegistryHolder.instance = None