DNS-AID 深度实战：当 DNS 协议遇上 AI Agent 发现——从 SRV 记录到生产级智能体注册的完全指南（2026）

一、背景：AI Agent 的「寻人启事」困境

1.1 智能体爆炸时代的尴尬

2026年的今天，AI Agent 已经不再是实验室里的概念验证。从 Claude Code 在终端里替你写代码，到 Hermes Agent 在 Slack 上自动处理工单，再到 OpenAI Codex 帮你维护 CI/CD 流水线——智能体的数量正在以指数级增长。

但这里有一个非常尴尬的问题：Agent 之间怎么找到对方？

假设你是一个运行在 Kubernetes 集群里的 MCP（Model Context Protocol）服务器，你想把自己的能力注册出去，让其他 Agent 可以调用你的「查询订单」工具。或者你是一个 AI 编程助手，需要找到当前开发团队使用的「代码审查 Agent」——你怎么知道它在哪？它绑定了哪个端口？它支持什么协议？

目前市场上的做法五花八门：

• OpenAI 的 Skills 目录：集中式注册，但只服务于 OpenAI 生态
• Anthropic 的 MCP 服务器列表：同样集中式，但社区维护
• AAIF（Agent AI Foundation）的 AgentGateway：网关模式，但需要额外部署基础设施
• 各种自建注册中心：每个团队造一个轮子

这些问题本质上是一个在计算机科学领域已经被解决过无数次的老问题——服务发现。但问题是，现有的服务发现方案（Consul、etcd、Eureka、Kubernetes DNS）都是为传统微服务设计的，它们假设你控制整个集群。而 AI Agent 的世界是异构的、跨组织的、跨越公网和内网的。

1.2 为什么传统服务发现方案不行

最关键的问题在于：这些方案都假设你拥有基础设施。而 AI Agent 是「租户」——你可能在自己的笔记本上跑一个 Agent，这个 Agent 需要找到某大厂的 API Agent，而那个 API Agent 跑在 AWS 上，背后还有 Cloudflare 挡着。

1.3 那为什么不直接用 DNS？

DNS（域名系统）是互联网上最成功、最古老的服务发现协议。它分布式、层次化、全球一致、自带缓存、安全扩展（DNSSEC）——这些特性几乎完美契合 AI Agent 的发现需求。

道理没错，但 DNS 的原始设计是给人看的——www.example.com 解析到 192.0.2.1。Agent 需要的东西比这复杂得多：

• 它需要知道对方支持什么协议（MCP？A2A？自定义？）
• 它需要知道对方的能力描述（能做什么工具调用？）
• 它需要验证对方身份（真的是我找的那个 Agent？）
• 它需要负载均衡（多个同类型 Agent 选哪个？）

这些需求催生了 DNS-AID——Linux 基金会（通过其旗下的 AAIF）于 2026 年 6 月 4 日正式发布的智能体发现标准。

二、核心概念：DNS-AID 到底是什么

2.1 一句话定义

DNS-AID（DNS-based Agent Identification and Discovery） 是一套基于现有 DNS 基础设施的开放标准，允许 AI 智能体通过 DNS 记录实现彼此的发现、验证和通信，无需额外搭建注册中心。

2.2 核心设计原则

Linux 基金会在设计 DNS-AID 时坚持了四条原则：

1. 零新基础设施
这是最有杀伤力的点。DNS 已经跑在每一台联网设备上。不需要部署 Consul 集群，不需要注册到某个中心化平台，不需要申请 API Key。如果你拥有一个域名，你就能让 Agent 被发现。

2. 供应商中立
DNS-AID 由 Infoblox 发起，德国电信和亚马逊贡献代码，Linux 基金会托管——注意这三家公司是竞品。Infoblox 做 DNS 设备，德国电信是运营商，亚马逊 AWS 是云服务商。他们能在同一个标准下合作，说明这个标准是真的「中性」的。

3. 安全优先
DNS 原生缺乏身份验证机制。DNS-AID 在草案中已经明确了 DNSSEC 和 DANE（DNS-based Authentication of Named Entities）的集成方案，确保 Agent 不会连接到冒充者。

4. 渐进式部署
你不需要一口气改完所有 DNS 配置。DNS-AID 设计了 fallback 机制：如果 Agent 找不到 DNS-AID 记录，可以回退到传统服务发现方式。

2.3 核心寻址格式

这是 DNS-AID 最核心的规范——通用智能体地址（Universal Agent Address, UAA）：

_index._agents.{domain}

举个例子，如果你的域名是 example.com，你想让 Agent 能发现你的服务，你需要创建一条 DNS 记录指向：

_index._agents.example.com

当一个 Agent 想找 example.com 域下注册的智能体时，它会先查询这个地址，获取该域下所有可用 Agent 的索引。

这里的 _index 和 _agents 是 DNS 标准的「服务名称」前缀，类似于你已经见过的 _ldap._tcp.example.com（LDAP 服务发现）或 _sip._udp.example.com（SIP 服务发现）。DNS-AID 沿用了这个既有惯例，保证了与 DNS 生态的完全兼容。

三、架构分析：从 DNS 协议到 Agent 注册

3.1 DNS-AID 的记录类型设计

DNS-AID 定义了三类 DNS 记录来实现完整的 Agent 发现功能：

3.1.1 Agent 索引记录（AIR）

这是入口记录，类型为标准的 SRV 记录（RFC 2782）。

