forked from xiaohei/taiji-AI-PAD
36 KiB
36 KiB
Agent DNS 域名映射功能实现方案
1. 功能概述
1.1 目标
实现 Agent Pod 的 IP 地址与 GoDaddy 域名子域名的自动映射，让用户可以通过域名访问 Agent 服务。
1.2 业务场景
用户创建 Agent → Agent Manager 返回 Pod IP → 自动创建 DNS A 记录 → 用户通过子域名访问 Agent
↓
用户删除 Agent → 删除 K8s Pod → 同步删除 DNS A 记录
1.3 核心信息
| 配置项 | 值 |
|---|---|
| 主域名 | taiji.ai.com |
| GoDaddy API Key | gGpZZZ2GxwVZ_7Qe1tr9rCZkErS46Za8uc4 |
| GoDaddy API Secret | AMs5RxZeaW3mFZDwXvU3Bs |
| 生产 API 地址 | https://api.godaddy.com |
| 测试 API 地址 | https://api.ote-godaddy.com |
2. 技术架构
2.1 系统架构图
┌─────────────────────────────────────────────────────────────────────────┐
│ MCP Server │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ │
│ │ agents.py │────▶│ godaddy_client │────▶│ GoDaddy API │ │
│ │ (Agent CRUD) │ │ (DNS管理) │ │ (外部服务) │ │
│ └────────┬────────┘ └──────────────────┘ └─────────────────┘ │
│ │ │
│ │ ┌──────────────────┐ │
│ └─────────────▶│ Agent Manager │ │
│ │ (K8s Pod管理) │ │
│ │ 返回 Pod IP │ │
│ └──────────────────┘ │
│ │
│ ┌─────────────────┐ ┌──────────────────┐ │
│ │ config.py │ │ models.py │ │
│ │ (GoDaddy配置) │ │ (subdomain字段) │ │
│ └─────────────────┘ └──────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
2.2 数据流程图
创建 Agent 流程:
┌──────────┐ ┌───────────────┐ ┌─────────────────┐ ┌──────────────┐
│ 用户请求 │───▶│ create_agent │───▶│ Agent Manager │───▶│ 获取 Pod IP │
└──────────┘ └───────────────┘ │ 创建 K8s Pod │ └──────┬───────┘
└─────────────────┘ │
▼
┌──────────┐ ┌───────────────┐ ┌─────────────────┐ ┌──────────────┐
│ 返回结果 │◀───│ 更新数据库 │◀───│ 创建 DNS A记录 │◀───│GoDaddy Client│
│(含子域名) │ │(存储subdomain)│ │ │ └──────────────┘
└──────────┘ └───────────────┘ └─────────────────┘
删除 Agent 流程:
┌──────────┐ ┌───────────────┐ ┌─────────────────┐ ┌──────────────┐
│ 用户请求 │───▶│ delete_agent │───▶│ Agent Manager │───▶│ 删除 K8s Pod │
└──────────┘ └───────────────┘ │ 删除 Pod │ └──────┬───────┘
└─────────────────┘ │
▼
┌──────────┐ ┌───────────────┐ ┌─────────────────┐ ┌──────────────┐
│ 返回结果 │◀───│ 删除数据库记录 │◀───│ 删除 DNS A记录 │◀───│GoDaddy Client│
└──────────┘ └───────────────┘ └─────────────────┘ └──────────────┘
3. 详细设计
3.1 子域名命名规则
格式: {agent-name}-{agent-id-short}.taiji.ai.com
示例:
- Agent 名称: my-chatbot
- Agent ID: 550e8400-e29b-41d4-a716-446655440000
- 生成子域名: my-chatbot-550e8400.taiji.ai.com
规则说明:
1. agent-name 转为小写，特殊字符替换为 '-'
2. agent-id 取前8位作为唯一标识
3. 子域名长度限制: 最大63字符（DNS标准）
4. 仅允许: a-z, 0-9, '-'（不能以'-'开头或结尾）
3.2 GoDaddy API 调用规范
3.2.1 认证方式
Authorization: sso-key {API_KEY}:{API_SECRET}
Content-Type: application/json
3.2.2 核心接口
| 操作 | 方法 | 端点 | 说明 |
|---|---|---|---|
| 创建/更新记录 | PUT | /v1/domains/{domain}/records/A/{name} |
创建或替换 A 记录 |
| 删除记录 | DELETE | /v1/domains/{domain}/records/A/{name} |
删除指定 A 记录 |
| 查询记录 | GET | /v1/domains/{domain}/records/A/{name} |
获取指定 A 记录 |
| 批量添加 | PATCH | /v1/domains/{domain}/records |
批量添加记录 |
3.2.3 请求/响应示例
创建 A 记录请求:
PUT https://api.godaddy.com/v1/domains/taiji.ai.com/records/A/my-chatbot-550e8400
Authorization: sso-key gGpZZZ2GxwVZ_7Qe1tr9rCZkErS46Za8uc4:AMs5RxZeaW3mFZDwXvU3Bs
Content-Type: application/json
[
{
"data": "203.0.113.50",
"ttl": 600
}
]
成功响应: 200 OK (无 body)
查询记录响应:
[
{
"data": "203.0.113.50",
"name": "my-chatbot-550e8400",
"ttl": 600,
"type": "A"
}
]
4. 代码实现方案
4.1 文件结构
services/mcp-server/
├── app/
│ ├── godaddy_client.py # [新增] GoDaddy DNS 客户端
│ ├── dns_service.py # [新增] DNS 服务层（业务逻辑封装）
│ └── routes/
│ └── agents.py # [修改] 集成 DNS 创建/删除
├── config.py # [修改] 添加 GoDaddy 配置
├── models.py # [修改] Agent 模型添加 subdomain 字段
├── schemas.py # [修改] 响应模型添加 subdomain
└── migrations/
└── versions/
└── xxx_add_subdomain_to_agent.py # [新增] 数据库迁移
4.2 配置项设计 (config.py)
class Settings(BaseSettings):
# ... 现有配置 ...
# GoDaddy DNS 配置
godaddy_api_key: str = os.getenv("GODADDY_API_KEY", "")
godaddy_api_secret: str = os.getenv("GODADDY_API_SECRET", "")
godaddy_base_url: str = os.getenv("GODADDY_BASE_URL", "https://api.godaddy.com")
godaddy_domain: str = os.getenv("GODADDY_DOMAIN", "taiji.ai.com")
godaddy_dns_ttl: int = int(os.getenv("GODADDY_DNS_TTL", "600")) # 10分钟
godaddy_enabled: bool = os.getenv("GODADDY_ENABLED", "true").lower() == "true"
环境变量配置:
# .env 文件
GODADDY_API_KEY=gGpZZZ2GxwVZ_7Qe1tr9rCZkErS46Za8uc4
GODADDY_API_SECRET=AMs5RxZeaW3mFZDwXvU3Bs
GODADDY_BASE_URL=https://api.godaddy.com
GODADDY_DOMAIN=taiji.ai.com
GODADDY_DNS_TTL=600
GODADDY_ENABLED=true
4.3 数据库模型变更 (models.py)
class Agent(Base):
__tablename__ = "agents"
# ... 现有字段 ...
# DNS 相关字段（新增）
subdomain = Column(String(255), nullable=True, index=True, comment="子域名")
full_domain = Column(String(512), nullable=True, comment="完整域名")
dns_record_id = Column(String(255), nullable=True, comment="DNS记录标识")
dns_status = Column(String(50), default="pending", comment="DNS状态: pending/active/failed/deleted")
dns_created_at = Column(DateTime, nullable=True, comment="DNS记录创建时间")
dns_updated_at = Column(DateTime, nullable=True, comment="DNS记录更新时间")
数据库迁移脚本:
# migrations/versions/xxx_add_subdomain_to_agent.py
"""add subdomain to agent
Revision ID: xxx
"""
def upgrade():
op.add_column('agents', sa.Column('subdomain', sa.String(255), nullable=True))
op.add_column('agents', sa.Column('full_domain', sa.String(512), nullable=True))
op.add_column('agents', sa.Column('dns_record_id', sa.String(255), nullable=True))
op.add_column('agents', sa.Column('dns_status', sa.String(50), server_default='pending'))
op.add_column('agents', sa.Column('dns_created_at', sa.DateTime, nullable=True))
op.add_column('agents', sa.Column('dns_updated_at', sa.DateTime, nullable=True))
op.create_index('ix_agents_subdomain', 'agents', ['subdomain'])
def downgrade():
op.drop_index('ix_agents_subdomain', 'agents')
op.drop_column('agents', 'dns_updated_at')
op.drop_column('agents', 'dns_created_at')
op.drop_column('agents', 'dns_status')
op.drop_column('agents', 'dns_record_id')
op.drop_column('agents', 'full_domain')
op.drop_column('agents', 'subdomain')
4.4 GoDaddy 客户端设计 (godaddy_client.py)
"""
GoDaddy DNS API 客户端
功能：
- 创建/更新/删除 DNS A 记录
- 查询 DNS 记录状态
- 自动重试和错误处理
"""
import httpx
import structlog
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
from config import settings
logger = structlog.get_logger(__name__)
@dataclass
class DNSRecord:
"""DNS 记录数据结构"""
name: str # 子域名名称
data: str # IP 地址
ttl: int = 600 # TTL（秒）
type: str = "A" # 记录类型
class GoDaddyError(Exception):
"""GoDaddy API 错误"""
def __init__(self, status_code: int, message: str, detail: Any = None):
self.status_code = status_code
self.message = message
self.detail = detail
super().__init__(f"GoDaddy API Error [{status_code}]: {message}")
class GoDaddyClient:
"""GoDaddy DNS API 客户端"""
def __init__(
self,
api_key: str = None,
api_secret: str = None,
base_url: str = None,
domain: str = None,
timeout: float = 30.0
):
self.api_key = api_key or settings.godaddy_api_key
self.api_secret = api_secret or settings.godaddy_api_secret
self.base_url = base_url or settings.godaddy_base_url
self.domain = domain or settings.godaddy_domain
self.timeout = timeout
self._client: Optional[httpx.AsyncClient] = None
@property
def _headers(self) -> Dict[str, str]:
"""构建请求头"""
return {
"Authorization": f"sso-key {self.api_key}:{self.api_secret}",
"Content-Type": "application/json",
"Accept": "application/json"
}
async def _get_client(self) -> httpx.AsyncClient:
"""获取或创建 HTTP 客户端"""
if self._client is None:
self._client = httpx.AsyncClient(
base_url=self.base_url,
headers=self._headers,
timeout=self.timeout
)
return self._client
async def close(self):
"""关闭客户端"""
if self._client:
await self._client.aclose()
self._client = None
async def create_a_record(self, name: str, ip: str, ttl: int = None) -> bool:
"""
创建或更新 A 记录
Args:
name: 子域名名称（不含主域名）
ip: IP 地址
ttl: TTL 秒数，默认使用配置值
Returns:
bool: 是否成功
"""
client = await self._get_client()
ttl = ttl or settings.godaddy_dns_ttl
url = f"/v1/domains/{self.domain}/records/A/{name}"
payload = [{"data": ip, "ttl": ttl}]
try:
response = await client.put(url, json=payload)
if response.status_code == 200:
logger.info("DNS A 记录创建成功", name=name, ip=ip, domain=self.domain)
return True
else:
error_detail = response.json() if response.content else None
raise GoDaddyError(
status_code=response.status_code,
message=f"创建 DNS 记录失败",
detail=error_detail
)
except httpx.HTTPError as e:
logger.error("GoDaddy API 请求失败", error=str(e), name=name)
raise GoDaddyError(500, f"HTTP 请求失败: {str(e)}")
async def delete_a_record(self, name: str) -> bool:
"""
删除 A 记录
Args:
name: 子域名名称
Returns:
bool: 是否成功
"""
client = await self._get_client()
url = f"/v1/domains/{self.domain}/records/A/{name}"
try:
response = await client.delete(url)
if response.status_code in (200, 204, 404):
# 404 也视为成功（记录不存在）
logger.info("DNS A 记录删除成功", name=name, domain=self.domain)
return True
else:
error_detail = response.json() if response.content else None
raise GoDaddyError(
status_code=response.status_code,
message=f"删除 DNS 记录失败",
detail=error_detail
)
except httpx.HTTPError as e:
logger.error("GoDaddy API 请求失败", error=str(e), name=name)
raise GoDaddyError(500, f"HTTP 请求失败: {str(e)}")
async def get_a_record(self, name: str) -> Optional[DNSRecord]:
"""
查询 A 记录
Args:
name: 子域名名称
Returns:
DNSRecord 或 None（不存在时）
"""
client = await self._get_client()
url = f"/v1/domains/{self.domain}/records/A/{name}"
try:
response = await client.get(url)
if response.status_code == 200:
records = response.json()
if records and len(records) > 0:
r = records[0]
return DNSRecord(
name=r.get("name", name),
data=r.get("data"),
ttl=r.get("ttl", 600),
type="A"
)
return None
elif response.status_code == 404:
return None
else:
error_detail = response.json() if response.content else None
raise GoDaddyError(
status_code=response.status_code,
message=f"查询 DNS 记录失败",
detail=error_detail
)
except httpx.HTTPError as e:
logger.error("GoDaddy API 请求失败", error=str(e), name=name)
raise GoDaddyError(500, f"HTTP 请求失败: {str(e)}")
async def update_a_record(self, name: str, ip: str, ttl: int = None) -> bool:
"""
更新 A 记录（实际上与 create 相同，PUT 会覆盖）
"""
return await self.create_a_record(name, ip, ttl)
async def list_a_records(self) -> List[DNSRecord]:
"""
列出所有 A 记录
Returns:
DNSRecord 列表
"""
client = await self._get_client()
url = f"/v1/domains/{self.domain}/records/A"
try:
response = await client.get(url)
if response.status_code == 200:
records = response.json()
return [
DNSRecord(
name=r.get("name"),
data=r.get("data"),
ttl=r.get("ttl", 600),
type="A"
)
for r in records
]
else:
error_detail = response.json() if response.content else None
raise GoDaddyError(
status_code=response.status_code,
message=f"列出 DNS 记录失败",
detail=error_detail
)
except httpx.HTTPError as e:
logger.error("GoDaddy API 请求失败", error=str(e))
raise GoDaddyError(500, f"HTTP 请求失败: {str(e)}")
# 全局客户端实例（单例模式）
_godaddy_client: Optional[GoDaddyClient] = None
def get_godaddy_client() -> GoDaddyClient:
"""获取 GoDaddy 客户端实例"""
global _godaddy_client
if _godaddy_client is None:
_godaddy_client = GoDaddyClient()
return _godaddy_client
4.5 DNS 服务层设计 (dns_service.py)
"""
DNS 服务层
封装 DNS 相关业务逻辑：
- 子域名生成规则
- DNS 记录生命周期管理
- 与数据库同步
"""
import re
import structlog
from datetime import datetime
from typing import Optional, Tuple
from config import settings
from .godaddy_client import get_godaddy_client, GoDaddyError
logger = structlog.get_logger(__name__)
class DNSService:
"""DNS 服务"""
def __init__(self):
self.domain = settings.godaddy_domain
self.enabled = settings.godaddy_enabled
def generate_subdomain(self, agent_name: str, agent_id: str) -> str:
"""
生成子域名
规则：
1. agent_name 转小写
2. 特殊字符替换为 '-'
3. 移除连续的 '-'
4. 添加 agent_id 前8位
5. 确保不以 '-' 开头或结尾
6. 总长度限制 63 字符
Args:
agent_name: Agent 名称
agent_id: Agent UUID
Returns:
子域名（不含主域名）
"""
# 清理名称
clean_name = agent_name.lower()
clean_name = re.sub(r'[^a-z0-9-]', '-', clean_name)
clean_name = re.sub(r'-+', '-', clean_name)
clean_name = clean_name.strip('-')
# 截取 ID 前8位
short_id = agent_id.replace('-', '')[:8].lower()
# 组合
subdomain = f"{clean_name}-{short_id}"
# 长度限制（DNS 标准最大63字符）
if len(subdomain) > 63:
max_name_len = 63 - len(short_id) - 1
clean_name = clean_name[:max_name_len].rstrip('-')
subdomain = f"{clean_name}-{short_id}"
return subdomain
def get_full_domain(self, subdomain: str) -> str:
"""获取完整域名"""
return f"{subdomain}.{self.domain}"
async def create_dns_record(
self,
agent_name: str,
agent_id: str,
ip: str
) -> Tuple[str, str]:
"""
为 Agent 创建 DNS 记录
Args:
agent_name: Agent 名称
agent_id: Agent UUID
ip: Pod IP 地址
Returns:
(subdomain, full_domain)
Raises:
GoDaddyError: DNS 创建失败
"""
if not self.enabled:
logger.warning("DNS 功能未启用，跳过创建")
return ("", "")
if not ip:
logger.warning("IP 地址为空，跳过 DNS 创建", agent_id=agent_id)
return ("", "")
subdomain = self.generate_subdomain(agent_name, agent_id)
full_domain = self.get_full_domain(subdomain)
client = get_godaddy_client()
try:
await client.create_a_record(subdomain, ip)
logger.info(
"DNS 记录创建成功",
agent_id=agent_id,
subdomain=subdomain,
full_domain=full_domain,
ip=ip
)
return (subdomain, full_domain)
except GoDaddyError as e:
logger.error(
"DNS 记录创建失败",
agent_id=agent_id,
subdomain=subdomain,
ip=ip,
error=str(e)
)
raise
async def delete_dns_record(self, subdomain: str) -> bool:
"""
删除 DNS 记录
Args:
subdomain: 子域名
Returns:
是否成功
"""
if not self.enabled:
logger.warning("DNS 功能未启用，跳过删除")
return True
if not subdomain:
logger.warning("子域名为空，跳过 DNS 删除")
return True
client = get_godaddy_client()
try:
await client.delete_a_record(subdomain)
logger.info("DNS 记录删除成功", subdomain=subdomain)
return True
except GoDaddyError as e:
logger.error("DNS 记录删除失败", subdomain=subdomain, error=str(e))
# 删除失败不阻塞 Agent 删除流程
return False
async def update_dns_record(
self,
subdomain: str,
new_ip: str
) -> bool:
"""
更新 DNS 记录的 IP 地址
Args:
subdomain: 子域名
new_ip: 新的 IP 地址
Returns:
是否成功
"""
if not self.enabled:
return False
client = get_godaddy_client()
try:
await client.update_a_record(subdomain, new_ip)
logger.info("DNS 记录更新成功", subdomain=subdomain, new_ip=new_ip)
return True
except GoDaddyError as e:
logger.error("DNS 记录更新失败", subdomain=subdomain, error=str(e))
return False
async def check_dns_record(self, subdomain: str) -> Optional[str]:
"""
检查 DNS 记录状态
Args:
subdomain: 子域名
Returns:
IP 地址，或 None（不存在）
"""
if not self.enabled:
return None
client = get_godaddy_client()
try:
record = await client.get_a_record(subdomain)
return record.data if record else None
except GoDaddyError:
return None
# 全局服务实例
_dns_service: Optional[DNSService] = None
def get_dns_service() -> DNSService:
"""获取 DNS 服务实例"""
global _dns_service
if _dns_service is None:
_dns_service = DNSService()
return _dns_service
4.6 Agent 路由修改 (agents.py)
# 在 create_agent 函数中集成 DNS 创建
from ..dns_service import get_dns_service, GoDaddyError as DNSError
@router.post("", response_model=AgentCard)
async def create_agent(
request: AgentCreateRequest,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> AgentCard:
"""创建新的 Agent"""
# ... 现有代码：创建 Agent 和 K8s Pod ...
# ========== 新增: DNS 记录创建 ==========
if agent.pod_ip and settings.godaddy_enabled:
try:
dns_service = get_dns_service()
subdomain, full_domain = await dns_service.create_dns_record(
agent_name=agent.name,
agent_id=str(agent.id),
ip=agent.pod_ip
)
# 更新数据库
agent.subdomain = subdomain
agent.full_domain = full_domain
agent.dns_status = "active"
agent.dns_created_at = datetime.utcnow()
# 更新 access_url
if full_domain:
agent.access_url = f"http://{full_domain}:{agent.service_port or 8000}"
logger.info(
"DNS 记录创建成功",
agent_id=str(agent.id),
subdomain=subdomain,
full_domain=full_domain
)
except DNSError as e:
# DNS 创建失败不阻塞 Agent 创建
logger.warning(
"DNS 记录创建失败，Agent 仍可通过 IP 访问",
agent_id=str(agent.id),
error=str(e)
)
agent.dns_status = "failed"
# ========================================
await db.commit()
await db.refresh(agent)
return _build_agent_card(agent)
@router.delete("/{agent_id}")
async def delete_agent(
agent_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
) -> Dict[str, Any]:
"""删除 Agent"""
# ... 现有代码：权限检查 ...
# ========== 新增: DNS 记录删除 ==========
if agent.subdomain and settings.godaddy_enabled:
try:
dns_service = get_dns_service()
await dns_service.delete_dns_record(agent.subdomain)
logger.info(
"DNS 记录删除成功",
agent_id=agent_id,
subdomain=agent.subdomain
)
except Exception as e:
# DNS 删除失败不阻塞 Agent 删除
logger.warning(
"DNS 记录删除失败",
agent_id=agent_id,
subdomain=agent.subdomain,
error=str(e)
)
# ========================================
# ... 现有代码：删除 K8s Pod 和数据库记录 ...
4.7 响应模型更新 (schemas.py)
class AgentCard(BaseModel):
"""Agent 卡片信息"""
# ... 现有字段 ...
# DNS 相关字段（新增）
subdomain: Optional[str] = None
full_domain: Optional[str] = None
dns_status: Optional[str] = None # pending/active/failed/deleted
5. API 接口设计
5.1 新增 DNS 管理接口
5.1.1 获取 Agent DNS 状态
GET /api/agents/{agent_id}/dns
Response 200:
{
"agent_id": "550e8400-e29b-41d4-a716-446655440000",
"subdomain": "my-chatbot-550e8400",
"full_domain": "my-chatbot-550e8400.taiji.ai.com",
"ip": "203.0.113.50",
"ttl": 600,
"dns_status": "active",
"created_at": "2026-01-14T12:00:00Z"
}
5.1.2 刷新 DNS 记录
POST /api/agents/{agent_id}/dns/refresh
Response 200:
{
"status": "success",
"message": "DNS 记录已刷新",
"old_ip": "10.0.0.5",
"new_ip": "203.0.113.50"
}
5.1.3 手动删除 DNS 记录
DELETE /api/agents/{agent_id}/dns
Response 200:
{
"status": "success",
"message": "DNS 记录已删除"
}
6. 错误处理策略
6.1 错误码定义
| 错误码 | 场景 | 处理方式 |
|---|---|---|
| DNS_001 | GoDaddy API 认证失败 | 检查 API Key/Secret，告警 |
| DNS_002 | 域名不存在或无权限 | 检查域名配置，告警 |
| DNS_003 | 记录创建失败 | 重试3次，失败后设置 dns_status=failed |
| DNS_004 | 记录删除失败 | 记录日志，不阻塞主流程 |
| DNS_005 | IP 地址无效 | 跳过 DNS 创建，设置 dns_status=pending |
6.2 失败恢复机制
# 定时任务：修复失败的 DNS 记录
async def repair_failed_dns_records():
"""
定时检查并修复 dns_status='failed' 的记录
执行频率：每5分钟
"""
async with get_db() as db:
# 查询失败的记录
result = await db.execute(
select(Agent).where(
Agent.dns_status == "failed",
Agent.pod_ip.isnot(None)
)
)
failed_agents = result.scalars().all()
dns_service = get_dns_service()
for agent in failed_agents:
try:
subdomain, full_domain = await dns_service.create_dns_record(
agent.name, str(agent.id), agent.pod_ip
)
agent.subdomain = subdomain
agent.full_domain = full_domain
agent.dns_status = "active"
agent.dns_updated_at = datetime.utcnow()
logger.info("DNS 记录修复成功", agent_id=str(agent.id))
except Exception as e:
logger.warning("DNS 记录修复失败", agent_id=str(agent.id), error=str(e))
await db.commit()
7. 测试方案
7.1 单元测试
# tests/test_godaddy_client.py
import pytest
from app.godaddy_client import GoDaddyClient, GoDaddyError
@pytest.fixture
def client():
return GoDaddyClient(
api_key="test_key",
api_secret="test_secret",
base_url="https://api.ote-godaddy.com", # 使用测试环境
domain="taiji.ai.com"
)
@pytest.mark.asyncio
async def test_create_a_record(client):
"""测试创建 A 记录"""
result = await client.create_a_record("test-agent", "203.0.113.50")
assert result is True
@pytest.mark.asyncio
async def test_delete_a_record(client):
"""测试删除 A 记录"""
result = await client.delete_a_record("test-agent")
assert result is True
@pytest.mark.asyncio
async def test_get_a_record(client):
"""测试查询 A 记录"""
await client.create_a_record("test-agent", "203.0.113.50")
record = await client.get_a_record("test-agent")
assert record is not None
assert record.data == "203.0.113.50"
7.2 集成测试
# tests/test_dns_integration.py
@pytest.mark.asyncio
async def test_agent_creation_with_dns():
"""测试 Agent 创建时自动创建 DNS 记录"""
# 1. 创建 Agent
response = await client.post("/api/agents", json={
"name": "test-chatbot",
"template": "base-agent",
"description": "Test agent"
})
assert response.status_code == 200
agent = response.json()
# 2. 验证 DNS 字段
assert agent["subdomain"] is not None
assert agent["full_domain"].endswith("taiji.ai.com")
assert agent["dns_status"] == "active"
# 3. 验证 GoDaddy 记录存在
dns_response = await client.get(f"/api/agents/{agent['id']}/dns")
assert dns_response.status_code == 200
# 4. 清理
await client.delete(f"/api/agents/{agent['id']}")
@pytest.mark.asyncio
async def test_agent_deletion_removes_dns():
"""测试 Agent 删除时同步删除 DNS 记录"""
# 1. 创建 Agent
response = await client.post("/api/agents", json={...})
agent = response.json()
subdomain = agent["subdomain"]
# 2. 删除 Agent
await client.delete(f"/api/agents/{agent['id']}")
# 3. 验证 DNS 记录已删除
dns_client = get_godaddy_client()
record = await dns_client.get_a_record(subdomain)
assert record is None
7.3 手动测试脚本
# scripts/test_godaddy_api.py
"""
手动测试 GoDaddy API 连通性
使用方法：
python scripts/test_godaddy_api.py
"""
import asyncio
import sys
sys.path.append(".")
from app.godaddy_client import GoDaddyClient
async def main():
client = GoDaddyClient()
print("=" * 50)
print("GoDaddy API 连通性测试")
print("=" * 50)
# 1. 测试创建记录
print("\n1. 创建测试 A 记录...")
try:
await client.create_a_record("test-dns-integration", "1.2.3.4")
print(" ✓ 创建成功")
except Exception as e:
print(f" ✗ 创建失败: {e}")
return
# 2. 测试查询记录
print("\n2. 查询 A 记录...")
try:
record = await client.get_a_record("test-dns-integration")
if record:
print(f" ✓ 查询成功: {record.name} -> {record.data}")
else:
print(" ✗ 记录不存在")
except Exception as e:
print(f" ✗ 查询失败: {e}")
# 3. 测试删除记录
print("\n3. 删除测试 A 记录...")
try:
await client.delete_a_record("test-dns-integration")
print(" ✓ 删除成功")
except Exception as e:
print(f" ✗ 删除失败: {e}")
# 4. 列出所有 A 记录
print("\n4. 列出所有 A 记录...")
try:
records = await client.list_a_records()
print(f" 共 {len(records)} 条记录:")
for r in records[:5]:
print(f" - {r.name} -> {r.data} (TTL: {r.ttl})")
if len(records) > 5:
print(f" ... 还有 {len(records) - 5} 条")
except Exception as e:
print(f" ✗ 列出失败: {e}")
await client.close()
print("\n" + "=" * 50)
print("测试完成")
if __name__ == "__main__":
asyncio.run(main())
8. 部署检查清单
8.1 环境变量配置
# 生产环境必须配置
GODADDY_API_KEY=gGpZZZ2GxwVZ_7Qe1tr9rCZkErS46Za8uc4
GODADDY_API_SECRET=AMs5RxZeaW3mFZDwXvU3Bs
GODADDY_DOMAIN=taiji.ai.com
GODADDY_BASE_URL=https://api.godaddy.com
GODADDY_ENABLED=true
GODADDY_DNS_TTL=600
8.2 数据库迁移
# 执行迁移
alembic upgrade head
# 验证字段
psql -c "SELECT column_name FROM information_schema.columns WHERE table_name='agents' AND column_name LIKE 'dns%';"
8.3 功能验证
- GoDaddy API 认证成功
- 创建 Agent 后 subdomain 字段有值
- full_domain 格式正确 (
xxx.taiji.ai.com) - 通过域名可以访问 Agent 服务
- 删除 Agent 后 DNS 记录被清理
- 失败情况下有正确的日志和 dns_status
9. 后续优化建议
9.1 短期优化
- DNS 缓存: 本地缓存 DNS 记录状态，减少 API 调用
- 批量操作: 支持批量创建/删除 DNS 记录
- 监控告警: DNS 操作失败时发送告警
9.2 中期优化
- HTTPS 支持: 集成 Let's Encrypt 自动申请 SSL 证书
- 自定义域名: 支持用户绑定自己的域名
- DNS 健康检查: 定期检查 DNS 解析是否正常
9.3 长期规划
- 多 DNS 提供商: 支持 Cloudflare、AWS Route53 等
- 智能 DNS: 基于地理位置的 DNS 解析
- 域名市场: 提供子域名购买/租赁服务