DNS-AID API Reference

July 30, 2026 · View on GitHub

Complete API documentation for DNS-AID - DNS-based Agent Identification and Discovery.

Relationship to IETF

This document describes the API of the DNS-AID reference implementation.

The DNS-AID specification is defined in the IETF draft: https://datatracker.ietf.org/doc/draft-mozleywilliams-dnsop-dnsaid/.

Quick Start
Core Functions
Domain Control Validation (DCV)
Data Models
Backends
JWS Signatures
Validation Utilities
CLI Reference
MCP Server
Invocation Module
SDK: Invocation & Telemetry

Quick Start

import asyncio
from dns_aid import publish, discover, verify, Protocol

async def main():
    # Publish an agent
    result = await publish(
        name="my-agent",
        domain="example.com",
        protocol="mcp",
        endpoint="agent.example.com",
        capabilities=["chat", "code-review"],
    )

    # Discover agents at a domain
    discovery = await discover("example.com", protocol=Protocol.MCP)

    # Verify an agent's DNS records
    verification = await verify("my-agent.example.com")

asyncio.run(main())

Core Functions

publish()

Publish an AI agent to DNS using this implementation of the DNS-AID specification.

async def publish(
    name: str,
    domain: str,
    protocol: str | Protocol,
    endpoint: str,
    port: int = 443,
    capabilities: list[str] | None = None,
    version: str = "1.0.0",
    description: str | None = None,
    ttl: int = 3600,
    backend: DNSBackend | None = None,
    cap_uri: str | None = None,
    cap_sha256: str | None = None,
    bap: list[str] | None = None,
    policy_uri: str | None = None,
    realm: str | None = None,
    ipv4_hint: str | None = None,
    ipv6_hint: str | None = None,
) -> PublishResult

Parameters

Parameter	Type	Required	Default	Description
`name`	`str`	Yes	-	Agent identifier (e.g., "chat", "network-specialist"). Must be DNS label format: lowercase alphanumeric with hyphens, 1-63 chars.
`domain`	`str`	Yes	-	Domain to publish under (e.g., "example.com")
`protocol`	`str \| Protocol`	Yes	-	Communication protocol: "mcp" or "a2a"
`endpoint`	`str`	Yes	-	Hostname where agent is reachable
`port`	`int`	No	443	Port number (1-65535)
`capabilities`	`list[str]`	No	`[]`	List of agent capabilities
`version`	`str`	No	"1.0.0"	Agent version (semver format)
`description`	`str`	No	`None`	Human-readable description
`ttl`	`int`	No	3600	DNS record TTL in seconds (60-86400)
`backend`	`DNSBackend`	No	`None`	DNS backend to use (uses default if None)
`cap_uri`	`str`	No	`None`	URI to capability document (DNS-AID custom param)
`cap_sha256`	`str`	No	`None`	Base64url SHA-256 digest of capability descriptor
`bap`	`list[str]`	No	`None`	Supported protocols with versions (e.g., `["mcp/1", "a2a/1"]`)
`policy_uri`	`str`	No	`None`	URI to agent policy document
`realm`	`str`	No	`None`	Multi-tenant scope identifier
`ipv4_hint`	`str`	No	`None`	IPv4 address hint for SVCB record (reduces A query round trips)
`ipv6_hint`	`str`	No	`None`	IPv6 address hint for SVCB record (reduces AAAA query round trips)

Returns

PublishResult - Contains the published agent and created DNS records.

Example

from dns_aid import publish

result = await publish(
    name="network-specialist",
    domain="example.com",
    protocol="mcp",
    endpoint="mcp.example.com",
    capabilities=["ipam", "dns", "vpn"],
    ttl=300,
)

if result.success:
    print(f"Published: {result.agent.fqdn}")
    print(f"Records: {result.records_created}")
else:
    print(f"Failed: {result.message}")

DNS Records Created

SVCB: {name}.{domain} (draft-02 flat primary owner) → Service binding record
SVCB (AliasMode, optional): {name}._agents.{domain} → walkable AliasMode pointer at the flat primary owner (suppressible via publish_walkable_alias=False)
TXT: {name}.{domain} → Capabilities and metadata (alongside the primary SVCB)

discover()

Discover AI agents at a domain using this implementation of the DNS-AID specification (Path A — DNS substrate).

async def discover(
    domain: str,
    protocol: str | Protocol | None = None,
    name: str | None = None,
    require_dnssec: bool = False,
    use_http_index: bool = False,
    enrich_endpoints: bool = True,
    verify_signatures: bool = False,
    trust_dnssec_pointers: bool = False,
    verify_dane: bool = False,
    *,
    # Path A in-memory filter kwargs (v0.19.0+)
    capabilities: list[str] | None = None,
    capabilities_any: list[str] | None = None,
    auth_type: str | None = None,
    intent: str | None = None,
    transport: str | None = None,
    realm: str | None = None,
    min_dnssec: bool = False,
    text_match: str | None = None,
    require_signed: bool = False,
    require_signature_algorithm: list[str] | None = None,
) -> DiscoveryResult

Parameters

Substrate parameters:

Parameter	Type	Required	Default	Description
`domain`	`str`	Yes	-	Domain to search for agents
`protocol`	`str \| Protocol`	No	`None`	Filter by protocol (None for all)
`name`	`str`	No	`None`	Filter by specific agent name (case-insensitive per RFC 1035)
`require_dnssec`	`bool`	No	`False`	Require every DNS-plane agent's DNS response to carry the resolver AD flag; raises `DNSSECError` if any does not. ARD / HTTP-catalog agents are exempt (no DNS SVCB record — their trust is `catalog_trust`). Exposed on SDK, CLI (`--require-dnssec`), and MCP.
`use_http_index`	`bool`	No	`False`	Use HTTP index endpoint instead of DNS-only discovery
`enrich_endpoints`	`bool`	No	`True`	Fetch cap docs / agent cards to enrich AgentRecords
`verify_signatures`	`bool`	No	`False`	Fetch JWKS and verify per-agent JWS signatures
`trust_dnssec_pointers`	`bool`	No	`False`	Opt-in. Also follow an off-domain ARD catalog pointer when its pointer record is DNSSEC-validated (AD flag). Off by default — the AD flag is only trustworthy with a validating resolver over a secure path. Exposed on SDK, CLI (`--trust-dnssec-pointers`), and MCP.
`verify_dane`	`bool`	No	`False`	Opt-in. Check each resolved agent endpoint's TLS certificate against its DANE/TLSA record — defense-in-depth on the endpoint that does not change the catalog/pointer trust decision. Demoted to `None` unless the agent's DNS response is DNSSEC-validated (DANE without DNSSEC carries no integrity guarantee, RFC 6698 §10.1). Surfaced on `AgentRecord.dane_verified`. SDK / CLI (`--verify-dane`) / MCP.

Filter kwargs (v0.19.0+, all keyword-only, all default no-op):

Parameter	Type	Description
`capabilities`	`list[str] \| None`	All-of capability match — every entry must be present on the agent. Empty list = no-match (explicit).
`capabilities_any`	`list[str] \| None`	Any-of capability match — at least one entry must be present. Empty list = no-match.
`auth_type`	`str \| None`	Case-insensitive exact match against `agent.auth_type`.
`intent`	`str \| None`	Match against `agent.category`; falls back to substring match across capabilities.
`transport`	`str \| None`	Match against the agent's protocol identifier (Path A surfaces protocol, not wire transport).
`realm`	`str \| None`	Exact match against `agent.realm`.
`min_dnssec`	`bool`	When `True`, only records whose DNS response was DNSSEC-validated (AD flag) pass. ARD / HTTP-catalog agents are exempt (no DNS SVCB record — their trust is `catalog_trust`) and pass through rather than being dropped.
`text_match`	`str \| None`	Case-insensitive substring match across `description`, `use_cases`, and `capabilities`. Empty string raises `ValueError`.
`require_signed`	`bool`	When `True`, only records whose JWS signature verified pass. Auto-enables `verify_signatures=True`.
`require_signature_algorithm`	`list[str] \| None`	Restrict `require_signed` matches to records whose verified algorithm is in this allow-list. Requires `require_signed=True`.