_index._agents.example.com.  3600  IN  SRV  0 5 443 agent-alpha.example.com.
_index._agents.example.com.  3600  IN  SRV  0 5 443 agent-beta.example.com.
_index._agents.example.com.  3600  IN  SRV  0 10 8443 agent-gamma.example.com.

SRV 记录的格式为：priority weight port target。

• priority：优先级，数值越小优先级越高（0 最高）
• weight：同优先级下的权重，用于负载均衡
• port：Agent 服务的端口
• target：Agent 的主机名

这种设计非常巧妙——它天然支持了 Agent 的负载均衡和故障转移。在上面的例子中，agent-gamma 的 weight 是 10，其他两个是 5，意味着 agent-gamma 会收到约 50% 的流量。

3.1.2 Agent 元数据记录（AMR）

光有 SRV 记录不够，Agent 还需要知道对方的元数据——支持什么协议、有什么能力、用什么认证方式。这部分通过 TXT 记录实现：

_agent._meta.example.com.  300  IN  TXT  "proto=mcp;version=1.1;auth=mtls"
_agent._meta.example.com.  300  IN  TXT  "capabilities=tools/query-order,tools/get-invoice,resources/catalog"
_agent._meta.example.com.  300  IN  TXT  "description=订单查询与发票管理Agent"
_agent._meta.example.com.  300  IN  TXT  "agent-type=api-gateway;vendor=acme-corp"

TXT 记录使用键值对格式（类似 DNS-SD 的风格），每个字段用分号分隔。目前草案中定义的元数据字段包括：

3.1.3 Agent 验证记录（AVR）

为了防止 DNS 劫持或中间人攻击，DNS-AID 要求使用 DNSSEC 对上述记录进行签名，并且可以通过 TLSA 记录（DANE 协议）公布 Agent 的 TLS 证书指纹：

_443._tcp.agent-alpha.example.com.  86400  IN  TLSA  3 1 1 (
    abcd1234efgh5678ijkl9012mnop3456qrst7890 )

这样，当 Agent B 通过 DNS 查询到 agent-alpha 的地址后，连接时可以直接验证其 TLS 证书是否匹配 TLSA 记录中的指纹，防止中间人攻击。

3.2 DNS-AID 的查询流程

完整的 Agent 发现流程分为三个步骤：

Step 1：发现入口

Agent-C → DNS: 查询 _index._agents.target-domain.com
DNS → Agent-C: 返回 SRV 记录列表
               [agent-alpha.example.com:443, agent-beta.example.com:443]

Step 2：获取元数据

Agent-C → DNS: 查询 _agent._meta.target-domain.com
DNS → Agent-C: 返回 TXT 记录
               [proto=mcp;version=1.1;capabilities=tools/query-order]

Step 3：验证身份（可选但推荐）

Agent-C → DNS: 查询 _443._tcp.agent-alpha.example.com
DNS → Agent-C: 返回 TLSA 记录
               [证书指纹: abcd1234...]
Agent-C → agent-alpha: TLS 连接 + 验证证书

三步完成之后，Agent-C 就知道了：

1. agent-alpha 在哪（IP:443）
2. 它支持什么协议和能力（MCP v1.1，能查订单）
3. 它确实是 agent-alpha（证书已被验证）

3.3 DNS-AID 与现有协议的协同

这里有一个很多人会混淆的点：DNS-AID 不是要替代 MCP 或 A2A 协议，而是要补充它们。

• MCP（Model Context Protocol）：定义了 Agent 和工具/数据源之间的交互方式（怎么调用工具）
• A2A（Agent-to-Agent）：定义了 Agent 之间的协作方式（怎么对话）
• DNS-AID：定义了 Agent 怎么找到对方（怎么发现）

打个比方：MCP 是 HTTP 协议，A2A 是 WebSocket 协议，DNS-AID 就是 DNS——你浏览器需要先 DNS 解析找到服务器 IP，然后才能建立 HTTP 连接。

+-------------------+      +-------------------+
|   Agent Client    |      |    Target Agent   |
+-------------------+      +-------------------+
         |                           |
         |  1. DNS-AID 发现          |
         |  (查询 SRV + TXT + TLSA) |
         |-------------------------->|
         |<--------------------------|
         |                           |
         |  2. MCP/A2A 通信          |
         |  (工具调用/Agent对话)      |
         |==========================>|
         |<==========================|
         |                           |

这个分层设计非常干净。你完全可以只用 DNS-AID 做发现，然后用 MCP 做通信，再用 A2A 做多 Agent 编排——它们不冲突，反而互补。

四、代码实战：从零搭建 DNS-AID 兼容的 Agent 注册系统

理论说够了，上代码。这一节我们会搭建一个完整的 Agent 注册与发现系统，包含：

1. DNS-AID 兼容的 DNS 服务器配置（用 CoreDNS）
2. Agent 注册脚本
3. Agent 发现客户端
4. 自动健康检查

4.1 环境准备

首先确保安装了 CoreDNS 和 Python 3.10+：

# macOS
brew install coredns

# Linux (Debian/Ubuntu)
# wget https://github.com/coredns/coredns/releases/download/v1.12.0/coredns_1.12.0_linux_amd64.tgz
# tar -xzf coredns_1.12.0_linux_amd64.tgz && sudo mv coredns /usr/local/bin/

# 验证
coredns -version
# CoreDNS-1.12.0

4.2 配置 CoreDNS 作为 DNS-AID 服务器

CoreDNS 是 CNCF 托管的 DNS 服务器，用 Go 编写，支持插件化扩展。我们将用它来提供 DNS-AID 记录。

创建一个 Corefile：

