Skip to content

reach.registry

Google Cloud Agent Registry client and skill manifest models.

Provide client access, authentication, and two-tier caching for Google Cloud Agent Registry.

AuthenticationError

Bases: RegistryError

Raised when authentication credentials cannot be obtained or are invalid.

Source code in src/reach/registry.py 
61
62
class AuthenticationError(RegistryError):
    """Raised when authentication credentials cannot be obtained or are invalid."""

NotFoundError

Bases: RegistryError

Raised when target project, location, or skill is not found (404).

Source code in src/reach/registry.py 
73
74
class NotFoundError(RegistryError):
    """Raised when target project, location, or skill is not found (404)."""

PermissionDeniedError

Bases: RegistryError

Raised when caller lacks required IAM permissions (403 Forbidden).

Source code in src/reach/registry.py 
65
66
class PermissionDeniedError(RegistryError):
    """Raised when caller lacks required IAM permissions (403 Forbidden)."""

RegistryClient

HTTP client for querying the Google Cloud Agent Registry REST API.

Source code in src/reach/registry.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
class RegistryClient:
    """HTTP client for querying the Google Cloud Agent Registry REST API."""

    def __init__(
        self,
        token: str | None = None,
        base_url: str = "https://agentregistry.googleapis.com/v1alpha",
        max_retries: int = 3,
        backoff_factor: float = 0.5,
        token_ttl_seconds: float = 3000.0,
        allow_custom_host: bool = False,
    ) -> None:
        """Initialize RegistryClient with optional bearer token and retry settings."""
        self.token = token
        self.base_url = self._validate_base_url(base_url, allow_custom_host=allow_custom_host)
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.token_ttl_seconds = token_ttl_seconds
        self._cached_token: str | None = None
        self._cached_token_expiry: float = 0.0

    @staticmethod
    def _validate_base_url(base_url: str, *, allow_custom_host: bool = False) -> str:
        """Validate that base_url uses HTTPS and points to an approved Google domain or loopback."""
        clean_url = base_url.strip().rstrip("/")
        parsed = urllib.parse.urlsplit(clean_url)
        if not parsed.scheme or not parsed.hostname:
            msg = f"Invalid Agent Registry base_url: {base_url!r}"
            raise ValueError(msg)

        if parsed.scheme not in ("http", "https"):
            msg = f"Agent Registry base_url must use HTTP or HTTPS scheme, got: {parsed.scheme!r}"
            raise ValueError(msg)

        is_loopback = parsed.hostname in ("127.0.0.1", "localhost", "::1")
        if parsed.scheme != "https" and not is_loopback:
            msg = f"Agent Registry base_url must use HTTPS for non-local endpoints: {base_url!r}"
            raise ValueError(msg)

        is_google_api = (
            parsed.hostname == "googleapis.com"
            or parsed.hostname.endswith(".googleapis.com")
            or parsed.hostname.endswith(".google.com")
        )
        if not is_loopback and not is_google_api and not allow_custom_host:
            msg = (
                f"Agent Registry base_url host must be a Google API domain or loopback, "
                f"got: {parsed.hostname!r}"
            )
            raise ValueError(msg)

        return clean_url

    def _get_token(self) -> str:
        """Resolve active authentication token, fetching dynamically or from cache."""
        if self.token:
            return self.token

        now = time.time()
        if self._cached_token and now < self._cached_token_expiry:
            return self._cached_token

        token = get_access_token()
        self._cached_token = token
        self._cached_token_expiry = now + self.token_ttl_seconds
        return token

    def _request(self, endpoint: str, params: dict[str, str] | None = None) -> dict[str, Any]:
        """Execute an authenticated GET request with rate-limit retry and error mapping."""
        query_string = f"?{urllib.parse.urlencode(params)}" if params else ""
        url = f"{self.base_url}/{endpoint.lstrip('/')}{query_string}"
        token = self._get_token()

        for attempt in range(self.max_retries + 1):
            req = urllib.request.Request(  # noqa: S310
                url,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Accept": "application/json",
                    "User-Agent": "skill-reach",
                },
            )
            try:
                with urllib.request.urlopen(req) as resp:  # noqa: S310
                    raw = resp.read().decode("utf-8")
                    return json.loads(raw)  # type: ignore[no-any-return]
            except urllib.error.HTTPError as http_err:
                status = http_err.code
                error_body = ""
                with contextlib.suppress(OSError):
                    error_body = http_err.read().decode("utf-8")
                http_err.close()

                if status == HTTPStatus.TOO_MANY_REQUESTS and attempt < self.max_retries:
                    sleep_time = self.backoff_factor * (2**attempt)
                    time.sleep(sleep_time)
                    continue

                self._handle_http_error(status, error_body, endpoint)
            except urllib.error.URLError as url_err:
                msg = f"Failed to connect to Agent Registry endpoint ({url}): {url_err.reason}"
                raise RegistryError(msg) from url_err

        msg = f"Agent Registry request exceeded maximum retries for {url}"
        raise RegistryError(msg)

    def _handle_http_error(self, status: int, body: str, endpoint: str) -> None:
        """Map HTTP error status codes to descriptive domain exceptions."""
        error_details = {}
        error_message = body
        with contextlib.suppress(json.JSONDecodeError, UnicodeDecodeError, AttributeError):
            parsed = json.loads(body)
            if isinstance(parsed, dict) and "error" in parsed:
                err_dict = parsed["error"]
                error_message = err_dict.get("message", body)
                error_details = err_dict

        if status == HTTPStatus.UNAUTHORIZED:
            self._cached_token = None
            self._cached_token_expiry = 0.0
            msg = (
                "Authentication failed when contacting Agent Registry. "
                "Run 'gcloud auth application-default login' to refresh credentials."
            )
            raise AuthenticationError(msg)

        if status == HTTPStatus.FORBIDDEN:
            # Check for service disabled
            details_list = error_details.get("details", [])
            for item in details_list:
                if isinstance(item, dict) and item.get("reason") == "SERVICE_DISABLED":
                    msg = (
                        "The Agent Registry API (agentregistry.googleapis.com) is disabled. "
                        "Enable it with: gcloud services enable agentregistry.googleapis.com"
                    )
                    raise ServiceDisabledError(msg)

            msg = (
                f"Permission denied on Agent Registry ({endpoint}). "
                "Ensure your account has the 'roles/agentregistry.user' role on the project."
            )
            raise PermissionDeniedError(msg)

        if status == HTTPStatus.NOT_FOUND:
            msg = f"Agent Registry resource not found ({endpoint}): {error_message}"
            raise NotFoundError(msg)

        msg = f"Agent Registry request failed with HTTP {status}: {error_message}"
        raise RegistryError(msg)

    def list_skills(
        self,
        project: str,
        location: str = "global",
        publisher: str | None = None,
    ) -> list[RegistrySkillData]:
        """Fetch all registered skills in the specified project and location."""
        endpoint = f"projects/{project}/locations/{location}/skills"
        skills: list[RegistrySkillData] = []
        page_token: str | None = None

        while True:
            params: dict[str, str] = {}
            if page_token:
                params["pageToken"] = page_token

            data = self._request(endpoint, params=params or None)
            raw_skills = data.get("skills", [])
            for s in raw_skills:
                try:
                    parsed = RegistrySkillData.model_validate(s)
                    if not _matches_publisher(parsed.publisher, publisher):
                        continue
                    skills.append(parsed)
                except (ValidationError, ValueError):
                    continue

            page_token = data.get("nextPageToken")
            if not page_token:
                break

        return skills

    def get_skill(
        self,
        project: str,
        location: str,
        skill_id: str,
    ) -> RegistrySkillData:
        """Fetch a single skill resource from Agent Registry."""
        endpoint = f"projects/{project}/locations/{location}/skills/{skill_id}"
        data = self._request(endpoint)
        return RegistrySkillData.model_validate(data)

    def fetch_manifest(
        self,
        project: str,
        location: str = "global",
        publisher: str | None = None,
    ) -> RegistryManifest:
        """Compile a complete RegistryManifest snapshot for project and location."""
        raw_skills = self.list_skills(project=project, location=location, publisher=publisher)
        return RegistryManifest(
            project=project,
            location=location,
            publisher=publisher,
            fetched_at=datetime.now(UTC),
            skills=tuple(raw_skills),
        )