All filter predicates compose with logical AND. None / False means "no constraint". When every filter is unset, the input list is returned unchanged with no allocation (no-op fast path keeps existing callers free of overhead).

Discovery Methods

Method	Endpoint	Use Case
DNS (default)	`_index._agents.{domain}` TXT record	Decentralized, cached, minimal round trips
HTTP Index	`https://_index._aiagents.{domain}/index-wellknown`	ANS-compatible, rich metadata (descriptions, model cards)
ARD ai-catalog	`https://{domain}/.well-known/ai-catalog.json`	ARD catalogs, auto-detected via `use_http_index=True`; carries publisher trust manifests

With use_http_index=True the fetcher probes the legacy index locations first and the ARD well-known location last, then auto-detects the document format (specVersion: "1.0" + entries[] → ARD; keyed object → legacy). ARD entries whose artifact type is application/mcp-server-card+json or application/a2a-agent-card+json become agents (protocol inferred from the media type); inline nested catalogs recurse (depth ≤ 3); registry entries and non-agent artifacts are skipped. An entry's trustManifest is preserved on AgentRecord.trust_manifest (pass-through — dns-aid does not verify signatures, attestation digests, or identity↔publisher alignment).

Returns

DiscoveryResult - Contains list of discovered agents and query metadata.

Raises

ValueError — text_match is an empty string, or require_signature_algorithm is set without require_signed=True.
DNSSECError — require_dnssec=True but a DNS-plane agent's response is not authenticated (AD flag unset). ARD / HTTP-catalog agents are exempt.

Example

from dns_aid import discover, Protocol

# Discover all agents (pure DNS - default)
result = await discover("example.com")

# Discover MCP agents only
result = await discover("example.com", protocol=Protocol.MCP)

# Discover specific agent
result = await discover("example.com", protocol="mcp", name="chat")

# Discover via HTTP index (ANS-compatible, richer metadata)
result = await discover("example.com", use_http_index=True)

# Filter by capabilities + auth type (v0.19.0+)
result = await discover(
    "example.com",
    capabilities=["payment-processing", "fraud-detection"],
    auth_type="oauth2",
    realm="prod",
)

# Trust-gated discovery: only signed agents with allow-listed algorithms
result = await discover(
    "example.com",
    require_signed=True,
    require_signature_algorithm=["ES256", "Ed25519"],
)

for agent in result.agents:
    print(f"{agent.name}: {agent.endpoint_url}")
    print(f"  Capabilities: {', '.join(agent.capabilities)}")
    if agent.description:
        print(f"  Description: {agent.description}")

verify()

Verify DNS-AID records as interpreted by this implementation, with security validation.

async def verify(fqdn: str) -> VerifyResult

Parameters

Parameter	Type	Required	Description
`fqdn`	`str`	Yes	Fully qualified domain name (e.g., "chat.example.com")

Returns

VerifyResult - Contains security validation results and score.

Example

from dns_aid import verify

result = await verify("chat.example.com")

print(f"Record exists: {result.record_exists}")
print(f"DNSSEC valid: {result.dnssec_valid}")
print(f"Security Score: {result.security_score}/100")
print(f"Rating: {result.security_rating}")

Domain Control Validation (DCV)

DCV is a stateless challenge/verify primitive that lets one party prove control of a domain to another using a short-lived TXT record at _agents-challenge.{domain}. It implements draft-ietf-dnsop-domain-verification-techniques-12 plus the bnd-req binding extension from draft-mozleywilliams-dnsop-dnsaid-02.

Role split:

Challenger — calls issue() and verify(); no DNS write credentials required.
Claimant — calls place() and revoke(); needs backend write credentials for the domain.

Wire format (space-separated key=value at _agents-challenge.{domain} TXT):

token=<32-char-base32>  [domain=<domain>]  [bnd-req=svc:<agent>@<issuer>]  expiry=<RFC3339Z>

from dns_aid.core import dcv

# Challenger
challenge = dcv.issue("example.com", agent_name="assistant", issuer_domain="orga.test")
# ... deliver challenge to claimant out-of-band ...

# Claimant
await dcv.place(challenge.domain, challenge.token, bnd_req=challenge.bnd_req)

# Challenger
result = await dcv.verify(challenge.domain, challenge.token,
                          expected_bnd_req=challenge.bnd_req)
if result.verified:
    await dcv.revoke(challenge.domain, token=challenge.token)  # claimant cleanup

dcv.issue()

Generate a stateless DCV challenge. Nothing is written to DNS — the returned DCVChallenge is delivered to the claimant out-of-band.

def issue(
    domain: str,
    *,
    agent_name: str | None = None,
    issuer_domain: str | None = None,
    ttl_seconds: int = 3600,
) -> DCVChallenge

Parameters

Parameter	Type	Required	Default	Description
`domain`	`str`	Yes	-	Domain the claimant must prove control of
`agent_name`	`str`	No	`None`	Agent name to scope the `bnd-req` field
`issuer_domain`	`str`	No	`None`	Issuer domain to scope the `bnd-req` field
`ttl_seconds`	`int`	No	`3600`	Challenge validity window (30–86400)

Returns a DCVChallenge. Raises ValueError if ttl_seconds is out of range; ValidationError for invalid domain or agent name.

dcv.place()

Write the DCV challenge TXT record to DNS via the configured backend.

async def place(
    domain: str,
    token: str,
    *,
    bnd_req: str | None = None,
    expiry_seconds: int = 3600,
    ttl: int = 300,
    backend: DNSBackend | None = None,
) -> DCVPlaceResult

Parameters

Parameter	Type	Required	Default	Description
`domain`	`str`	Yes	-	Zone to write the challenge into
`token`	`str`	Yes	-	Token from the challenger (32-char lowercase base32)
`bnd_req`	`str`	No	`None`	Binding scope from the issued challenge
`expiry_seconds`	`int`	No	`3600`	Placed-record validity (30–86400). Prefer aligning with `DCVChallenge.expiry`.
`ttl`	`int`	No	`300`	DNS record TTL — keep short for quick cleanup
`backend`	`DNSBackend`	No	`None`	Defaults to `DNS_AID_BACKEND` env var

Returns a DCVPlaceResult.

dcv.verify()

Resolve _agents-challenge.{domain} and confirm the token is present, unexpired, and (optionally) bound to the expected scope.

async def verify(
    domain: str,
    token: str,
    *,
    nameserver: str | None = None,
    port: int = 53,
    expected_bnd_req: str | None = None,
    require_dnssec: bool = False,
) -> DCVVerifyResult

Parameters

Parameter	Type	Required	Default	Description
`domain`	`str`	Yes	-	Domain to check
`token`	`str`	Yes	-	Token originally issued by the challenger
`nameserver`	`str`	No	`None`	Operator-trusted nameserver IP (testbeds only)
`port`	`int`	No	`53`	DNS port
`expected_bnd_req`	`str`	No	`None`	When set, record `bnd-req` must match exactly
`require_dnssec`	`bool`	No	`False`	When `True`, resolver must set AD flag (silently downgraded when `nameserver=` is used)

Fail-closed contract

Condition	Result
Missing `expiry=` field	`verified=False`
Malformed `expiry=`	`verified=False`
Bare token (no `token=` prefix)	Not matched
`domain=` mismatched with queried domain	Record skipped
Invalid nameserver IP	`DCVVerifyResult(verified=False)` — never raises
`require_dnssec=True` + no AD flag	`verified=False`
`>10` challenge records (DoS guard)	`verified=False`

Returns a DCVVerifyResult.

Security: The nameserver parameter accepts any syntactically valid IP including loopback, link-local (169.254/16), and RFC1918 ranges. It is operator-trusted and intentionally omitted from the MCP tool surface. Do not expose it to untrusted callers.

dcv.revoke()