# Corefile — DNS-AID Agent Registry
.:5353 {
    # 权限控制：只响应特定域名的查询
    # 生产环境建议加上 ACL 限制
    acl {
        allow net 0.0.0.0/0
        block
    }

    # 静态记录 — 核心 DNS-AID 配置
    file /etc/coredns/agents.db

    # 日志
    log
    errors

    # 健康检查端点
    health :8080

    # Metrics
    prometheus :9153

    # 缓存
    cache 3600
}

agents.db 是 zone 文件，存放 DNS-AID 记录：

; agents.db — DNS-AID Agent Registry Zone File
; 域名：agents.example.com

$TTL 3600
@   IN  SOA ns1.agents.example.com. admin.agents.example.com. (
    2026061001  ; Serial
    7200        ; Refresh
    3600        ; Retry
    1209600     ; Expire
    3600        ; Minimum TTL
)

; NS 记录
@   IN  NS  ns1.agents.example.com.
ns1 IN  A   192.168.1.100

; ============================================
; DNS-AID 核心记录
; ============================================

; Agent Index Record (AIR) — SRV
; 通知系统 Agent (priority=0, weight=5, port=443)
_index._agents.agents.example.com.  3600 IN SRV 0 5 443 notify-agent.agents.example.com.

; 代码审查 Agent (priority=0, weight=10, port=443)
_index._agents.agents.example.com.  3600 IN SRV 0 10 443 code-review-agent.agents.example.com.

; 文档生成 Agent (priority=1, weight=5, port=8080)
_index._agents.agents.example.com.  3600 IN SRV 1 5 8080 docs-agent.agents.example.com.

; Agent Metadata Records (AMR) — TXT
; 通知系统 Agent 元数据
_agent._meta.agents.example.com.  300 IN TXT (
    "proto=mcp;version=1.1;"
    "capabilities=tools/send-notification,tools/list-channels;"
    "auth=mtls;agent-type=api-gateway;"
    "vendor=acme-corp"
)

; 代码审查 Agent 元数据
_agent._meta.agents.example.com.  300 IN TXT (
    "proto=a2a;version=2.0;"
    "capabilities=tools/review-code,tools/check-style,tools/analyze-complexity;"
    "auth=oauth2;agent-type=coding-agent;"
    "vendor=open-source"
)

; 文档生成 Agent 元数据
_agent._meta.agents.example.com.  300 IN TXT (
    "proto=mcp;version=1.0;"
    "capabilities=tools/generate-docs,tools/update-readme;"
    "auth=api-key;agent-type=utility;"
    "vendor=community"
)

; Agent 主机解析
notify-agent.agents.example.com.     3600 IN A 10.0.1.50
code-review-agent.agents.example.com. 3600 IN A 10.0.1.51
docs-agent.agents.example.com.        3600 IN A 10.0.1.52

4.3 Agent 注册客户端（Python）

高级场景中，Agent 不应该手动维护 DNS 记录，而是应该有一个自动化注册机制。下面是一个 DNS-AID 注册客户端：

"""
dns_aid_registrar.py — DNS-AID Agent Registration Client
支持自动注册、心跳维护和健康检查
"""

import json
import logging
import socket
import time
import dns.resolver
import dns.update
import dns.query
import dns.tsigkeyring
from dataclasses import dataclass, field, asdict
from typing import Optional
from enum import Enum
import requests

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("dns-aid-registrar")


class AgentProtocol(str, Enum):
    MCP = "mcp"
    A2A = "a2a"
    CUSTOM = "custom"


class AuthMethod(str, Enum):
    MTLS = "mtls"
    OAUTH2 = "oauth2"
    API_KEY = "api-key"
    NONE = "none"


@dataclass
class AgentCapability:
    """Agent 能力描述"""
    name: str           # 如 tools/query-order
    description: str    # 能力描述
    parameters: dict    # OpenAPI 风格的参数定义


@dataclass
class AgentMetadata:
    """DNS-AID Agent 元数据"""
    proto: AgentProtocol
    version: str
    capabilities: list[str]
    auth: AuthMethod
    agent_type: str
    vendor: str
    description: str = ""
    port: int = 443
    weight: int = 5
    priority: int = 0
    ttl: int = 300
    health_endpoint: str = "/health"


