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
6.3 KiB
ReferenceResolver
Parses and resolves package references in all 3 supported schemes, implementing Package Registry Standard v1.0.0 §5.3 and §4.2.
from cleveractors.registry import ReferenceResolver
Constructor
ReferenceResolver(
client: RegistryClient | None = None,
local_store: LocalPackageStore | None = None,
)
| Parameter | Type | Default | Description |
|---|---|---|---|
client |
RegistryClient | None |
None |
Client for resolving registry: references |
local_store |
LocalPackageStore | None |
None |
Store for resolving local: references |
Both are optional. Resolution fails with InvalidPackageReferenceError if you
attempt to resolve a reference type without the corresponding backend configured.
Reference Schemes
Three reference schemes are supported per §5.3:
| Scheme | Format | Backend Required |
|---|---|---|
| Registry | server:namespace/name@version |
RegistryClient |
| ID | ID:pkg_<type>_<40-hex-sha1> |
None (inline parsing) |
| Local | local:<path> |
LocalPackageStore |
Static Methods
parse
@staticmethod
def parse(ref_str: str) -> PackageReference
Parse a reference string into a PackageReference. Delegates to
PackageReference.from_string and wraps ValueError into
InvalidPackageReferenceError.
ref = ReferenceResolver.parse("registry.example.com:acme/agent@v1.0.0")
print(ref.reference_type) # ReferenceType.REGISTRY
print(ref.namespace) # "acme"
resolve_version
def resolve_version(
version: str,
available_versions: list[str],
) -> str
Resolve a version string (concrete or alias) against available versions per §4.2.
Static helper; does not require a ReferenceResolver instance.
| Version Format | Behavior |
|---|---|
vX.Y.Z |
Concrete: matched directly |
latest, vx, x |
Global alias: resolves to newest concrete version |
vX.x |
Major alias: resolves to newest with major X |
vX.Y.x |
Minor alias: resolves to newest Z in X.Y series |
from cleveractors.registry import resolve_version
available = ["v1.0.0", "v1.0.1", "v1.1.0", "v2.0.0"]
print(resolve_version("v1.0.0", available)) # "v1.0.0"
print(resolve_version("latest", available)) # "v2.0.0"
print(resolve_version("v1.x", available)) # "v1.1.0"
print(resolve_version("v1.0.x", available)) # "v1.0.1"
is_concrete_version / is_version_alias
def is_concrete_version(version: str) -> bool
def is_version_alias(version: str) -> bool
print(is_concrete_version("v1.2.3")) # True
print(is_concrete_version("latest")) # False
print(is_version_alias("latest")) # True
print(is_version_alias("v3.x")) # True
print(is_version_alias("v1.2.3")) # False
print(is_version_alias("unknown")) # False
Async Methods
resolve
async def resolve(
self,
ref_str: str,
package_type: str = "actor",
) -> PackageId
Parse and resolve a reference string to a concrete PackageId.
Resolution strategy by reference type:
| Type | Strategy |
|---|---|
| ID | Parse the Package ID string directly — no network call |
| Local | Resolve via LocalPackageStore from the filesystem |
| Registry | Query the configured RegistryClient to resolve by name/namespace/version |
from cleveractors.registry import RegistryClient, ReferenceResolver
client = RegistryClient(base_url="https://registry.example.com")
resolver = ReferenceResolver(client=client)
# ID reference — resolved inline, no network call
pid = await resolver.resolve(
"ID:pkg_act_a1b2c3d4e5f67890abcdef1234567890abcdef"
)
print(pid.sha1_hex)
# Registry reference — queries the server
pid = await resolver.resolve(
"registry.example.com:acme/web-search@v1.0.0",
package_type="skill",
)
print(pid.id_string)
await resolver.close()
Async Context Manager
async with ReferenceResolver(client=client) as resolver:
pid = await resolver.resolve("ID:pkg_act_...")
# resolver.close() called automatically
Real-Life Example: Multi-Tenant SaaS Provisioning
Resolving agent templates per tenant namespace, detecting and rejecting malformed references:
import asyncio
from cleveractors.registry import (
RegistryClient,
ReferenceResolver,
)
from cleveractors.registry.exceptions import (
InvalidPackageReferenceError,
PackageNotFoundError,
)
from cleveractors.registry.types import PackageReference, ReferenceType
TENANT_AGENTS = {
"tenant-alpha": "registry.example.com:alpha/categorization-agent@v2.x",
"tenant-beta": "registry.example.com:beta/categorization-agent@v1.x",
}
async def provision_tenant(tenant_id: str) -> None:
ref_str = TENANT_AGENTS.get(tenant_id)
if ref_str is None:
raise ValueError(f"Unknown tenant: {tenant_id}")
async with RegistryClient(
base_url="https://registry.example.com"
) as client:
resolver = ReferenceResolver(client=client)
# Step 1: Parse — validates reference format
try:
ref = resolver.parse(ref_str)
except InvalidPackageReferenceError as exc:
print(f"Invalid reference for {tenant_id}: {exc}")
return
# Step 2: Verify it is a registry reference (not ID, not local)
if ref.reference_type != ReferenceType.REGISTRY:
print(
f"Expected registry reference for {tenant_id}, "
f"got {ref.reference_type.value}"
)
return
# Step 3: Resolve — queries registry and handles version alias
try:
package_id = await resolver.resolve(
ref_str, package_type="actor"
)
except PackageNotFoundError as exc:
print(f"Package not found for {tenant_id}: {exc}")
return
except InvalidPackageReferenceError as exc:
print(
f"Resolution failed for {tenant_id}: {exc}"
)
return
print(
f"[{tenant_id}] Resolved {ref.namespace}/{ref.name}"
f"@{ref.version} → {package_id.id_string}"
)
if __name__ == "__main__":
asyncio.run(provision_tenant("tenant-alpha"))