Delete the DCV challenge TXT record. Should be called immediately after a successful verify() to prevent token reuse within the validity window.

async def revoke(
    domain: str,
    *,
    token: str,
    backend: DNSBackend | None = None,
) -> DCVRevokeResult

Parameters

Parameter	Type	Required	Default	Description
`domain`	`str`	Yes	-	Zone to remove the challenge from
`token`	`str`	Yes	-	Token that was placed (must match the record in DNS)
`backend`	`DNSBackend`	No	`None`	Defaults to `DNS_AID_BACKEND` env var

Returns a DCVRevokeResult. The token must be present in DNS before deletion (check-then-delete reduces racing with concurrent challengers; not atomic — expiry= remains the true security gate).

DCVChallenge

Issued challenge, delivered to the claimant out-of-band.

Field	Type	Description
`token`	`str`	Base32-encoded nonce to place in DNS
`domain`	`str`	Domain being challenged
`fqdn`	`str`	Full owner name (`_agents-challenge.{domain}`)
`txt_value`	`str`	Verbatim TXT RDATA to place
`expiry`	`datetime`	UTC expiry time
`bnd_req`	`str \| None`	Binding scope (`svc:<agent>@<issuer>`), if set

DCVPlaceResult

Field	Type	Description
`fqdn`	`str`	Full owner name where the challenge was placed
`domain`	`str`	Zone domain
`expires_at`	`datetime`	UTC time the placed challenge expires

DCVVerifyResult

Field	Type	Description
`verified`	`bool`	`True` if a valid, unexpired matching record was found
`domain`	`str`	Domain that was queried
`token`	`str`	Token that was checked
`fqdn`	`str`	Full owner name queried
`expired`	`bool`	`True` if a matching record was found but past `expiry`
`dnssec_validated`	`bool`	`True` if `require_dnssec=True` and resolver set AD=1
`error`	`str \| None`	Human-readable failure reason when `verified=False`

DCVRevokeResult

Field	Type	Description
`removed`	`bool`	`True` if the challenge record was deleted
`domain`	`str`	Zone domain
`fqdn`	`str`	Full owner name that was targeted

Data Models

AgentRecord

Represents an AI agent published via DNS-AID.

from dns_aid import AgentRecord, Protocol

agent = AgentRecord(
    name="network-specialist",
    domain="example.com",
    protocol=Protocol.MCP,
    target_host="mcp.example.com",
    port=443,
    capabilities=["ipam", "dns", "vpn"],
    version="1.0.0",
    description="Network automation agent",
    ttl=3600,
)

Fields

Field	Type	Required	Default	Description
`name`	`str`	Yes	-	Agent identifier (1-63 chars, DNS label format)
`domain`	`str`	Yes	-	Domain where agent is published
`protocol`	`Protocol`	Yes	-	Communication protocol
`target_host`	`str`	Yes	-	Hostname where agent is reachable
`port`	`int`	No	443	Port number
`ipv4_hint`	`str`	No	`None`	IPv4 address hint
`ipv6_hint`	`str`	No	`None`	IPv6 address hint
`capabilities`	`list[str]`	No	`[]`	Agent capabilities
`version`	`str`	No	"1.0.0"	Agent version
`description`	`str`	No	`None`	Description
`ttl`	`int`	No	3600	DNS TTL (60-86400)
`cap_uri`	`str`	No	`None`	URI to capability document (DNS-AID)
`cap_sha256`	`str`	No	`None`	SHA-256 digest of capability descriptor
`bap`	`list[str]`	No	`[]`	Supported protocols with versions
`policy_uri`	`str`	No	`None`	URI to agent policy document
`realm`	`str`	No	`None`	Multi-tenant scope identifier
`capability_source`	`str`	No	`None`	Where capabilities came from: `cap_uri`, `well_known`, `agent_card`, `http_index`, `ard_catalog`, `txt_fallback`, `none`
`endpoint_source`	`str`	No	`None`	Where endpoint came from: `dns_svcb`, `dns_svcb_enriched`, `http_index`, `http_index_fallback`, `ard_card` (real endpoint from a fetched ARD agent/server card), `ard_inline`, `direct`, `directory`
`trust_manifest`	`TrustManifest`	No	`None`	Publisher trust claims from an ARD ai-catalog entry (identity, attestations, provenance, signature) — pass-through, not verified
`catalog_trust`	`str \| None`	No	`None`	ARD-sourced records only — how the catalog was trusted: `tls_domain` (on-domain), `dnssec` (DNSSEC-validated off-domain pointer), or `jws` (JWS-signed off-domain). `None` for pure-DNS records.
`dnssec_validated`	`bool`	No	`False`	`True` when this agent's DNS response carried the resolver AD flag. Set for DNS-plane agents when `require_dnssec` / `min_dnssec` / `verify_dane` is used; ARD / HTTP-catalog agents are exempt and stay `False`. AD-flag based — not independent DNSKEY→DS→RRSIG chain validation.
`dane_verified`	`bool \| None`	No	`None`	DANE/TLSA endpoint-certificate binding (opt-in `verify_dane=True`): `True` = endpoint cert matched its DNSSEC-anchored TLSA record; `False` = TLSA mismatch; `None` = not checked, no TLSA record, or not DNSSEC-anchored.

The ARD ai-catalog guide covers the full discovery flow (DNS pointer → catalog → per-agent DNS-first → card dereferencing), publishing the host-anywhere _catalog._agents / _index._agents pointer (dns-aid index publish-catalog, publish_catalog_pointer library/MCP), and the trust model.

Properties

Property	Type	Description
`fqdn`	`str`	Full DNS-AID record name (draft-02 flat primary owner): `{name}.{domain}`
`walkable_fqdn`	`str`	Optional walkable AliasMode form: `{name}._agents.{domain}`
`legacy_fqdn`	`str`	Legacy -01 form: `_{name}._{protocol}._agents.{domain}` (used only by the back-compat discovery path)
`endpoint_url`	`str`	Full URL: `https://{target_host}:{port}`
`svcb_target`	`str`	SVCB target with trailing dot

Methods

Method	Returns	Description
`to_svcb_params()`	`dict[str, str]`	SVCB record parameters
`to_txt_values()`	`list[str]`	TXT record values

DiscoveryResult

Result of a DNS-AID discovery query.

Fields

Field	Type	Description
`query`	`str`	DNS query made
`domain`	`str`	Domain that was queried
`agents`	`list[AgentRecord]`	Discovered agents
`dnssec_validated`	`bool`	Whether DNSSEC was verified
`cached`	`bool`	Whether result was cached
`query_time_ms`	`float`	Query latency in milliseconds

Properties

Property	Type	Description
`count`	`int`	Number of agents discovered

PublishResult

Result of publishing an agent to DNS.

Fields

Field	Type	Description
`agent`	`AgentRecord`	The published agent
`records_created`	`list[str]`	DNS records created
`zone`	`str`	DNS zone used
`backend`	`str`	DNS backend used
`success`	`bool`	Whether publish succeeded
`message`	`str \| None`	Status message

VerifyResult

Result of verifying an agent's DNS records.

Fields

Field	Type	Description
`fqdn`	`str`	FQDN that was verified
`record_exists`	`bool`	DNS record exists
`svcb_valid`	`bool`	SVCB record is valid
`dnssec_valid`	`bool`	DNSSEC chain validated
`dane_valid`	`bool \| None`	DANE/TLSA verified
`endpoint_reachable`	`bool`	Endpoint responds
`endpoint_latency_ms`	`float \| None`	Response time

Properties

Property	Type	Description
`security_score`	`int`	Security score (0-100)
`security_rating`	`str`	"Excellent", "Good", "Fair", or "Poor"

Security Scoring

Check	Points
Record exists	20
SVCB valid	20
DNSSEC valid	30
DANE valid	15
Endpoint reachable	15
Total	100

Protocol

Enumeration of supported agent communication protocols.

from dns_aid import Protocol