class DNSAIDRegistrar:
    """
    DNS-AID Agent 注册器
    
    功能：
    1. 自动注册 AIR 和 AMR 记录
    2. 定期健康检查
    3. 心跳续期（自动更新 TTL）
    4. 优雅注销
    """

    def __init__(
        self,
        domain: str,
        agent_name: str,
        metadata: AgentMetadata,
        dns_server: str = "127.0.0.1",
        dns_port: int = 5353,
        tsig_key_name: str = "",
        tsig_key_secret: str = "",
    ):
        self.domain = domain
        self.agent_name = agent_name
        self.metadata = metadata
        self.dns_server = dns_server
        self.dns_port = dns_port
        self._running = False

        # TSIG 签名（用于安全更新 DNS 记录）
        if tsig_key_name and tsig_key_secret:
            self.keyring = dns.tsigkeyring.from_text({
                tsig_key_name: tsig_key_secret
            })
        else:
            self.keyring = None

        # 解析本地 IP（支持多网卡自定义）
        self.ip_address = self._get_local_ip()

    def _get_local_ip(self) -> str:
        """获取本机 IP 地址"""
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            s.connect(("8.8.8.8", 80))
            ip = s.getsockname()[0]
        except Exception:
            ip = "127.0.0.1"
        finally:
            s.close()
        return ip

    def _build_srv_record(self) -> str:
        """构建 SRV 记录值"""
        return (
            f"{self.agent_name}.{self.domain}. "
            f"{self.metadata.ttl} "
            f"IN SRV "
            f"{self.metadata.priority} "
            f"{self.metadata.weight} "
            f"{self.metadata.port} "
            f"{self.agent_name}.{self.domain}."
        )

    def _build_txt_records(self) -> list[str]:
        """构建 TXT 记录值"""
        records = [
            f"proto={self.metadata.proto.value}",
            f"version={self.metadata.version}",
            f"capabilities={','.join(self.metadata.capabilities)}",
            f"auth={self.metadata.auth.value}",
            f"agent-type={self.metadata.agent_type}",
            f"vendor={self.metadata.vendor}",
        ]
        if self.metadata.description:
            records.insert(0, f"description={self.metadata.description}")
        return records

    def register(self) -> bool:
        """
        注册 Agent 到 DNS-AID
        
        1. 注册 AIR（SRV 记录）
        2. 注册 AMR（TXT 记录）
        3. 注册 A/AAAA 记录（Agent 主机名 → IP）
        4. 健康检查回调（可选）
        """
        logger.info(f"正在注册 Agent: {self.agent_name}.{self.domain} → {self.ip_address}")

        update = dns.update.Update(
            self.domain,
            keyring=self.keyring,
            keyname="dns-aid-update" if self.keyring else None,
        )

        # 1. SRV 记录 — Agent Index
        update.add(
            "_index._agents",
            self.metadata.ttl,
            dns.rdatatype.SRV,
            self.metadata.priority,
            self.metadata.weight,
            self.metadata.port,
            f"{self.agent_name}.{self.domain}.",
        )

        # 2. TXT 记录 — Agent 元数据
        txt_values = self._build_txt_records()
        for txt in txt_values:
            update.add(
                "_agent._meta",
                self.metadata.ttl,
                dns.rdatatype.TXT,
                txt,
            )

        # 3. A 记录 — Agent 主机解析
        update.add(
            self.agent_name,
            self.metadata.ttl,
            dns.rdatatype.A,
            self.ip_address,
        )

        try:
            response = dns.query.tcp(
                update,
                self.dns_server,
                port=self.dns_port,
                timeout=5,
            )
            if response.rcode() == dns.rcode.NOERROR:
                logger.info(f"✅ Agent 注册成功: {self.agent_name}.{self.domain}")
                return True
            else:
                logger.error(f"❌ DNS 更新失败: rcode={response.rcode()}")
                return False
        except Exception as e:
            logger.error(f"❌ DNS 连接失败: {e}")
            return False

    def health_check(self) -> bool:
        """
        健康检查 — 检查 Agent 是否存活
        如果健康检查失败，自动降低优先级或注销
        """
        health_url = (
            f"http://{self.ip_address}:{self.metadata.port}"
            f"{self.metadata.health_endpoint}"
        )
        try:
            resp = requests.get(health_url, timeout=3)
            if resp.status_code == 200:
                return True
            return False
        except requests.RequestException:
            return False

    def heartbeat(self):
        """
        心跳续期 — 更新 DNS 记录 TTL
        
        使用场景：
        - Agent 启动后每 120 秒执行一次
        - 如果不执行，DNS 记录自然过期后 Agent 自动下线
        - 优雅处理 Agent 崩溃/断网场景
        """
        update = dns.update.Update(
            self.domain,
            keyring=self.keyring,
            keyname="dns-aid-update" if self.keyring else None,
        )

        # 更新 SRV 记录 TTL
        update.replace(
            "_index._agents",
            self.metadata.ttl,
            dns.rdatatype.SRV,
            self.metadata.priority,
            self.metadata.weight,
            self.metadata.port,
            f"{self.agent_name}.{self.domain}.",
        )

        try:
            dns.query.tcp(update, self.dns_server, port=self.dns_port, timeout=5)
            logger.debug(f"❤️  Agent 心跳发送成功 ({self.agent_name})")
        except Exception as e:
            logger.warning(f"⚠️  心跳失败: {e}")

    def unregister(self):
        """
        优雅注销 — Agent 下线时清理 DNS 记录
        """
        logger.info(f"正在注销 Agent: {self.agent_name}.{self.domain}")

        update = dns.update.Update(
            self.domain,
            keyring=self.keyring,
            keyname="dns-aid-update" if self.keyring else None,
        )

        # 删除 SRV 记录
        update.delete("_index._agents", dns.rdatatype.SRV)

        # 删除相关的 A 记录
        update.delete(self.agent_name, dns.rdatatype.A)

        try:
            dns.query.tcp(update, self.dns_server, port=self.dns_port, timeout=5)
            logger.info(f"✅ Agent 已注销: {self.agent_name}.{self.domain}")
        except Exception as e:
            logger.warning(f"⚠️  注销失败（可能已过期）: {e}")

    def run(self, heartbeat_interval: int = 120):
        """
        运行 Agent 注册器（主循环）
        
        Args:
            heartbeat_interval: 心跳间隔（秒），建议设为 TTL 的 1/3
        """
        if not self.register():
            logger.error("首次注册失败，终止运行")
            return

        self._running = True
        logger.info(
            f"🔄 Agent 注册器已启动，心跳间隔={heartbeat_interval}s"
        )

        try:
            while self._running:
                time.sleep(heartbeat_interval)
                
                # 健康检查
                if not self.health_check():
                    logger.warning("⚠️  健康检查失败，尝试重新注册...")
                    self.register()
                    continue

                # 心跳续期
                self.heartbeat()

        except KeyboardInterrupt:
            logger.info("收到中断信号，正在优雅关闭...")
        finally:
            self.unregister()
            self._running = False
            logger.info("Agent 注册器已停止")


