Created comprehensive MkDocs-formatted API documentation under docs/registry/ covering all public API surfaces from the registry subsystem with real-life constructive examples: - index.md: Architecture overview, module relationships, quickstart - types.md: PackageType, PackageId, PackageReference, PackageContent - client.md: RegistryClient with all 4 endpoints, auth modes, async patterns - canonical.md: Canonicalizer pipeline — NFC, RFC-8785, SHA-1, lifecycle stripping - resolver.md: ReferenceResolver — 3 reference schemes, version alias resolution - exceptions.md: RegistryError hierarchy — all 9 typed exceptions with HTTP mapping - cache.md: RegistryCache — LRU eviction, TTL, SHA-1 tamper detection, singleflight - integration.md: 4 end-to-end workflows (ordering pipeline, email categorization, CI/CD verification, multi-tenant provisioning) Updated mkdocs.yml nav tree with docs/registry/ entries. Refs: #51
8.6 KiB
Exceptions
Typed exception hierarchy for the Package Registry Client, mapping server-side error types from Package Registry Standard v1.0.0 §13.2 to Python exception classes.
from cleveractors.registry.exceptions import RegistryError
Class Hierarchy
CleverAgentsException
└── RegistryError # Base — all registry errors
├── PackageNotFoundError # 404 — package does not exist
├── InvalidPackageIdError # 400 / local — bad Package ID format
├── InvalidPackageReferenceError # 400 / local — bad reference string
│ └── CircularLocalReferenceError # §11.3.2 — always fatal
├── VersionNotFoundError # 404 — version/alias cannot be resolved
├── ValidationError # 400 — package failed validation
├── AuthenticationRequiredError # 401 — auth needed
├── AccessDeniedError # 403 — insufficient permissions
├── ConflictError # 409 — duplicate or conflicting data
└── RegistryNetworkError # 5xx + transport — connection/timeout/5xx
RegistryError (base)
class RegistryError(CleverAgentsException):
message: str # Human-readable description
details: dict[str, Any] | None # Optional supplementary info from server
original_reference: str | None # The string that triggered the error
All subclasses inherit message, details, and original_reference.
Exception Reference
PackageNotFoundError
The requested package does not exist in the registry (HTTP 404, type "PackageNotFound").
from cleveractors.registry.exceptions import PackageNotFoundError
try:
content = await client.get_package("pkg_act_nonexistent")
except PackageNotFoundError as exc:
print(f"Package not found: {exc.message}")
InvalidPackageIdError
The provided Package ID format is invalid (HTTP 400, type "InvalidPackageId").
Raised locally by PackageId.from_string() and PackageId.__post_init__() when the
ID does not match the pkg_<type>_<40-hex-sha1> format:
from cleveractors.registry.exceptions import InvalidPackageIdError
try:
pid = PackageId.from_string("bad-format")
except InvalidPackageIdError as exc:
print(f"Bad ID: {exc.message}")
InvalidPackageReferenceError
The provided package reference string could not be parsed or resolved (HTTP 400,
type "InvalidPackageReference").
Raised by ReferenceResolver.parse() and ReferenceResolver.resolve() when
the reference string is malformed, the referenced resource cannot be found, or
the required backend (client or local store) is not configured.
CircularLocalReferenceError
A circular local: reference chain was detected (§11.3.2).
Always raised regardless of lenient-mode settings, because cycles represent a
correctness violation, not a graceful-degradation scenario. This is a subclass of
InvalidPackageReferenceError — catch the parent class to handle both cycles
and other reference failures uniformly.
VersionNotFoundError
The requested version or version alias does not exist (HTTP 404, type "VersionNotFound").
from cleveractors.registry.exceptions import VersionNotFoundError
try:
result = await client.resolve_package(
package_type="actor",
namespace="acme",
name="my-agent",
version="v99.0.0", # does not exist
)
except VersionNotFoundError as exc:
print(f"Version not found: {exc.message}")
ValidationError
The package failed server-side validation (HTTP 400, type "ValidationError").
AuthenticationRequiredError
Authentication is required for this operation (HTTP 401, type "AuthenticationRequired").
AccessDeniedError
Access to the requested resource is denied (HTTP 403, type "AccessDenied").
Authenticated, but the credential lacks the required permission.
ConflictError
The request conflicts with existing data (HTTP 409, type "Conflict").
Typically raised when attempting to publish a duplicate Package ID.
RegistryNetworkError
Connection errors, timeouts, DNS failures, or HTTP 5xx server errors.
class RegistryNetworkError(RegistryError):
status_code: int | None # HTTP status for server errors
url: str | None # The URL being accessed
from cleveractors.registry.exceptions import RegistryNetworkError
try:
content = await client.get_package("pkg_act_...")
except RegistryNetworkError as exc:
print(f"Network error: {exc.message}")
print(f"Status: {exc.status_code}, URL: {exc.url}")
HTTP Status Mapping
The exception_for_status function maps HTTP status codes to exception classes:
from cleveractors.registry.exceptions import exception_for_status
exc = exception_for_status(404, "Package not found")
print(type(exc).__name__) # "PackageNotFoundError"
| Status | Exception |
|---|---|
| 400 | InvalidPackageIdError |
| 401 | AuthenticationRequiredError |
| 403 | AccessDeniedError |
| 404 | PackageNotFoundError |
| 409 | ConflictError |
| 500–599 | RegistryNetworkError |
| other | RegistryError |
Server Error Type to Exception Map
The internal _ERROR_TYPE_MAP also maps server-side error type strings
(from structured error responses §13.1) to exception classes for automatic
conversion:
_ERROR_TYPE_MAP = {
"PackageNotFound": PackageNotFoundError,
"InvalidPackageId": InvalidPackageIdError,
"InvalidPackageReference": InvalidPackageReferenceError,
"VersionNotFound": VersionNotFoundError,
"ValidationError": ValidationError,
"AuthenticationRequired": AuthenticationRequiredError,
"AccessDenied": AccessDeniedError,
"Conflict": ConflictError,
"InternalServerError": RegistryNetworkError,
}
Real-Life Example: Email Categorization System
Catching and handling specific exception types in a system that resolves categorization templates from a registry:
import asyncio
from cleveractors.registry import RegistryClient
from cleveractors.registry.exceptions import (
AccessDeniedError,
AuthenticationRequiredError,
PackageNotFoundError,
RegistryError,
RegistryNetworkError,
VersionNotFoundError,
)
async def load_categorization_templates(
client: RegistryClient,
) -> list[dict]:
"""
Load email categorization templates from the registry.
Handles specific error types with appropriate recovery strategies.
"""
templates = []
template_refs = [
("skill", "acme", "email-categorizer", "v1.x"),
("skill", "acme", "spam-detector", "latest"),
("skill", "acme", "priority-classifier", "v2.0.0"),
]
for pkg_type, ns, name, version in template_refs:
try:
result = await client.resolve_package(
package_type=pkg_type,
namespace=ns,
name=name,
version=version,
)
content = await client.get_package(result["package_id"])
templates.append(content)
except PackageNotFoundError:
print(f"Template {ns}/{name} not registered — skipping")
continue
except VersionNotFoundError:
print(
f"Version {version} of {ns}/{name} not found — "
f"trying latest..."
)
try:
result = await client.resolve_package(
package_type=pkg_type, namespace=ns, name=name
)
content = await client.get_package(result["package_id"])
templates.append(content)
except RegistryError as fallback_exc:
print(f"Fallback also failed: {fallback_exc}")
except AuthenticationRequiredError:
print("API key expired or missing — cannot proceed")
break
except AccessDeniedError:
print(f"Access denied to {ns}/{name} — check permissions")
continue
except RegistryNetworkError as exc:
print(
f"Network failure (status={exc.status_code}) — "
f"will retry on next cycle"
)
continue
return templates
if __name__ == "__main__":
async def main() -> None:
async with RegistryClient(
base_url="https://registry.example.com",
api_key="service-api-key",
) as client:
templates = await load_categorization_templates(client)
print(f"Loaded {len(templates)} templates")
asyncio.run(main())