Protocol.MCP   # Model Context Protocol (Anthropic)
Protocol.A2A   # Agent-to-Agent (Google)
Protocol.HTTPS # Standard HTTPS

Backends

DNSBackend Interface

Abstract base class for DNS providers.

from dns_aid.backends.base import DNSBackend

class CustomBackend(DNSBackend):
    @property
    def name(self) -> str:
        return "custom"

    async def create_svcb_record(self, zone, name, priority, target, params, ttl) -> str:
        ...

    async def create_txt_record(self, zone, name, values, ttl) -> str:
        ...

    async def delete_record(self, zone, name, record_type) -> bool:
        ...

    async def list_records(self, zone, name_pattern, record_type):
        ...

    async def zone_exists(self, zone) -> bool:
        ...

Route53Backend

AWS Route 53 implementation.

from dns_aid.backends.route53 import Route53Backend

# Auto-discover zones from AWS
backend = Route53Backend()

# Or specify zone ID directly
backend = Route53Backend(zone_id="Z1234567890ABC")

# Use with publish
from dns_aid import publish
from dns_aid.core.publisher import set_default_backend

set_default_backend(backend)
result = await publish(name="my-agent", ...)

Requirements: pip install dns-aid[route53] and AWS credentials configured.

InfobloxBloxOneBackend

Infoblox UDDI (Universal DDI) implementation.

from dns_aid.backends.infoblox import InfobloxBloxOneBackend

# From environment variables (recommended)
backend = InfobloxBloxOneBackend()

# Or with explicit configuration
backend = InfobloxBloxOneBackend(
    api_key="your-api-key",
    dns_view="default",  # DNS view name
    base_url="https://csp.infoblox.com",  # Optional
)

# Use as context manager
async with InfobloxBloxOneBackend() as backend:
    zones = await backend.list_zones()
    print(zones)

Environment Variables:

Variable	Required	Default	Description
`INFOBLOX_API_KEY`	Yes	-	Infoblox UDDI API key
`INFOBLOX_DNS_VIEW`	No	`default`	DNS view name
`INFOBLOX_BASE_URL`	No	`https://csp.infoblox.com`	API URL

DNS-AID Compliance: Infoblox UDDI is not fully compliant with the DNS-AID draft. It only supports alias mode SVCB (priority 0) and lacks alpn, port, and mandatory parameters. For full compliance, use Route53Backend, InfobloxNIOSBackend, NS1Backend, or DDNSBackend.

InfobloxNIOSBackend

Infoblox NIOS on-premise WAPI implementation. Supports full ServiceMode SVCB with custom DNS-AID parameters.

from dns_aid.backends.infoblox import InfobloxNIOSBackend

# From environment variables (recommended)
backend = InfobloxNIOSBackend()

# Or with explicit configuration
backend = InfobloxNIOSBackend(
    host="nios.example.com",
    username="admin",
    password="your-password",
    dns_view="default",       # DNS view name
    wapi_version="2.13.7",    # WAPI version
    verify_ssl=False,         # TLS certificate verification
)

Environment Variables:

Variable	Required	Default	Description
`NIOS_HOST`	Yes	-	Grid Manager hostname or IP
`NIOS_USERNAME`	Yes	-	WAPI username
`NIOS_PASSWORD`	Yes	-	WAPI password
`NIOS_DNS_VIEW`	No	`default`	DNS view name
`NIOS_WAPI_VERSION`	No	`2.13.7`	WAPI version
`NIOS_VERIFY_SSL`	No	`false`	Verify TLS certificate

DNS-AID Compliance: NIOS WAPI supports ServiceMode SVCB records (priority > 0) with full SVC parameters including custom DNS-AID keys (key65400–key65409). NIOS natively supports private-use SVCB keys via the supports_private_svcb_keys property.

NS1Backend

NS1 (IBM NS1 Connect) REST API v1 implementation.

from dns_aid.backends.ns1 import NS1Backend

backend = NS1Backend()  # reads NS1_API_KEY from env

# Or with explicit configuration
backend = NS1Backend(
    api_key="your-api-key",
    base_url="https://api.nsone.net/v1",  # default
)

Variable	Required	Default	Description
`NS1_API_KEY`	Yes	-	NS1 API key with DNS read/write permissions
`NS1_BASE_URL`	No	`https://api.nsone.net/v1`	API base URL (for private/dedicated deployments)

DNS-AID Compliance: NS1 supports ServiceMode SVCB records with full SVC parameters including private-use keys (key65400–key65409). NS1 natively accepts private-use SVCB keys — cap_uri, policy_uri, and realm go directly into the SVCB record without TXT demotion.

AkamaiEdgeDNSBackend

Akamai Edge DNS implementation using the Config DNS API v2 with EdgeGrid authentication.

from dns_aid.backends.akamai_edgedns import AkamaiEdgeDNSBackend

backend = AkamaiEdgeDNSBackend()  # reads AKAMAI_* env vars or ~/.edgerc

# Or with explicit configuration
backend = AkamaiEdgeDNSBackend(
    host="akab-xxxx.luna.akamaiapis.net",
    client_token="...",
    client_secret="...",
    access_token="...",
)

Variable	Required	Default	Description
`AKAMAI_HOST`	No*	-	EdgeGrid API hostname
`AKAMAI_CLIENT_TOKEN`	No*	-	EdgeGrid client token
`AKAMAI_CLIENT_SECRET`	No*	-	EdgeGrid client secret
`AKAMAI_ACCESS_TOKEN`	No*	-	EdgeGrid access token
`AKAMAI_EDGERC`	No	`~/.edgerc`	Path to `.edgerc` credentials file
`AKAMAI_EDGERC_SECTION`	No	`default`	Section within `.edgerc`

* Either all four AKAMAI_* credential variables or an ~/.edgerc file. Partial env credentials raise an error naming the missing variables rather than silently falling back.

DNS-AID Compliance: Akamai Edge DNS supports ServiceMode SVCB records with full SVC parameters including private-use keys (key65400–key65408) via the supports_private_svcb_keys property — cap_uri, policy_uri, bap, and realm go directly into the SVCB record without TXT demotion.

Concurrency: Akamai serializes modifications per zone. The backend serializes its own writes per zone and automatically retries transient 409 concurrentZoneModification responses with exponential backoff, so concurrent publish_agent() calls are safe.

CloudflareBackend

Cloudflare DNS REST API v4 implementation.

from dns_aid.backends.cloudflare import CloudflareBackend

backend = CloudflareBackend()  # reads CLOUDFLARE_API_TOKEN from env

# Or with explicit configuration
backend = CloudflareBackend(
    api_token="your-api-token",
    zone_id="optional-zone-id",  # auto-discovered from domain if omitted
)

Variable	Required	Default	Description
`CLOUDFLARE_API_TOKEN`	Yes	-	API token with Zone → DNS → Edit (and Read) permissions
`CLOUDFLARE_ZONE_ID`	No	-	Zone ID (auto-discovered by zone name if omitted)

DNS-AID Compliance: Cloudflare supports ServiceMode SVCB records (priority > 0) with full SVC parameters including private-use keys (key65400–key65409). Cloudflare natively accepts private-use SVCB keys via the supports_private_svcb_keys property — cap_uri, cap_sha256, bap, policy_uri, and realm go directly into the SVCB record without TXT demotion. TXT records are still written for human-readable metadata (capabilities, version, description), each value stored as its own RFC 1035 character-string.

DDNSBackend

RFC 2136 Dynamic DNS implementation. Works with BIND, Windows DNS, PowerDNS, Knot DNS, and any RFC 2136 compliant server.

from dns_aid.backends.ddns import DDNSBackend

# From environment variables (recommended)
backend = DDNSBackend()

# Or with explicit configuration
backend = DDNSBackend(
    server="ns1.example.com",
    key_name="dns-aid-key",
    key_secret="YourBase64SecretHere==",
    key_algorithm="hmac-sha256",  # default
    port=53,                       # default
    timeout=10.0,                  # default
)

# Or load from BIND key file
backend = DDNSBackend(key_file="/etc/bind/dns-aid-key.conf")

