ard-sdk

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

Unaffiliated community implementation. This SDK is an independent, third-party project. It is not affiliated with, endorsed by, or sponsored by the ARD specification authors, the ARD working group, or any of its stakeholders. "Agentic Resource Discovery" and "ARD" refer to the draft specification this package targets; all trademarks remain with their owners. Spec quotations are for interoperability only.

Highlights

Area	What you get
🧱 Models	Lenient, round-trippable models for manifests and registry responses — unknown fields preserved.
✅ Validation	Explicit INGEST vs. PUBLISH profiles; received docs are lenient, requests are strict.
🔎 Discovery	Hardened static ladder: well-known, robots.txt, HTML, and an optional DNS rung.
📡 Client	Scoped client for /search, /explore and /agents, with safe cursor ownership and bounded auto-paging.
🔗 Federation	Explicit source groups, partial failures, and lossless URN deduplication.
🛡️ Trust	Operator-owned verify() returning a TrustReport — evidence, never a policy decision.
📤 Publish	Strict manifest publisher (CatalogBuilder) and a dependency-free registry ASGI adapter.
🧪 Testing	In-process MockRegistry with scripted faults and referrals 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.

Verify a catalog entry

ArdClient.verify() returns a TrustReport of independent evidence — identity/authority binding, optional signatures, attestations and provenance — without reading the search relevance score or making an accept/reject decision. That decision is the application's: the SDK reports what it could establish, never auto-rejecting on missing evidence.

from ard.trust import TrustVerdict

async with ArdClient() as ard:
    report = await ard.verify(entry)

    print(report.identity_domain, report.authority_binding.status)
    print(report.overall)            # TrustVerdict.VERIFIED / UNVERIFIED / FAILED

    if report.overall is TrustVerdict.FAILED:
        # a present claim contradicted its evidence — distinct from "no claim made"
        ...

By default an identity below the publisher domain is accepted, bounded by a Public Suffix List check; pass strict=True to require an exact domain match. Signatures, attestations and provenance are marked UNVERIFIED until you supply the corresponding signature_verifier= / fetch_attestations= arguments — they are never waved through merely because the JSON fields exist. The pure, I/O-free form ard.trust.verify(entry) does no network calls and covers the authority-binding phase implemented today.

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.

Test against an in-process registry

MockRegistry spins up the same ArdRegistry adapter with deterministic in-memory handlers, and mock.client() returns an httpx.AsyncClient wired straight to it — no sockets. Hand that client to ArdClient(http=...) and your client code talks to the mock:

from ard import CatalogEntry
from ard.http import ArdClient
from ard.testing import MockRegistry

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

mock = MockRegistry([weather])

async with ArdClient(http=mock.client()) as ard:
    registry = ard.registry(mock.base_url)
    page = await registry.search("weather")
    for hit in page:
        print(hit.display_name, hit.score)

    # Every outbound request is recorded — the assertion surface for your tests.
    assert any(b"/search" in r.url.path for r in mock.requests)

scripted= injects faults and malformed responses per endpoint, referrals= populates federation responses, and explore=True / listing=True enable the optional handlers — so client-side retry, federation and error-mapping paths can be exercised without a live server. repeat_page_token=True simulates a misbehaving registry that re-emits the cursor it was just given, so you can assert your pages() cap holds.

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.2.0 is tagged v0.2.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.2,<0.3.