# ============================================
# 使用示例
# ============================================

if __name__ == "__main__":
    # 定义 Agent 元数据
    metadata = AgentMetadata(
        proto=AgentProtocol.MCP,
        version="1.1",
        capabilities=[
            "tools/send-notification",
            "tools/list-channels",
            "tools/get-notification-history",
        ],
        auth=AuthMethod.MTLS,
        agent_type="api-gateway",
        vendor="acme-corp",
        description="企业级通知系统 Agent，支持多渠道消息推送",
        port=443,
        weight=10,
        priority=0,
        ttl=300,
    )

    # 创建注册器
    registrar = DNSAIDRegistrar(
        domain="agents.example.com",
        agent_name="notify-agent",
        metadata=metadata,
        dns_server="192.168.1.100",
        dns_port=5353,
        tsig_key_name="dns-aid-update.",
        tsig_key_secret="your-tsig-secret-key==",
    )

    # 启动（每 100 秒发一次心跳，TTL=300 意味着 3 次心跳失败后 Agent 自动下线）
    registrar.run(heartbeat_interval=100)

4.4 Agent 发现客户端

有注册就得有发现。下面是一个 Agent 发现客户端：

"""
dns_aid_discover.py — DNS-AID Agent Discovery Client

查询指定域下的所有可用 Agent，并获取其元数据。
支持协议过滤、能力匹配、加权随机负载均衡。
"""

import random
import dns.resolver
import dns.rdatatype
from dataclasses import dataclass
from typing import Optional


@dataclass
class DiscoveredAgent:
    """发现的 Agent 信息"""
    hostname: str
    port: int
    priority: int
    weight: int
    metadata: dict
    ip_address: Optional[str] = None


class DNSAIDDiscover:
    """DNS-AID Agent 发现客户端"""

    def __init__(self, dns_server: str = "8.8.8.8", timeout: int = 5):
        self.resolver = dns.resolver.Resolver()
        self.resolver.nameservers = [dns_server]
        self.resolver.timeout = timeout
        self.resolver.lifetime = timeout

    def discover_agents(self, domain: str) -> list[DiscoveredAgent]:
        """
        发现指定域下的所有 Agent
        
        Args:
            domain: 目标域名（如 agents.example.com）
        
        Returns:
            DiscoveredAgent 列表
        """
        # Step 1: 查询 SRV 记录（AIR）
        srv_query = f"_index._agents.{domain}"
        try:
            srv_answers = self.resolver.resolve(srv_query, "SRV")
        except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN, dns.exception.Timeout):
            print(f"❌ 未找到 Agent 索引: {srv_query}")
            return []

        # Step 2: 查询 TXT 记录（AMR）
        txt_query = f"_agent._meta.{domain}"
        txt_records = []
        try:
            txt_answers = self.resolver.resolve(txt_query, "TXT")
            for txt in txt_answers:
                txt_records.append(txt.to_text())
        except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN):
            pass

        # 解析 TXT 为字典
        metadata_map = self._parse_txt_records(txt_records)

        # Step 3: 构造 Agent 列表
        agents = []
        for srv in srv_answers:
            target = str(srv.target).rstrip(".")
            agent = DiscoveredAgent(
                hostname=target,
                port=srv.port,
                priority=srv.priority,
                weight=srv.weight,
                metadata=metadata_map.copy(),
            )

            # 解析 A 记录
            try:
                a_answers = self.resolver.resolve(target, "A")
                agent.ip_address = str(a_answers[0])
            except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN):
                pass

            agents.append(agent)

        return agents

    def _parse_txt_records(self, txt_records: list[str]) -> dict:
        """将 TXT 记录解析为元数据字典"""
        meta = {}
        for record in txt_records:
            # TXT 记录格式："key1=val1;key2=val2;..."
            content = record.strip('"')
            for pair in content.split(";"):
                pair = pair.strip()
                if "=" in pair:
                    key, value = pair.split("=", 1)
                    meta[key.strip()] = value.strip()
        return meta

    def weighted_select(self, agents: list[DiscoveredAgent]) -> Optional[DiscoveredAgent]:
        """
        加权随机选择一个 Agent（支持负载均衡）
        
        算法：
        1. 按 priority 分组（优先选 priority 最小的组）
        2. 在同组内按 weight 做加权随机
        """
        if not agents:
            return None

        # 按 priority 分组
        min_priority = min(a.priority for a in agents)
        candidates = [a for a in agents if a.priority == min_priority]

        if len(candidates) == 1:
            return candidates[0]

        # 加权随机选择
        total_weight = sum(a.weight for a in candidates)
        r = random.randint(1, total_weight)
        
        cumulative = 0
        for agent in candidates:
            cumulative += agent.weight
            if r <= cumulative:
                return agent

        return candidates[-1]

    def filter_by_protocol(self, agents: list[DiscoveredAgent], proto: str) -> list[DiscoveredAgent]:
        """按协议过滤 Agent"""
        return [
            a for a in agents
            if a.metadata.get("proto") == proto
        ]

    def filter_by_capability(self, agents: list[DiscoveredAgent], capability: str) -> list[DiscoveredAgent]:
        """按能力过滤 Agent（支持通配符 *）"""
        def _match(cap_list: str, target: str) -> bool:
            caps = cap_list.split(",")
            for c in caps:
                if c.strip() == target:
                    return True
                # 通配符匹配：tools/* 匹配 tools/query-order
                if target.endswith("/*"):
                    prefix = target.rstrip("*")
                    if c.strip().startswith(prefix):
                        return True
            return False

        return [
            a for a in agents
            if _match(a.metadata.get("capabilities", ""), capability)
        ]