# Use as context manager
async with DDNSBackend() as backend:
    exists = await backend.zone_exists("example.com")

Environment Variables:

Variable	Required	Default	Description
`DDNS_SERVER`	Yes	-	DNS server hostname or IP
`DDNS_KEY_NAME`	Yes	-	TSIG key name
`DDNS_KEY_SECRET`	Yes	-	TSIG key secret (base64)
`DDNS_KEY_ALGORITHM`	No	`hmac-sha256`	TSIG algorithm
`DDNS_PORT`	No	`53`	DNS server port
`DDNS_TIMEOUT`	No	`10`	Query timeout in seconds

Supported TSIG Algorithms:

hmac-sha256 (recommended)
hmac-sha384
hmac-sha512
hmac-sha224
hmac-md5 (legacy)

Full DNS-AID Compliance: DDNSBackend supports ServiceMode SVCB records (priority > 0) with all required parameters (alpn, port, mandatory).

MockBackend

In-memory backend for testing.

from dns_aid.backends.mock import MockBackend

backend = MockBackend()

# Pre-populate zones
backend = MockBackend(zones={"example.com": {}})

JWS Signatures

Application-layer signature verification as an alternative to DNSSEC.

generate_keypair()

Generate an EC P-256 keypair for signing DNS records.

from dns_aid.core.jwks import generate_keypair

private_key, public_key = generate_keypair()
# private_key: EllipticCurvePrivateKey
# public_key: EllipticCurvePublicKey

export_jwks()

Export public key as JWKS JSON for hosting at .well-known/dns-aid-jwks.json.

from dns_aid.core.jwks import export_jwks

jwks_dict = export_jwks(public_key, kid="dns-aid-2024")
# {
#   "keys": [{
#     "kty": "EC",
#     "crv": "P-256",
#     "kid": "dns-aid-2024",
#     "use": "sig",
#     "x": "...",
#     "y": "..."
#   }]
# }

sign_record()

Sign a DNS record payload with a private key.

from dns_aid.core.jwks import sign_record, RecordPayload

payload = RecordPayload(
    fqdn="payment.example.com",
    target="payment.example.com",
    port=443,
    alpn="mcp",
)

jws_compact = sign_record(payload, private_key)
# Returns: "eyJhbGciOiJFUzI1NiIs..."

verify_signature()

Verify a JWS signature against a public key.

from dns_aid.core.jwks import verify_signature

is_valid, payload = verify_signature(jws_compact, public_key)
# is_valid: bool
# payload: RecordPayload if valid, None if invalid

Publishing with Signatures

from dns_aid import publish

result = await publish(
    name="payment",
    domain="example.com",
    protocol="mcp",
    endpoint="payment.example.com",
    sign=True,
    private_key_path="./keys/private.pem",
)

Discovery with Verification

from dns_aid import discover

agents = await discover(
    "example.com",
    verify_signatures=True,  # Verify JWS sig= parameter
)

Validation Utilities

Input validation functions for security compliance.

from dns_aid.utils.validation import (
    validate_agent_name,
    validate_domain,
    validate_protocol,
    validate_endpoint,
    validate_port,
    validate_ttl,
    validate_capabilities,
    validate_fqdn,
    ValidationError,
)

Functions

Function	Input	Returns	Description
`validate_agent_name(name)`	`str`	`str`	Validate/normalize agent name
`validate_domain(domain)`	`str`	`str`	Validate/normalize domain
`validate_protocol(protocol)`	`str`	`Literal["mcp", "a2a"]`	Validate protocol
`validate_endpoint(endpoint)`	`str`	`str`	Validate endpoint hostname
`validate_port(port)`	`int`	`int`	Validate port (1-65535)
`validate_ttl(ttl)`	`int`	`int`	Validate TTL (60-604800)
`validate_capabilities(caps)`	`list[str]`	`list[str]`	Validate capability list
`validate_fqdn(fqdn)`	`str`	`str`	Validate DNS-AID FQDN

ValidationError

Custom exception with structured error details.

try:
    validate_agent_name("INVALID NAME!")
except ValidationError as e:
    print(f"Field: {e.field}")
    print(f"Message: {e.message}")
    print(f"Value: {e.value}")

CLI Reference

The dns-aid CLI provides command-line access to all DNS-AID functions.

Commands

# Publish an agent (auto-updates index)
dns-aid publish --name my-agent --domain example.com --protocol mcp \
    --endpoint agent.example.com --capability chat --capability code

# Publish without updating index
dns-aid publish --name internal-bot --domain example.com --protocol mcp \
    --endpoint bot.example.com --no-update-index

# Discover agents (pure DNS - default)
dns-aid discover example.com
dns-aid discover example.com --protocol mcp

# Discover via HTTP index (ANS-compatible)
dns-aid discover example.com --use-http-index

# Verify an agent
dns-aid verify my-agent.example.com

# List all agents in a zone
dns-aid list example.com

# Delete an agent (auto-removes from index)
dns-aid delete --name my-agent --domain example.com --protocol mcp --force

# Delete without updating index
dns-aid delete --name my-agent --domain example.com --protocol mcp --force --no-update-index

# List available DNS zones
dns-aid zones

# Agent Index Commands
dns-aid index list example.com           # List agents in domain's index
dns-aid index sync example.com           # Sync index with actual DNS records

Agent Communication Commands

# Send a message to an A2A agent (discover-first: DNS → agent card → invoke)
dns-aid message --domain ai.infoblox.com --name security-analyzer \
    "Analyze security of marketing.ai.infoblox.com"

# Send a message to an A2A agent (direct endpoint)
dns-aid message --endpoint https://security-analyzer.ai.infoblox.com \
    "Analyze DNS-AID security posture"

# JSON output
dns-aid message --endpoint https://chat.example.com "Hello" --json

# Custom timeout (seconds)
dns-aid message --domain example.com --name chat "Hello" --timeout 60

Option	Description
`--domain`	Domain for DNS discovery (used with `--name`)
`--name`	Agent name for DNS discovery (used with `--domain`)
`--endpoint`	Direct endpoint URL (skips discovery)
`--json`	Output raw JSON response
`--timeout`	Request timeout in seconds (default: 30)

# List tools on a remote MCP agent (discover-first)
dns-aid list-tools --domain example.com --name network-specialist

# List tools via direct endpoint
dns-aid list-tools --endpoint https://mcp.example.com/mcp

# Call a tool on a remote MCP agent
dns-aid call --endpoint https://mcp.example.com/mcp search_flights \
    --arguments '{"origin": "SFO", "destination": "JFK"}'

# Call with discover-first
dns-aid call --domain example.com --name network-specialist get_subnets \
    --arguments '{"network": "10.0.0.0/8"}'

Environment Variables

General:

Variable	Description
`DNS_AID_BACKEND`	Default backend: "route53", "cloudflare", "ns1", "infoblox", "nios", "akamai-edgedns", "ddns", or "mock"
`DNS_AID_LOG_LEVEL`	Logging level: DEBUG, INFO, WARNING, ERROR

AWS Route 53:

Route 53 uses boto3's credential chain. No env vars are required if ~/.aws/credentials or an IAM role is configured.

Variable	Required	Description
`AWS_ACCESS_KEY_ID`	No	AWS access key (or use `aws configure` / IAM role)
`AWS_SECRET_ACCESS_KEY`	No	AWS secret key
`AWS_DEFAULT_REGION`	No	AWS region (default: us-east-1)
`AWS_PROFILE`	No	Named profile from `~/.aws/credentials`

Infoblox UDDI:

