cleveractors-core/docs/registry/client.md

RegistryClient

Async HTTP client implementing all 4 API endpoints of the Package Registry Standard v1.0.0 §8.

from cleveractors.registry import RegistryClient

---

Constructor

RegistryClient(
    base_url: str,
    api_key: Optional[str] = None,
    timeout: float = 30.0,
    allow_insecure: bool = False,
)

Parameter	Type	Default	Description
base_url	str	required	Registry server base URL (e.g. "https://registry.example.com")
api_key	str | None	None	Optional API key for authenticated operations (§9.2)
timeout	float	30.0	Request timeout in seconds
allow_insecure	bool	False	Suppress the HTTP-without-TLS security warning

The constructor emits a warning when an API key is used over HTTP without allow_insecure=True, because the standard requires HTTPS in production (§12.2).

---

Endpoints

The client implements all 4 endpoints defined by the standard:

Method	Endpoint	Standard Ref
get_package(package_id)	GET /packages/{id}	§8.2.1
resolve_package(type, ns, name, version)	GET /{type}/{ns}/{name}?version={v}	§8.2.2
browse(type, namespace)	GET /browse	§8.4.1
discover()	GET /.well-known/cleverthis-packages	§8.4.2

All methods are async and return dictionaries. They map HTTP errors to typed exceptions from the exception hierarchy.

---

get_package

async def get_package(self, package_id: str) -> dict[str, Any]

Retrieve raw package content by its globally unique Package ID.

content = await client.get_package(
    "pkg_act_a1b2c3d4e5f67890abcdef1234567890abcdef"
)
print(content["name"])  # e.g. "order-validator"

Raises: PackageNotFoundError, InvalidPackageIdError, RegistryNetworkError.

---

resolve_package

async def resolve_package(
    self,
    package_type: str,
    namespace: str,
    name: str,
    version: Optional[str] = None,
) -> dict[str, Any]

Resolve a named package reference to its concrete Package ID. Performs version alias resolution according to §4.2.

# Resolve with a concrete version
result = await client.resolve_package(
    package_type="actor",
    namespace="acme",
    name="order-validator",
    version="v1.0.0",
)
print(result["package_id"])  # "pkg_act_..."

# Resolve with the `latest` alias (default when version is None)
latest = await client.resolve_package(
    package_type="skill",
    namespace="acme",
    name="code-analysis",
)
print(latest["package_id"])

Returns a dict with keys package_id (str) and type (str).

Raises: PackageNotFoundError, VersionNotFoundError, RegistryNetworkError.

---

browse

async def browse(
    self,
    package_type: Optional[str] = None,
    namespace: Optional[str] = None,
) -> dict[str, Any]

Browse published packages with optional type and namespace filters.

# List all actor packages
results = await client.browse(package_type="actor")
for pkg in results["packages"]:
    print(f"{pkg['name']} ({pkg['namespace']}) — {pkg['description']}")

# List all packages in a namespace
results = await client.browse(namespace="acme")
print(f"Found {results['count']} packages in acme namespace")

Returns a dict with keys packages (list of package summaries) and count (int).

Raises: RegistryNetworkError.

---

discover

async def discover(self) -> dict[str, Any]

Retrieve registry metadata via the well-known discovery endpoint (§8.4.2).

meta = await client.discover()
print(meta["name"])              # "CleverThis Package Registry"
print(meta["version"])           # "1.0.0"
print(meta["supported_types"])   # ["actor", "graph", "stream", ...]
print(meta["authentication"])    # ["anonymous", "api_key"]

Raises: RegistryNetworkError.

---

Authentication

The client supports two authentication modes defined in §9:

Mode	How to use	When
Anonymous	Omit api_key	Reading public packages
API Key	Pass api_key="..."	Publishing or accessing private packages

The API key is sent as a Bearer token in the Authorization header:

Authorization: Bearer <api_key>

---

Error Mapping

HTTP errors from the server are automatically converted to typed Python exceptions. See exceptions.md for the complete mapping.

HTTP Status	Exception
400	InvalidPackageIdError
401	AuthenticationRequiredError
403	AccessDeniedError
404	PackageNotFoundError
409	ConflictError
5xx	RegistryNetworkError

Server-side structured error responses (§13.1) are parsed and the type field is matched against the error-type-to-exception map, allowing the server to indicate specific error conditions (e.g. VersionNotFound) even when the HTTP status code is less specific.

---

Async Context Manager

RegistryClient supports async context management for automatic cleanup:

async with RegistryClient(base_url="https://registry.example.com") as client:
    content = await client.get_package("pkg_act_...")
# client.close() is called automatically on exit

Alternatively, call close() explicitly:

client = RegistryClient(base_url="https://registry.example.com")
try:
    content = await client.get_package("pkg_act_...")
finally:
    await client.close()

---

Real-Life Example: Supplier Agent Registration

Registering a supplier agent that validates part numbers in purchase orders by pulling an order-validation template from the package registry:

import asyncio
from typing import Any
from cleveractors.registry import RegistryClient, RegistryCache, CacheFactory


async def register_supplier_agent() -> dict[str, Any]:
    client = RegistryClient(
        base_url="https://registry.example.com",
        api_key="supplier-api-key",
    )
    cache = RegistryCache(client, max_size=256, ttl=300.0)

    # 1. Discover registry capabilities
    meta = await client.discover()
    if "actor" not in meta.get("supported_types", []):
        raise RuntimeError("Registry does not support actor packages")

    # 2. Resolve the order-validator template to its concrete ID
    resolved = await cache.resolve_package(
        package_type="actor",
        namespace="acme",
        name="order-validator",
        version="v1.0.0",
    )
    package_id = resolved["package_id"]

    # 3. Fetch the full template content
    content = await cache.get_package(package_id)
    print(f"Loaded template: {content.get('name', package_id)}")
    print(f"Stats — hits: {cache.stats.hits}, misses: {cache.stats.misses}")

    await cache.close()
    return content


if __name__ == "__main__":
    result = asyncio.run(register_supplier_agent())
    print(f"Agent config: {list(result.keys())}")