__init__ 

__init__(
    token: str | None = None,
    base_url: str = "https://agentregistry.googleapis.com/v1alpha",
    max_retries: int = 3,
    backoff_factor: float = 0.5,
    token_ttl_seconds: float = 3000.0,
    allow_custom_host: bool = False,
) -> None

Initialize RegistryClient with optional bearer token and retry settings.

Source code in src/reach/registry.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def __init__(
    self,
    token: str | None = None,
    base_url: str = "https://agentregistry.googleapis.com/v1alpha",
    max_retries: int = 3,
    backoff_factor: float = 0.5,
    token_ttl_seconds: float = 3000.0,
    allow_custom_host: bool = False,
) -> None:
    """Initialize RegistryClient with optional bearer token and retry settings."""
    self.token = token
    self.base_url = self._validate_base_url(base_url, allow_custom_host=allow_custom_host)
    self.max_retries = max_retries
    self.backoff_factor = backoff_factor
    self.token_ttl_seconds = token_ttl_seconds
    self._cached_token: str | None = None
    self._cached_token_expiry: float = 0.0

fetch_manifest 

fetch_manifest(
    project: str,
    location: str = "global",
    publisher: str | None = None,
) -> RegistryManifest

Compile a complete RegistryManifest snapshot for project and location.

Source code in src/reach/registry.py 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
def fetch_manifest(
    self,
    project: str,
    location: str = "global",
    publisher: str | None = None,
) -> RegistryManifest:
    """Compile a complete RegistryManifest snapshot for project and location."""
    raw_skills = self.list_skills(project=project, location=location, publisher=publisher)
    return RegistryManifest(
        project=project,
        location=location,
        publisher=publisher,
        fetched_at=datetime.now(UTC),
        skills=tuple(raw_skills),
    )