Variable	Description
`INFOBLOX_API_KEY`	Infoblox UDDI API key (required)
`INFOBLOX_DNS_VIEW`	DNS view name (default: "default")
`INFOBLOX_BASE_URL`	API URL (default: https://csp.infoblox.com)

Infoblox NIOS (On-Prem):

Variable	Description
`NIOS_HOST`	Grid Manager hostname or IP (required)
`NIOS_USERNAME`	WAPI username (required)
`NIOS_PASSWORD`	WAPI password (required)
`NIOS_DNS_VIEW`	DNS view name (default: "default")
`NIOS_WAPI_VERSION`	WAPI version (default: "2.13.7")
`NIOS_VERIFY_SSL`	Verify TLS certificate (default: "false")

DDNS (RFC 2136):

Variable	Description
`DDNS_SERVER`	DNS server hostname or IP (required)
`DDNS_KEY_NAME`	TSIG key name (required)
`DDNS_KEY_SECRET`	TSIG key secret, base64 (required)
`DDNS_KEY_ALGORITHM`	TSIG algorithm (default: hmac-sha256)
`DDNS_PORT`	DNS server port (default: 53)

Akamai Edge DNS:

Akamai supports two credential sources. All four AKAMAI_* vars take precedence over ~/.edgerc when set.

Variable	Description
`AKAMAI_HOST`	EdgeGrid API hostname (e.g. `akab-xxxx.luna.akamaiapis.net`)
`AKAMAI_CLIENT_TOKEN`	EdgeGrid client token
`AKAMAI_CLIENT_SECRET`	EdgeGrid client secret
`AKAMAI_ACCESS_TOKEN`	EdgeGrid access token
`AKAMAI_EDGERC`	Path to `.edgerc` credentials file (default: `~/.edgerc`)
`AKAMAI_EDGERC_SECTION`	Section within `.edgerc` (default: `default`)

MCP Server

The MCP server (dns-aid-mcp) exposes DNS-AID as tools for AI assistants.

Starting the Server

# Stdio transport (for Claude Desktop)
dns-aid-mcp

# HTTP transport (for remote access)
dns-aid-mcp --transport http --port 8000

# HTTP with custom host binding
dns-aid-mcp --transport http --host 0.0.0.0 --port 8000

Available Tools

Tool	Description
`publish_agent_to_dns`	Publish an agent to DNS (auto-updates index)
`discover_agents_via_dns`	Discover agents at a domain (supports `use_http_index` param)
`verify_agent_dns`	Verify agent DNS records
`list_published_agents`	List all agents in a zone
`delete_agent_from_dns`	Delete an agent from DNS (auto-updates index)
`list_agent_index`	List agents in domain's index
`sync_agent_index`	Sync index with actual DNS records
`send_a2a_message`	Send a message to an A2A agent. Accepts `domain` + `name` (discover-first) or `endpoint` (direct).

Health Endpoints (HTTP Transport)

Endpoint	Method	Description
`/`	GET	Server info and available tools
`/health`	GET	Basic health check
`/ready`	GET	Readiness check (DNS backend available)

Claude Desktop Integration

Add to ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "dns-aid": {
      "command": "dns-aid-mcp"
    }
  }
}

Error Handling

All functions may raise exceptions. Recommended pattern:

from dns_aid import publish, discover
from dns_aid.utils.validation import ValidationError

try:
    result = await publish(
        name="my-agent",
        domain="example.com",
        protocol="mcp",
        endpoint="agent.example.com",
    )
    if not result.success:
        print(f"Publish failed: {result.message}")