# ============================================
# 使用示例
# ============================================

if __name__ == "__main__":
    discover = DNSAIDDiscover(dns_server="192.168.1.100")

    # 发现所有 Agent
    print("正在发现 Agent...")
    agents = discover.discover_agents("agents.example.com")
    print(f"发现 {len(agents)} 个 Agent:\n")

    for agent in agents:
        print(f"  🤖 {agent.hostname}:{agent.port}")
        print(f"     IP: {agent.ip_address or 'N/A'}")
        print(f"     Priority: {agent.priority}, Weight: {agent.weight}")
        print(f"     Protocol: {agent.metadata.get('proto', 'N/A')}")
        print(f"     Capabilities: {agent.metadata.get('capabilities', 'N/A')}")
        print(f"     Auth: {agent.metadata.get('auth', 'N/A')}")
        print()

    # 按协议过滤
    mcp_agents = discover.filter_by_protocol(agents, "mcp")
    print(f"MCP 协议 Agent: {len(mcp_agents)} 个")

    # 加权随机选择
    selected = discover.weighted_select(agents)
    if selected:
        print(f"\n🎯 加权选择: {selected.hostname}:{selected.port}")

    # 能力过滤
    notify_agents = discover.filter_by_capability(agents, "tools/*")
    print(f"\n包含工具调用的 Agent: {len(notify_agents)} 个")

4.5 运行效果

启动 CoreDNS：

coredns -conf Corefile
# 输出：
# .:5353
# CoreDNS-1.12.0
# linux/amd64, go1.23.4, 2970e5c4-dirty

启动 Agent 注册：

python3 dns_aid_registrar.py
# 输出：
# 正在注册 Agent: notify-agent.agents.example.com → 10.0.1.50
# ✅ Agent 注册成功: notify-agent.agents.example.com
# 🔄 Agent 注册器已启动，心跳间隔=100s

另一个终端执行发现查询：

dig @127.0.0.1 -p 5353 _index._agents.agents.example.com SRV
# 输出：
# _index._agents.agents.example.com. 3600 IN SRV 0 10 443 notify-agent.agents.example.com.

dig @127.0.0.1 -p 5353 _agent._meta.agents.example.com TXT
# 输出：
# _agent._meta.agents.example.com. 300 IN TXT "proto=mcp;version=1.1;..."

如果用 Python 发现客户端：

python3 dns_aid_discover.py
# 输出：
# 正在发现 Agent...
# 发现 3 个 Agent:
#
#   🤖 notify-agent.agents.example.com:443
#      IP: 10.0.1.50
#      Priority: 0, Weight: 10
#      Protocol: mcp
#      Capabilities: tools/send-notification,tools/list-channels
#      Auth: mtls
#
#   🤖 code-review-agent.agents.example.com:443
#      IP: 10.0.1.51
#      Priority: 0, Weight: 10
#      Protocol: a2a
#      ...

五、性能优化与生产级部署

5.1 DNS 缓存策略

DNS-AID 的性能瓶颈在于 DNS 查询延迟。好在 DNS 协议的缓存机制非常成熟：

┌────────────────────────────────────────────────────┐
│                  Agent 发现延迟                      │
├────────────────────────────────────────────────────┤
│  无缓存:  30-150ms  (DNS 递归查询)                  │
│  有缓存:   0.1-1ms  (本地 DNS 缓存命中)              │
│  预解析:   0ms      (启动时预取 + 后台刷新)           │
└────────────────────────────────────────────────────┘

生产级优化策略：