get_skill 

get_skill(
    project: str, location: str, skill_id: str
) -> RegistrySkillData

Fetch a single skill resource from Agent Registry.

Source code in src/reach/registry.py 
427
428
429
430
431
432
433
434
435
436
def get_skill(
    self,
    project: str,
    location: str,
    skill_id: str,
) -> RegistrySkillData:
    """Fetch a single skill resource from Agent Registry."""
    endpoint = f"projects/{project}/locations/{location}/skills/{skill_id}"
    data = self._request(endpoint)
    return RegistrySkillData.model_validate(data)

list_skills 

list_skills(
    project: str,
    location: str = "global",
    publisher: str | None = None,
) -> list[RegistrySkillData]

Fetch all registered skills in the specified project and location.

Source code in src/reach/registry.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
def list_skills(
    self,
    project: str,
    location: str = "global",
    publisher: str | None = None,
) -> list[RegistrySkillData]:
    """Fetch all registered skills in the specified project and location."""
    endpoint = f"projects/{project}/locations/{location}/skills"
    skills: list[RegistrySkillData] = []
    page_token: str | None = None

    while True:
        params: dict[str, str] = {}
        if page_token:
            params["pageToken"] = page_token

        data = self._request(endpoint, params=params or None)
        raw_skills = data.get("skills", [])
        for s in raw_skills:
            try:
                parsed = RegistrySkillData.model_validate(s)
                if not _matches_publisher(parsed.publisher, publisher):
                    continue
                skills.append(parsed)
            except (ValidationError, ValueError):
                continue

        page_token = data.get("nextPageToken")
        if not page_token:
            break

    return skills

RegistryError

Bases: RuntimeError

Base exception for Google Cloud Agent Registry operations.

Source code in src/reach/registry.py 
57
58
class RegistryError(RuntimeError):
    """Base exception for Google Cloud Agent Registry operations."""

RegistryManifest

Bases: BaseModel

Persisted snapshot of skills metadata in a project and location.

Source code in src/reach/registry.py 
109
110
111
112
113
114
115
116
117
118
class RegistryManifest(BaseModel):
    """Persisted snapshot of skills metadata in a project and location."""

    model_config = ConfigDict(frozen=True)

    project: str
    location: str
    publisher: str | None = None
    fetched_at: datetime
    skills: tuple[RegistrySkillData, ...] = ()

RegistrySkillData

Bases: BaseModel

Represent raw skill metadata returned by the Agent Registry REST API.

Source code in src/reach/registry.py 
 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
103
104
105
106
class RegistrySkillData(BaseModel):
    """Represent raw skill metadata returned by the Agent Registry REST API."""

    model_config = ConfigDict(frozen=True, extra="ignore", populate_by_name=True)

    name: str
    display_name: str = Field(alias="displayName", default="")
    description: str = ""
    type: str = "SIMPLE"
    state: str = "STATE_ACTIVE"
    target_state: str = Field(alias="targetState", default="TARGET_STATE_ACTIVE")
    default_revision: str | None = Field(alias="defaultRevision", default=None)
    skill_id: str | None = Field(alias="skillId", default=None)
    publisher: str | None = None
    create_time: str | None = Field(alias="createTime", default=None)
    update_time: str | None = Field(alias="updateTime", default=None)

    @property
    def identifier(self) -> str:
        """Return the display name if available, otherwise trailing segment of resource name."""
        if self.display_name:
            return self.display_name
        return self.name.rsplit("/", 1)[-1]

    @property
    def revision_slug(self) -> str:
        """Extract the revision ID from defaultRevision or return 'default'."""
        if self.default_revision and "/" in self.default_revision:
            return self.default_revision.rsplit("/", 1)[-1]
        return "default"

identifier property 

identifier: str

Return the display name if available, otherwise trailing segment of resource name.

revision_slug property 

revision_slug: str

Extract the revision ID from defaultRevision or return 'default'.

ServiceDisabledError

Bases: RegistryError

Raised when the Agent Registry API is not enabled on the target project.

Source code in src/reach/registry.py 
69
70
class ServiceDisabledError(RegistryError):
    """Raised when the Agent Registry API is not enabled on the target project."""