except ValidationError as e:
    print(f"Invalid input: {e.field} - {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

Invocation Module (`core/invoke.py`)

Single source of truth for agent invocation. Both CLI and MCP server delegate to these functions.

send_a2a_message()

Send a message to an A2A agent using discover-first or direct endpoint.

from dns_aid.core.invoke import send_a2a_message, InvokeResult

# Discover-first (DNS → agent card → invoke)
result: InvokeResult = await send_a2a_message(
    message="Analyze DNS-AID security posture",
    domain="ai.infoblox.com",
    name="security-analyzer",
    timeout=30.0,
)

# Direct endpoint (skip discovery)
result = await send_a2a_message(
    message="Hello",
    endpoint="https://chat.example.com",
)

print(result.text)       # Extracted text response
print(result.raw)        # Full JSON-RPC response dict
print(result.error)      # Error message if failed (None on success)

Parameter	Type	Required	Description
`message`	`str`	Yes	Message text to send
`domain`	`str`	No	Domain for DNS discovery (used with `name`)
`name`	`str`	No	Agent name for DNS discovery (used with `domain`)
`endpoint`	`str`	No	Direct endpoint URL (skips discovery). Either `domain`+`name` or `endpoint` required.
`timeout`	`float`	No	Request timeout in seconds (default: 30)

call_mcp_tool()

Call a tool on a remote MCP agent via JSON-RPC tools/call.

from dns_aid.core.invoke import call_mcp_tool

result = await call_mcp_tool(
    endpoint="https://mcp.example.com/mcp",
    tool_name="search_flights",
    arguments={"origin": "SFO", "destination": "JFK"},
)

list_mcp_tools()

List available tools on a remote MCP agent via JSON-RPC tools/list.

from dns_aid.core.invoke import list_mcp_tools

result = await list_mcp_tools(
    endpoint="https://mcp.example.com/mcp",
)

resolve_a2a_endpoint()

Resolve an A2A agent endpoint via DNS discovery and agent card fetch.

from dns_aid.core.invoke import resolve_a2a_endpoint

endpoint_url = await resolve_a2a_endpoint(
    domain="ai.infoblox.com",
    name="security-analyzer",
)
# Returns: "https://security-analyzer.ai.infoblox.com:443"

Resolution chain:

DNS discovery (discover(domain, protocol="a2a", name=name))
Agent card fetch (/.well-known/agent-card.json) for canonical URL
Host mismatch protection: if agent card URL hostname differs from DNS endpoint, DNS wins

InvokeResult

Returned by all invocation functions.

Field	Type	Description
`text`	`str \| None`	Extracted text from response
`raw`	`dict \| None`	Full response payload
`error`	`str \| None`	Error message if invocation failed

SDK: Invocation & Telemetry

The Tier 1 SDK provides agent invocation with automatic telemetry capture, and community-wide ranking queries.

Top-Level Functions

invoke()

async def invoke(
    agent: AgentRecord,
    *,
    method: str | None = None,
    arguments: dict | None = None,
    timeout: float | None = None,
    config: SDKConfig | None = None,
) -> InvocationResult

One-shot agent invocation with telemetry. Creates an AgentClient, calls the agent, returns the result with an attached signal.

Examples:

import dns_aid

# MCP agent: list tools
result = await dns_aid.discover("example.com", protocol="mcp")
resp = await dns_aid.invoke(result.agents[0], method="tools/list")
print(resp.signal.invocation_latency_ms)  # 148.2

# A2A agent: send a message (standard JSON-RPC message/send)
result = await dns_aid.discover("ai.infoblox.com", protocol="a2a")
resp = await dns_aid.invoke(
    result.agents[0],
    method="message/send",
    arguments={
        "message": {
            "messageId": "unique-id",
            "role": "user",
            "parts": [{"kind": "text", "text": "What is DNS-AID?"}],
        }
    },
)
# Standard A2A methods (message/send, tasks/get, etc.) are automatically
# wrapped in a JSON-RPC 2.0 envelope by the A2A protocol handler.

rank()

async def rank(
    agents: list[AgentRecord],
    *,
    method: str | None = None,
    arguments: dict | None = None,
    config: SDKConfig | None = None,
) -> list[RankedAgent]

Invoke multiple agents and rank by telemetry performance (composite score).

Protocol Handlers

The SDK routes invocations through protocol-specific handlers:

Protocol	Handler	Wire Format
MCP	`MCPProtocolHandler`	MCP Streamable HTTP (modern, spec 2025-03-26+) for `tools/call` and `tools/list`, with transparent legacy plain JSON-RPC POST fallback when the target rejects the modern transport
A2A	`A2AProtocolHandler`	JSON-RPC 2.0 for standard methods (`message/send`, `tasks/get`); generic payload for custom methods
HTTPS	`HTTPSProtocolHandler`	HTTP POST with JSON body

MCP Protocol Handler delegates transport to the official mcp Python SDK (mcp.client.streamable_http.streamablehttp_client and mcp.ClientSession) for the modern path. The handler injects a per-invocation telemetry adapter (latency, TTFB, response size, cost headers, TLS version) and propagates the dns-aid X-DNS-AID-Caller-Domain header on every request when DNS_AID_CALLER_DOMAIN is set. On transport mismatch (HTTP 405/406, refused initialize via JSON-RPC -32601) the handler transparently falls back to the legacy plain JSON-RPC POST path; the fallback event is logged as a structured warning (transport.legacy_fallback) carrying the endpoint, the failure reason, and the modern attempt latency.

A2A Protocol Handler automatically wraps standard A2A methods in a JSON-RPC 2.0 envelope:

# Standard methods (message/send, message/stream, tasks/get, tasks/cancel, etc.)
# are wrapped in: {"jsonrpc": "2.0", "method": "...", "params": {...}, "id": "..."}

# Non-standard/custom methods use generic format for backward compatibility:
# {"method": "custom_task", ...arguments}

AgentClient

The main SDK class. Use as async context manager for connection reuse.

from dns_aid.sdk import AgentClient, SDKConfig

config = SDKConfig(timeout_seconds=30.0, caller_id="my-app")

async with AgentClient(config=config) as client:
    result = await client.invoke(agent, method="tools/list")
    ranked = client.rank()

Methods:

Method	Description
`invoke(agent, method, arguments, timeout, credentials, credential_provider, auth_handler)`	Invoke agent, return `InvocationResult`
`rank(strategy)`	Rank all invoked agents by composite score
`fetch_rankings(fqdns, limit)`	Fetch community-wide rankings from telemetry API
`signals`	Property: list of all collected `InvocationSignal` objects

invoke() — credential resolution (v0.21.0+)

AgentClient.invoke() resolves authentication credentials in this precedence order (the first non-empty source wins; subsequent sources are not consulted):

auth_handler — explicit AuthHandler instance for full caller control
credentials — pre-fetched credentials dict (existing behavior)
credential_provider — async callback awaited lazily at invoke time (v0.21.0+)
No-auth fallback when all three are absent

The credential_provider parameter accepts an async callable that takes the target AgentRecord and returns a credentials dict. It enables short-lived delegation tokens (RFC 8693 token exchange), per-target credential scoping, AWS STS assume-role per invocation, and dynamic secret stores (Vault, KMS).

from dns_aid.core.models import AgentRecord

async def per_target_provider(agent: AgentRecord) -> dict[str, str]:
    # Provider receives the AgentRecord — can derive credentials from
    # agent.fqdn, agent.realm, agent.connect_meta, etc.
    return {"token": await mint_jwt_for(agent.fqdn)}

async with AgentClient(config=config) as client:
    resp = await client.invoke(
        agent,
        method="tools/list",
        credential_provider=per_target_provider,
    )

SDKConfig.credential_provider_timeout (default 30s, env var DNS_AID_CREDENTIAL_PROVIDER_TIMEOUT) bounds the await. Hanging providers surface as CredentialProviderError with the underlying TimeoutError preserved as __cause__. Provider exceptions never leak credential values into logs or the wrapped error's serialised surface.

See security-credentials.md for the per-handler security matrix and the full credential-handling posture.

fetch_rankings()

async def fetch_rankings(
    self,
    fqdns: list[str] | None = None,
    limit: int = 50,
) -> list[dict]

Fetch community-wide rankings from a configured telemetry API endpoint. Returns pre-computed composite scores based on aggregated telemetry.

Parameters:

Parameter	Type	Default	Description
`fqdns`	`list[str] \| None`	`None`	Filter rankings to specific agent FQDNs
`limit`	`int`	`50`	Maximum number of rankings to return

Returns: List of ranking dictionaries with agent_fqdn, composite_score, etc.

Example:

async with AgentClient(config) as client:
    # Get all rankings
    rankings = await client.fetch_rankings()

    # Get rankings for specific agents only
    rankings = await client.fetch_rankings(
        fqdns=["booking.example.com"],
        limit=10
    )

    for r in rankings:
        print(f"{r['agent_fqdn']}: {r['composite_score']}")

Note: Requires telemetry_api_url to be configured in SDKConfig. Returns empty list if not configured.

SDKConfig

from dns_aid.sdk import SDKConfig

config = SDKConfig(
    timeout_seconds=30.0,            # Default request timeout
    caller_id="my-app",              # Caller identifier for signals
    persist_signals=False,           # Auto-save signals to PostgreSQL
    database_url=None,               # DB URL (falls back to DATABASE_URL env)
    otel_enabled=False,              # Enable OpenTelemetry export (v0.23.0+)
    otel_endpoint=None,              # OTLP endpoint URL — use http:// (plaintext) or https:// (TLS)
    otel_export_format="otlp",       # "otlp" | "console" | "noop"
    otel_sampler=None,               # Sampler name; None = OTEL default (v0.23.0+)
    otel_environment=None,           # deployment.environment resource attr (v0.23.0+)
    otel_metric_labels=[],           # Opt-in high-cardinality labels: fqdn|caller|tool (v0.23.0+)
    http_push_url=None,              # POST signals to remote telemetry API
    directory_api_url=None,          # Base URL for AgentClient.search() and fetch_rankings()
    telemetry_api_url=None,          # Deprecated alias for directory_api_url
    credential_provider_timeout=30.0,  # Max seconds to wait for credential_provider callback (v0.21.0+)
)

# Or from environment variables:
config = SDKConfig.from_env()

Environment Variables:

Variable	Default	Description
`DNS_AID_SDK_TIMEOUT`	30.0	Request timeout in seconds
`DNS_AID_SDK_CALLER_ID`	None	Caller identifier
`DNS_AID_SDK_PERSIST_SIGNALS`	false	Enable DB persistence
`DATABASE_URL`	None	PostgreSQL connection URL
`DNS_AID_SDK_OTEL_ENABLED`	false	Enable OpenTelemetry (v0.23.0+)
`DNS_AID_SDK_OTEL_ENDPOINT`	None	OTLP collector URL — `http://` plaintext, `https://` TLS
`DNS_AID_SDK_OTEL_EXPORT_FORMAT`	otlp	`otlp` \| `console` \| `noop`
`DNS_AID_SDK_OTEL_SAMPLER`	None	Sampler name; lower precedence than `OTEL_TRACES_SAMPLER` (v0.23.0+)
`DNS_AID_SDK_OTEL_ENVIRONMENT`	None	Sets `deployment.environment` resource attr (v0.23.0+)
`DNS_AID_SDK_OTEL_METRIC_LABELS`	None	Comma-separated opt-in labels: `fqdn,caller,tool` (v0.23.0+)
`DNS_AID_SDK_HTTP_PUSH_URL`	None	POST signals to this URL
`DNS_AID_SDK_DIRECTORY_API_URL`	None	Base URL for `AgentClient.search()` + `fetch_rankings()` (v0.19.0+)
`DNS_AID_SDK_TELEMETRY_API_URL`	None	Deprecated alias for `DNS_AID_SDK_DIRECTORY_API_URL`

Standard OpenTelemetry env vars are also honored: OTEL_TRACES_SAMPLER, OTEL_TRACES_SAMPLER_ARG, OTEL_PROPAGATORS, OTEL_RESOURCE_ATTRIBUTES, OTEL_EXPORTER_OTLP_HEADERS, OTEL_EXPORTER_OTLP_ENDPOINT. See docs/integrations/opentelemetry.md for the full guide (sampling, propagation, managed-collector auth, failure modes).

The resolved_directory_url property returns directory_api_url when set, falling back to telemetry_api_url for backwards compatibility. Using the legacy alias emits a DeprecationWarning once per process.

AgentClient.search() — Path B cross-domain search (v0.19.0+)

async def search(
    self,
    q: str | None = None,
    *,
    protocol: Literal["mcp", "a2a", "https"] | None = None,
    domain: str | None = None,
    capabilities: list[str] | None = None,
    min_security_score: int | None = None,
    verified_only: bool = False,
    intent: str | None = None,
    auth_type: str | None = None,
    transport: str | None = None,
    realm: str | None = None,
    limit: int = 20,
    offset: int = 0,
) -> SearchResponse

Issues GET {directory_api_url}/api/v1/search and returns a typed SearchResponse. Path B is opt-in: invoking search() without directory_api_url configured raises DirectoryConfigError immediately, before any network work.

Parameters

Parameter	Type	Default	Description
`q`	`str \| None`	`None`	Free-text query, or `None` for browse-all-with-filters mode.
`protocol`	`Literal["mcp","a2a","https"] \| None`	`None`	Restrict to a protocol.
`domain`	`str \| None`	`None`	Restrict to a single domain.
`capabilities`	`list[str] \| None`	`None`	All-of capability match.
`min_security_score`	`int \| None`	`None`	Minimum security score (0–100).
`verified_only`	`bool`	`False`	Restrict to DCV-verified domains.
`intent`	`str \| None`	`None`	Action intent filter (`query` / `command` / `transaction` / `subscription`).
`auth_type`	`str \| None`	`None`	Auth type filter.
`transport`	`str \| None`	`None`	Transport filter (`streamable-http`, `https`, `sse`, `stdio`).
`realm`	`str \| None`	`None`	Multi-tenant realm filter.
`limit`	`int`	`20`	Page size, 1–10000.
`offset`	`int`	`0`	Pagination offset.

Returns

SearchResponse — query echo, ranked results, pagination state.

Raises

DirectoryConfigError — directory_api_url not configured.
DirectoryAuthError — Directory rejected credentials (HTTP 401/403).
DirectoryRateLimitedError — Directory rate-limited (HTTP 429); the retry_after_seconds detail mirrors the Retry-After header.
DirectoryUnavailableError — Transient: connect refused, timeout, 5xx, 404, unexpected redirect, oversized response, or response shape the SDK can't validate.
RuntimeError — Client not in an async context manager.

Example

from dns_aid.sdk import AgentClient, SDKConfig

async with AgentClient(config=SDKConfig.from_env()) as client:
    response = await client.search(
        "payment processing",
        protocol="mcp",
        capabilities=["payment-processing"],
        min_security_score=70,
    )
    for result in response.results:
        print(f"{result.score:.2f}  {result.agent.fqdn}  T{result.trust.trust_tier}")
        print(f"   sec={result.trust.security_score}  trust={result.trust.trust_score}")

    # Paginate
    while response.has_more:
        response = await client.search(q="payment", offset=response.next_offset)

Composition pattern (zero-trust)

Path B returns directory-attested candidates. Path A re-verifies via DNS substrate before invoking — directory is opt-in convenience, never a trust bottleneck.

from dns_aid.sdk import AgentClient
from dns_aid.core.discoverer import discover

async with AgentClient() as client:
    # 1. Cross-domain candidate discovery
    response = await client.search(q="fraud detection", min_security_score=70)

    for candidate in response.results:
        # 2. DNS-substrate re-verification with signature requirement
        verified = await discover(
            candidate.agent.domain,
            name=candidate.agent.name,
            require_signed=True,
            require_signature_algorithm=["ES256", "Ed25519"],
        )
        if verified.agents:
            # 3. Safe to invoke
            ...

SearchResponse / SearchResult / TrustAttestation / Provenance (v0.19.0+)

Typed result models in dns_aid.sdk.search. All immutable (frozen=True) and forward-compatible (extra="ignore") so directory schema additions don't break SDK consumers.

class SearchResponse(BaseModel):
    query: str | None              # Echo of the q parameter from the directory.
    results: list[SearchResult]    # Ranked results, length <= limit.
    total: int                     # Matches across all pages (after skip-and-log).
    limit: int
    offset: int

    @property
    def has_more(self) -> bool: ...
    @property
    def next_offset(self) -> int | None: ...

class SearchResult(BaseModel):
    agent: AgentRecord
    score: float                       # Raw relevance score, not normalized.
    trust: TrustAttestation            # Defaults to all-zero if directory omits.
    provenance: Provenance | None      # Built when first_seen / last_seen present.

class TrustAttestation(BaseModel):
    security_score: int = 0            # 0–100
    trust_score: int = 0               # 0–100
    popularity_score: int = 0          # 0–100
    trust_tier: int = 0                # 0–3 (untiered / basic / enhanced / continuous)
    safety_status: Literal["active", "blocked"] = "active"
    dnssec_valid: bool | None = None
    dane_valid: bool | None = None
    svcb_valid: bool | None = None
    endpoint_reachable: bool | None = None
    protocol_verified: bool | None = None
    threat_flags: dict[str, Any] = {}
    breakdown: dict[str, Any] | None = None   # Directory's trust_breakdown.
    badges: list[str] | None = None           # Directory's trust_badges.

class Provenance(BaseModel):
    discovery_level: int = 0           # 0 observed / 1 beacon / 2 manifest / 3 federated
    first_seen: datetime
    last_seen: datetime
    last_verified: datetime | None = None
    company: dict[str, Any] | None = None

Directory exceptions (v0.19.0+)

from dns_aid.sdk import (
    DirectoryError,                # Base exception class.
    DirectoryConfigError,          # directory_api_url not configured (78 EX_CONFIG).
    DirectoryUnavailableError,     # Transient (75 EX_TEMPFAIL).
    DirectoryRateLimitedError,     # 429; carries retry_after_seconds detail.
    DirectoryAuthError,            # 401/403 (77 EX_NOPERM); does NOT inherit from Unavailable.
)

Every exception carries a details: dict[str, Any] attribute with structured fields (directory_url, status_code, underlying, etc.) so callers can dispatch on type and inspect details for richer error handling.

InvocationResult

Returned by invoke(). Contains response data and telemetry signal.

Field	Type	Description
`success`	bool	Whether invocation succeeded
`data`	dict \| str \| None	Response payload
`signal`	InvocationSignal	Telemetry signal for this call
`error_message`	str \| None	Error description if failed

InvocationSignal

Per-call telemetry captured automatically.

Field	Type	Description
`id`	UUID	Unique signal identifier
`agent_fqdn`	str	Agent DNS-AID FQDN
`agent_endpoint`	str	Endpoint URL used
`protocol`	str	Protocol (mcp, a2a, https)
`method`	str	Method called
`status`	InvocationStatus	success, error, timeout, refused
`invocation_latency_ms`	float	Total invocation time
`ttfb_ms`	float	Time to first byte
`http_status_code`	int	HTTP response status
`cost_units`	float	Cost from X-Cost-Units header
`cost_currency`	str	Currency from X-Cost-Currency header
`response_size_bytes`	int	Response payload size
`tls_version`	str	TLS version used
`timestamp`	datetime	When the call was made
`caller_id`	str	Caller identifier from config

Ranking

ranked = client.rank()  # Default: WeightedCompositeStrategy

for r in ranked:
    print(f"{r.agent_fqdn}: {r.composite_score:.1f}")

Scoring Formula (WeightedComposite):

composite = 0.40 * reliability   (success_rate * 100)
          + 0.30 * latency       (100 * (1 - avg_latency/5000))
          + 0.15 * cost          (relative to cheapest)
          + 0.15 * freshness     (recency weighted)

Available Strategies:

WeightedCompositeStrategy (default)
LatencyFirstStrategy — prioritizes lowest latency
ReliabilityFirstStrategy — prioritizes highest success rate

HTTP Telemetry Push (Optional)

The SDK can optionally push signals to an external telemetry collection endpoint:

config = SDKConfig(
    http_push_url="https://your-telemetry-server.example.com/signals"
)

async with AgentClient(config=config) as client:
    # Signals automatically pushed in a background thread
    await client.invoke(agent, method="tools/list")

Disabled by default (http_push_url=None). Configure via SDKConfig or the DNS_AID_SDK_HTTP_PUSH_URL environment variable.

Version

import dns_aid
print(dns_aid.__version__)  # "0.6.0"