class CachingDNSAIDDiscover(DNSAIDDiscover):
    """带缓存的 DNS-AID 发现客户端"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._cache = {}
        self._cache_ttl = {}

    def discover_agents_cached(self, domain: str, max_age: int = 60) -> list[DiscoveredAgent]:
        """
        带缓存的 Agent 发现
        
        - 缓存命中 → 直接返回（<1ms）
        - 缓存过期 → 后台异步刷新
        - 刷新失败 → 使用过期缓存（服务降级）
        """
        now = time.time()

        # 缓存命中且未过期
        if domain in self._cache and now < self._cache_ttl.get(domain, 0):
            return self._cache[domain]

        try:
            agents = self.discover_agents(domain)
            self._cache[domain] = agents
            self._cache_ttl[domain] = now + max_age
            return agents
        except Exception:
            # 降级：返回过期缓存
            if domain in self._cache:
                logger.warning(f"⚠️  发现失败，使用过期缓存: {domain}")
                return self._cache[domain]
            raise

关键 TTL 建议：

5.2 高可用部署架构

生产环境的 DNS-AID 需要跨可用区部署：

# docker-compose.yml — DNS-AID 高可用部署
version: "3.8"

services:
  # 主 DNS 服务器（可用区 A）
  dns-aid-master:
    image: coredns/coredns:1.12.0
    ports:
      - "5353:5353/udp"
      - "5353:5353/tcp"
    volumes:
      - ./Corefile.master:/etc/coredns/Corefile
      - ./agents.db:/etc/coredns/agents.db
    environment:
      - COREDNS_VERBOSE=true
    restart: always
    healthcheck:
      test: ["CMD", "dig", "@127.0.0.1", "-p", "5353", "health.agents.example.com", "A"]
      interval: 10s
      retries: 3

  # 从 DNS 服务器（可用区 B）
  dns-aid-slave:
    image: coredns/coredns:1.12.0
    ports:
      - "5353:5353/udp"
      - "5353:5353/tcp"
    volumes:
      - ./Corefile.slave:/etc/coredns/Corefile
    command: >
      -conf /etc/coredns/Corefile
    depends_on:
      - dns-aid-master
    restart: always

  # Agent 注册 API
  agent-registry-api:
    build: ./registry-api
    ports:
      - "8080:8080"
    environment:
      - DNS_SERVER=dns-aid-master
      - DNS_PORT=5353
      - TSIG_KEY_NAME=dns-aid-update.
      - TSIG_KEY_SECRET=${TSIG_SECRET}
    depends_on:
      - dns-aid-master
    restart: always

  # 健康检查器
  health-checker:
    build: ./health-checker
    environment:
      - DISCOVER_DOMAIN=agents.example.com
      - DNS_SERVER=127.0.0.1
      - CHECK_INTERVAL=30
    depends_on:
      - dns-aid-master
    restart: always

5.3 大规模场景下的性能数据

在 Infoblox 的内部测试中，DNS-AID 在以下规模下表现稳定：

对比其他方案：

六、DNS-AID vs 现有方案：技术对比与选型决策

6.1 能力矩阵对比

6.2 选型建议

用 DNS-AID 的场景：

• 你的 Agent 需要跨公网发现（跨公司、跨云、跨组织）
• 你希望零运维——不想部署 Consul/etcd/注册中心
• 你已经在使用 DNSSEC（安全基线好）
• Agent 发现频率不高于几秒级别（能接受 TTL 级延迟）

不用 DNS-AID 的场景：

• 你的 Agent 都在同一个 K8s 集群内部（用 K8s DNS 即可）
• 你需要毫秒级的注册/注销（DNS TTL 做不到）
• 你的 Agent 需要根据实时负载做动态路由（DNS 做不到）

6.3 混合架构

大多数生产环境会采用混合架构：

┌──────────────────────────────────────────────┐
│            Agent Mesh（混合架构）              │
├──────────────────────────────────────────────┤
│                                              │
│  ┌─────────────┐     ┌─────────────┐         │
│  │  K8s 集群 A  │     │  K8s 集群 B  │         │
│  │  (内部 K8s   │     │  (内部 K8s   │         │
│  │   DNS 发现)  │     │   DNS 发现)  │         │
│  └──────┬──────┘     └──────┬──────┘         │
│         │                    │                 │
│         └────────┬───────────┘                 │
│                  │                             │
│         ┌────────▼────────┐                    │
│         │   DNS-AID       │                    │
│         │   (跨集群发现)   │                    │
│         └────────▲────────┘                    │
│                  │                             │
│         ┌────────┴────────┐                    │
│         │   第三方 Agent   │                    │
│         │   (外部 SaaS)    │                    │
│         └─────────────────┘                    │
│                                              │
└──────────────────────────────────────────────┘

• 集群内部：用 Kubernetes DNS + CoreDNS 做服务发现（低延迟）
• 跨集群/跨组织：通过 DNS-AID 注册入口，对外暴露
• 外部 Agent：通过 DNS-AID 被内部 Agent 发现

这种架构在实践中效果最好——你拿到了 DNS-AID 的「零基础设施」优势，又不会受限于它的 TTL 延迟。

七、DNS-AID 的安全模型深度解析

7.1 威胁模型

DNS-AID 面临的四大安全威胁：

7.2 强制 DNSSEC 验证

在发现客户端中，我们应该强制要求 DNSSEC 验证：

class SecureDNSAIDDiscover(DNSAIDDiscover):
    """带 DNSSEC 验证的 DNS-AID 发现客户端"""

    def __init__(self, dns_server: str = "1.1.1.1", timeout: int = 5):
        super().__init__(dns_server, timeout)
        # 启用 DNSSEC
        self.resolver.use_dnssec = True
        # 要求 AD（Authenticated Data）位
        self.resolver.flags = dns.flags.AD | dns.flags.RD

    def verify_with_dnssec(self, domain: str) -> bool:
        """
        验证 DNS-AID 记录的 DNSSEC 签名
        
        返回 True 表示签名有效，False 表示无效或未签名
        """
        try:
            # 查询 DNSKEY 记录
            dnskey = self.resolver.resolve(domain, "DNSKEY")
            
            # 验证 RRSIG
            rrsig = self.resolver.resolve(f"_index._agents.{domain}", "RRSIG")
            
            # 简化的验证（实际应使用 dnspython 的验证器）
            if self.resolver.response.flags & dns.flags.AD:
                logger.info(f"✅ DNSSEC 验证通过: {domain}")
                return True
            else:
                logger.warning(f"⚠️  DNSSEC 验证失败: {domain}")
                return False
                
        except (dns.resolver.NoAnswer, dns.dnssec.ValidationFailure) as e:
            logger.error(f"❌ DNSSEC 验证错误: {e}")
            return False

    def secure_discover(self, domain: str) -> Optional[list[DiscoveredAgent]]:
        """安全发现 — 要求 DNSSEC 验证通过"""
        if not self.verify_with_dnssec(domain):
            logger.error(f"❌ 拒绝发现: {domain} 的 DNSSEC 验证失败")
            return None
        return self.discover_agents(domain)

7.3 DANE 证书绑定

DANE（DNS-based Authentication of Named Entities）允许我们通过 TLSA 记录公布 Agent 的 TLS 证书信息：

# 生成 Agent 自签名证书
openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:prime256v1 \
  -days 365 -nodes \
  -keyout agent-key.pem \
  -out agent-cert.pem \
  -subj "/CN=notify-agent.agents.example.com"

# 计算证书指纹（SHA-256）
openssl x509 -in agent-cert.pem -noout -fingerprint -sha256
# 输出: SHA256 Fingerprint=AB:CD:12:34:...

# 生成 TLSA 记录值
# 用法: openssl x509 -in agent-cert.pem -noout -pubkey |
#       openssl pkey -pubin -outform DER |
#       openssl dgst -sha256 -hex

对应的 TLSA 记录：

_443._tcp.notify-agent.agents.example.com.  86400  IN  TLSA  3 1 1 (
    abcd1234efgh5678ijkl9012mnop3456qrst7890 )

这样配置后，Agent 客户端在连接时：

1. 通过 DNS-AID 查询到 notify-agent 的 IP 和端口
2. 通过 DNS 查询 _443._tcp.notify-agent... 获取 TLSA 记录
3. 建立 TLS 连接，获取对方证书
4. 验证证书指纹是否与 TLSA 记录匹配
5. 匹配 → 安全连接；不匹配 → 断开连接

这个流程相当于把你的 Agent 的身份锚定到了 DNS 链上——只要 DNS 是安全的（DNSSEC），Agent 身份就是安全的。

八、未来展望：DNS-AID 在 Agent Mesh 中的角色

8.1 从单体架构到 Agent Mesh

2025-2026年的 AI Agent 架构正在经历一场类似微服务架构的演进：从单体 Agent 到 Agent Mesh。

单体 Agent               Agent Mesh
┌─────────────┐    ┌──────────────────────────┐
│  全能 Agent   │    │  ┌────────┐  ┌────────┐  │
│  (LLM + 工具) │    │  │ 代码Agent│  │ 审查Agent│  │
├─────────────┤    │  └────────┘  └────────┘  │
│ LLM 上下文   │    │  ┌────────┐  ┌────────┐  │
│ 工具调用     │    │  │ 部署Agent│  │ 监控Agent│  │
│ 外部 API     │    │  └────────┘  └────────┘  │
└─────────────┘    └──────────────────────────┘
                        Agent → Agent 通信
                        通过 DNS-AID 发现

在这种架构下，DNS-AID 扮演的角色类似于微服务时代的 Consul——但更轻量、更标准化、更开放。

8.2 DNS-AID + MCP + A2A 的三位一体

2026年下半年，最有可能形成的技术栈是：

发现层:  DNS-AID     ← Linux 基金会 / AAIF
通信层:  MCP 1.1+    ← Anthropic 主导的社区
协作层:  A2A 2.0+    ← Google 主导的 Agent-to-Agent

这三层叠加后，开发者可以：

1. 用 DNS-AID 让 Agent 互相发现（零基础设施）
2. 用 MCP 让 Agent 调用工具和数据源
3. 用 A2A 协调多 Agent 协作完成任务

8.3 值得关注的后续演进

根据 Linux 基金会公布的路标和社区讨论，DNS-AID 在以下方向有明确的演进计划：

Q3 2026：多域联合查询
当前规范只支持单域查询。AAIF 已在讨论跨域联合查询的标准——Agent A 可以通过 _agents.{domain-b} 发现 domain-b 下的 Agent，类似于 DNS 的 CNAME 链式解析。

Q4 2026：能力描述标准化
目前的 TXT 记录的能力描述是自由文本格式。社区正在推动基于 JSON Schema 的能力描述规范，让 Agent 可以更精确地匹配对方的能力：

_agent._meta.agents.example.com.  300  IN  TXT  (
    "proto=mcp;version=1.1;"
    "schema_url=https://schema.agents.example.com/capability.json"
)

2027 H1：联邦身份集成
与 OAuth 2.0 和 DID（去中心化身份）集成，让 Agent 可以在首次发现后自动协商访问令牌，实现零配置的跨组织 Agent 调用。

九、总结

DNS-AID 不是又一个要你学习的复杂协议——它是互联网最古老也最成功的服务发现协议（DNS）在 Agent 时代的一次优雅扩展。

它的核心洞见是：我们已经有了一个全球分布式、高可用、自带缓存的发现系统——DNS。为什么不把它用起来？

三个关键判断：

1. 对开发者：如果你有域名，你立刻就能让你的 Agent 被全球发现。不需要注册中心，不需要 API Key，不需要额外运维。
2. 对架构师：DNS-AID 定位在发现层，和 MCP（通信层）、A2A（协作层）正交互补。它不是替代品，而是拼图中缺失的最后一块。
3. 对团队：小团队可以直接用静态 DNS 记录上路；大团队可以通过 CoreDNS + 自动化注册实现大规模 Agent 管理——切换成本接近于零。

最后，我想用 Linux 基金会 CEO Jim Zemlin 的话作为结尾：

"AI 智能体正迅速成为现代互联网的连接纽带。但如果缺乏安全、开放的发现基础设施，这种连接反而会带来风险。DNS-AID 有助于将智能体发现机制锚定在互联网已信赖的 DNS 基础设施之上。"

当你的 Agent 在几个月后需要找到另一个 Agent 协作时——DNS-AID 可能就是那个「怎么找到对方」的答案。而现在，正是上车的最好时机。