ard-sdk

Typed, async Python SDK for the Agentic Resource Discovery (ARD) v0.9 draft.

The package provides:

• lenient, round-trippable models for manifests and registry responses;
• explicit validation profiles for ingesting versus publishing;
• hardened static discovery across well-known, robots.txt, HTML and optional DNS rungs;
• a scoped client for /search, /explore and /agents;
• safe cursor ownership and bounded auto-paging;
• federated search with explicit source groups, partial failures and lossless URN deduplication;
• a strict manifest publisher and dependency-free registry ASGI adapter;
• an in-process MockRegistry for client tests.

Python 3.13 or newer is required. ARD is still a draft, so the SDK remains 0.x and may make breaking changes as the normative artifacts converge.

Install

uv add ard-sdk

DNS SVCB/TXT discovery is optional:

uv add 'ard-sdk[dns]'

httpx is currently a core dependency; there is no [http] extra.

Discover a domain

import asyncio

from ard.http import ArdClient


async def main() -> None:
    async with ArdClient() as ard:
        found = await ard.discover("acme.com")
        for discovered in found.manifests:
            print(discovered.url, discovered.mechanism)

        # Domain -> manifest -> application/ai-registry+json entries.
        for registry in ard.registries_in(found):
            page = await registry.search("flight booking agent")
            for hit in page:
                print(hit.display_name, hit.score, hit.source)


asyncio.run(main())

An empty settled discovery result means the domain advertises no ARD. settled=False means some rung could not answer, so absence was not established.

Resolve every discovered manifest and its nested catalogs as one bounded graph:

resolved = await ard.resolve_domain("acme.com", max_depth=3, max_fetches=100)
for entry in resolved.entries:
    print(entry.identifier, resolved.source_for(entry.identifier))

for failure in resolved.errors:  # partial failures never erase successful branches
    print(failure.source, failure.message)

resolve_domain() calls discovery once and reuses its parsed roots; the fetch budget applies to nested URL catalogs across all roots. Its default PUBLIC_WEB policy requires HTTPS, public addresses, no userinfo, and no redirects. For one already-known manifest URL, use resolve_catalog(url, recursive=True, policy=PUBLIC_WEB); recursion is off there by default.

Query one registry

Credentials are scoped to a registry client; they are never ambient on ArdClient:

async with ArdClient() as ard:
    registry = ard.registry("https://registry.acme.com/api/v1/", token="secret")

    page = await registry.search(
        "book a flight",
        filter={"type": ["application/a2a-agent-card+json"]},
        page_size=20,
    )

    async for page in registry.pages("book a flight", max_pages=10, page_size=20):
        for hit in page:
            print(hit.identifier)

SearchPage.next() and RegistryClient.pages() resend only cursors issued by that registry. Page caps and repeated-token detection prevent an untrusted server from creating an infinite walk.

Search several registries

Scores from different registries are not comparable. Federation therefore returns one explicit group per queried source and preserves the source's native order:

from ard import FederationMode

async with ArdClient() as ard:
    internal = ard.registry("https://internal.example/api", token="internal-secret")
    public = ard.registry("https://public.example/api")

    results = await ard.search(
        "book a flight",
        registries=[internal, public],
        federation=FederationMode("referrals"),
        max_pages=2,
        max_concurrency=5,
    )

    for group in results:
        print(group.source, group.complete, group.next_token)
        for hit in group:
            print(hit.display_name, hit.score, hit.source)

    for failure in results.errors:
        print(failure.source, failure.error)

    # Secondary identity index: every source's complete metadata variant is retained.
    for identifier, same_resource in results.by_urn.items():
        print(identifier, [(hit.source, hit.result.display_name) for hit in same_resource.hits])

federation="referrals" asks registries to return referrals but does not follow them. Following is a separate operator decision:

results = await ard.search(
    "book a flight",
    registries=[internal],
    federation=FederationMode("referrals"),
    follow_referrals=True,  # explicit accept-all; bounded by max_referrals
)

For production trust rules, pass referral_policy=. It receives each referral and returns either None or the exact RegistryClient approved for that peer. Automatically followed referrals use a separate anonymous HTTP pool, preventing borrowed headers, cookies, default auth and TLS client identity from crossing the referral boundary. Strict PUBLIC_WEB discovery and resolution use the same isolation; pass an uncredentialed anonymous_http= when public traffic needs custom transport configuration.

Parse and validate manifests

from ard import Manifest, Profile, validate

manifest = Manifest.model_validate_json(body)
report = validate(manifest, Profile.INGEST)

for issue in report.issues:
    print(issue.severity, issue.code, issue.path)

Received documents preserve unknown fields because the v0.9 prose, CDDL, JSON Schema and OpenAPI currently disagree. Requests constructed by the SDK reject unknown fields.

Type an artifact in your application

ARD owns the envelope, not MCP, A2A or another artifact's schema. CatalogEntry.type remains open and inline data remains a mapping, so validate it directly with the model from that protocol's package:

card = MCPServerCard.model_validate(entry.data) if entry.data is not None else None

For a referenced artifact, the application chooses its own authentication, transport and decoder. No SDK codec registry or artifact-fetch policy sits between them.

Publish a catalog

The authoring surface makes reference-versus-inline delivery explicit and validates with the strict publish profile before producing output:

from ard import CatalogEntry
from ard.publish import CatalogBuilder, MediaType

weather = CatalogEntry.model_validate(
    {
        "identifier": "urn:air:acme.com:server:weather",
        "displayName": "Weather",
        "type": MediaType.MCP_SERVER_CARD,
        "url": "https://api.acme.com/weather.json",
        "capabilities": ["WeatherTool"],
        "representativeQueries": ["weather now", "forecast tomorrow"],
    }
)

catalog = CatalogBuilder(host="Acme AI", identifier="did:web:acme.com").entry(weather).build()

catalog.write_well_known("public")  # public/.well-known/ai-catalog.json
app = catalog.asgi()  # optional dependency-free dynamic route

Serve a registry

ArdRegistry is a dependency-free ASGI adapter. Handler inputs already contain endpoint defaults and clamped limits; the adapter injects result sources and owns wire validation:

from ard.server import ArdRegistry, SearchHit, SearchPage

registry = ArdRegistry(base_url="https://registry.acme.com/api/v1")


@registry.search
async def search(query):
    hits = await index.search(query.text, query.filter, limit=query.page_size)
    return SearchPage([SearchHit(hit.entry, hit.score) for hit in hits])


app = registry.asgi()

Omit the optional @registry.explore handler and the adapter returns the required 501 response. @registry.list receives the specification's undefined filter syntax as an opaque string. The official upstream manifest and in-process registry conformance modes run in CI.

Development

uv sync
uv run pytest
uv run ruff check src tests
uv run ruff format --check src tests
uv run mypy src tests
uv run lint-imports

The design baseline and known specification drift are documented in docs/.

Release

Releases are built from a clean main commit that has passed CI. The package version and source tag must agree: version 0.1.0 is tagged v0.1.0.

uv build
uv run --with twine twine check dist/*
uv publish dist/*

After publication, verify the supported boundary from a clean environment by installing the version range used by downstream applications: ard-sdk>=0.1,<0.2.