DNS-AID API Reference

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
`cap_metadata`	`dict[str, Any]`	No	`{}`	Operator-defined fields of the fetched capability document (every key beyond the known schema). Opaque, unverified passthrough — judge integrity via `cap_sha256_verified`, same as `capabilities`
`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"

FilesExpand file tree

api-reference.md

Latest commit

History

api-reference.md

File metadata and controls

DNS-AID API Reference

Relationship to IETF

Table of Contents

Quick Start

Core Functions

publish()

Parameters

Returns

Example

DNS Records Created

discover()

Parameters

Discovery Methods

Returns

Raises

Example

verify()

Parameters

Returns

Example

Domain Control Validation (DCV)

dcv.issue()

Parameters

dcv.place()

Parameters

dcv.verify()

Parameters

Fail-closed contract

dcv.revoke()

Parameters

DCVChallenge

DCVPlaceResult

DCVVerifyResult

DCVRevokeResult

Data Models

AgentRecord

Fields

Properties

Methods

DiscoveryResult

Fields

Properties

PublishResult

Fields

VerifyResult

Fields

Properties

Security Scoring

Protocol

Backends

DNSBackend Interface

Route53Backend

InfobloxBloxOneBackend

InfobloxNIOSBackend

NS1Backend

AkamaiEdgeDNSBackend

CloudflareBackend

DDNSBackend

MockBackend

JWS Signatures

generate_keypair()

export_jwks()

sign_record()

verify_signature()

Publishing with Signatures

Discovery with Verification

Validation Utilities

Functions

ValidationError

CLI Reference

Commands

Agent Communication Commands

Environment Variables

MCP Server

Invocation Module (`core/invoke.py`)