anp-agent

Original source / Raw SKILL.md

---
name: anp-agent
description: ANP 协议跨 Agent 调用技能。通过 did:wba 去中心化身份，调用 ANP 网络中的任意 Agent（如高德地图、酒店预订、快递查询等）。当用户提到 ANP、调用 Agent、订酒店、查快递、查地图、路线规划时触发此技能。
---

# ANP Agent Skill

通过 ANP (Agent Network Protocol) 协议，使用去中心化身份 (did:wba) 调用远程 Agent。

## 使用场景

当用户需要：
- 调用 ANP 网络中的 Agent（高德地图、酒店、快递等）
- 搜索地点、规划路线、查询天气
- 预订酒店、查询快递
- 发现新的 ANP Agent

## 调用流程

### 1. 连接 Agent（查看能力）

给定 AD URL，获取 Agent 的可用方法：

```bash
python scripts/anp_cli.py connect "<AD_URL>"
```

示例：
```bash
python scripts/anp_cli.py connect "https://agent-connect.ai/mcp/agents/amap/ad.json"
```

### 2. 调用方法

使用已注册 ID 或 AD URL 调用：

```bash
python scripts/anp_cli.py call <id|ad_url> <method> '<json_params>'
```

示例：
```bash
# 搜索北京咖啡厅
python scripts/anp_cli.py call amap maps_text_search '{"keywords":"咖啡厅","city":"北京"}'

# 查询天气
python scripts/anp_cli.py call amap maps_weather '{"city":"上海"}'
```

### 3. 管理 Agent

```bash
# 列出已注册
python scripts/anp_cli.py list

# 添加新 Agent
python scripts/anp_cli.py add <id> "<ad_url>"

# 移除
python scripts/anp_cli.py remove <id>
```

## 已注册 Agent

| ID | 名称 | AD URL |
|----|------|--------|
| amap | 高德地图 | https://agent-connect.ai/mcp/agents/amap/ad.json |
| kuaidi | 快递查询 | https://agent-connect.ai/mcp/agents/kuaidi/ad.json |
| hotel | 酒店预订 | https://agent-connect.ai/agents/hotel-assistant/ad.json |
| juhe | 聚合查询 | https://agent-connect.ai/mcp/agents/juhe/ad.json |
| navigation | Agent导航 | https://agent-search.ai/agents/navigation/ad.json |

## 高德地图常用方法

| 方法 | 功能 | 参数示例 |
|------|------|----------|
| maps_text_search | 搜索地点 | `{"keywords":"咖啡厅","city":"北京"}` |
| maps_weather | 查询天气 | `{"city":"上海"}` |
| maps_direction_driving | 驾车路线 | `{"origin":"经度,纬度","destination":"经度,纬度"}` |
| maps_around_search | 周边搜索 | `{"location":"经度,纬度","keywords":"美食"}` |

## 目录结构

```
anp-agent/
├── SKILL.md              # 本文件
├── config/
│   ├── did.json          # DID 文档（公钥）
│   ├── private-key.pem   # 私钥（签名用）
│   ├── agents.json       # 已注册的 Agent 列表
│   └── .gitignore        # 防止私钥泄露
└── scripts/
    └── anp_cli.py        # 主程序
```

## 依赖

```bash
pip install anp aiohttp
```


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### scripts/anp_cli.py

```python
#!/usr/bin/env python3
"""
ANP 动态调用工具 - 给 AD URL 就能连

用法：
    # 直接用 AD URL 连接查看能力
    python anp_cli.py connect "https://xxx/ad.json"
    
    # 调用已注册的 Agent
    python anp_cli.py call amap maps_text_search '{"keywords":"咖啡厅","city":"北京"}'
    
    # 添加新 Agent
    python anp_cli.py add myagent "https://example.com/ad.json"
    
    # 列出所有已注册 Agent
    python anp_cli.py list
"""

import asyncio
import json
import sys
from pathlib import Path
from typing import Optional
import aiohttp

from anp.anp_crawler import ANPCrawler

CONFIG_DIR = Path(__file__).parent.parent / "config"
AGENTS_FILE = CONFIG_DIR / "agents.json"

def load_agents():
    if AGENTS_FILE.exists():
        with open(AGENTS_FILE) as f:
            return json.load(f)
    return {"agents": []}

def save_agents(data):
    with open(AGENTS_FILE, "w") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

def get_crawler():
    return ANPCrawler(
        did_document_path=str(CONFIG_DIR / "did.json"),
        private_key_path=str(CONFIG_DIR / "private-key.pem")
    )

async def fetch_ad(ad_url: str) -> dict:
    """获取 AD 文档"""
    async with aiohttp.ClientSession() as session:
        async with session.get(ad_url, timeout=aiohttp.ClientTimeout(total=15)) as resp:
            if resp.status == 200:
                return await resp.json()
            raise Exception(f"获取 AD 失败: HTTP {resp.status}")

async def get_interface(ad: dict) -> tuple:
    """从 AD 获取接口定义和 RPC 端点"""
    interfaces = ad.get("interfaces", [])
    if not interfaces:
        return None, None, []
    
    interface = interfaces[0]
    interface_url = interface.get("url")
    
    rpc_endpoint = None
    methods = []
    
    if interface_url:
        async with aiohttp.ClientSession() as session:
            async with session.get(interface_url, timeout=aiohttp.ClientTimeout(total=15)) as resp:
                if resp.status == 200:
                    interface_doc = await resp.json()
                    servers = interface_doc.get("servers", [])
                    if servers:
                        rpc_endpoint = servers[0].get("url")
                    methods = interface_doc.get("methods", [])
    
    return interface_url, rpc_endpoint, methods

async def connect(ad_url: str, show_methods: bool = True):
    """连接到 Agent，显示其能力"""
    print(f"正在连接: {ad_url}\n")
    
    ad = await fetch_ad(ad_url)
    
    print(f"Agent: {ad.get('name', 'Unknown')}")
    print(f"DID: {ad.get('did', 'N/A')}")
    print(f"描述: {ad.get('description', 'N/A')}")
    
    interface_url, rpc_endpoint, methods = await get_interface(ad)
    
    if rpc_endpoint:
        print(f"RPC 端点: {rpc_endpoint}")
    
    if show_methods and methods:
        print(f"\n可用方法 ({len(methods)} 个):")
        for m in methods[:20]:
            desc = m.get("description", "")[:60]
            print(f"  - {m.get('name')}: {desc}")
        if len(methods) > 20:
            print(f"  ... 还有 {len(methods) - 20} 个方法")
    
    return ad, rpc_endpoint, methods

async def call_method(ad_url_or_id: str, method: str, params: dict):
    """调用 Agent 方法"""
    agents_data = load_agents()
    
    ad_url = ad_url_or_id
    for agent in agents_data.get("agents", []):
        if agent.get("id") == ad_url_or_id:
            ad_url = agent.get("ad_url")
            break
    
    print(f"正在获取 Agent 信息...")
    ad = await fetch_ad(ad_url)
    _, rpc_endpoint, _ = await get_interface(ad)
    
    if not rpc_endpoint:
        print("错误: 无法获取 RPC 端点")
        return
    
    print(f"调用: {method}")
    print(f"端点: {rpc_endpoint}")
    print(f"参数: {params}\n")
    
    crawler = get_crawler()
    result = await crawler.execute_json_rpc(
        endpoint=rpc_endpoint,
        method=method,
        params=params
    )
    
    print("=== 结果 ===")
    print(json.dumps(result, indent=2, ensure_ascii=False))

async def add_agent(agent_id: str, ad_url: str):
    """添加新 Agent 到配置"""
    print(f"正在验证 Agent: {ad_url}")
    
    try:
        ad = await fetch_ad(ad_url)
        name = ad.get("name", agent_id)
        
        agents_data = load_agents()
        
        for agent in agents_data.get("agents", []):
            if agent.get("id") == agent_id:
                agent["ad_url"] = ad_url
                agent["name"] = name
                save_agents(agents_data)
                print(f"✅ 已更新: {agent_id} -> {name}")
                return
        
        agents_data["agents"].append({
            "id": agent_id,
            "name": name,
            "ad_url": ad_url
        })
        save_agents(agents_data)
        print(f"✅ 已添加: {agent_id} -> {name}")
        print(f"   AD: {ad_url}")
        
    except Exception as e:
        print(f"❌ 添加失败: {e}")

def list_agents():
    """列出所有已注册 Agent"""
    agents_data = load_agents()
    agents = agents_data.get("agents", [])
    
    if not agents:
        print("暂无已注册的 Agent")
        print("使用 'python anp_cli.py add <id> <ad_url>' 添加")
        return
    
    print(f"\n已注册的 ANP Agent ({len(agents)} 个):\n")
    for agent in agents:
        print(f"  {agent.get('id')}: {agent.get('name')}")
        print(f"      {agent.get('ad_url')}")
        print()

def show_help():
    print("""
ANP 动态调用工具

用法:
    python anp_cli.py <命令> [参数...]

命令:
    connect <ad_url>                      连接并查看 Agent 能力
    call <id|ad_url> <method> <params>    调用方法
    add <id> <ad_url>                     添加新 Agent
    remove <id>                           移除 Agent
    list                                  列出所有 Agent

示例:
    # 连接查看
    python anp_cli.py connect "https://agent-connect.ai/mcp/agents/amap/ad.json"
    
    # 调用方法 (用已注册的 id)
    python anp_cli.py call amap maps_text_search '{"keywords":"咖啡厅","city":"北京"}'
    
    # 直接用 AD URL 调用
    python anp_cli.py call "https://xxx/ad.json" some_method '{"key":"value"}'
    
    # 添加新 Agent
    python anp_cli.py add myagent "https://example.com/ad.json"
""")

async def main():
    if len(sys.argv) < 2:
        show_help()
        return
    
    cmd = sys.argv[1]
    
    if cmd in ["help", "-h", "--help"]:
        show_help()
    
    elif cmd == "connect":
        if len(sys.argv) < 3:
            print("用法: python anp_cli.py connect <ad_url>")
            return
        await connect(sys.argv[2])
    
    elif cmd == "call":
        if len(sys.argv) < 5:
            print("用法: python anp_cli.py call <id|ad_url> <method> '<params_json>'")
            return
        target = sys.argv[2]
        method = sys.argv[3]
        params = json.loads(sys.argv[4])
        await call_method(target, method, params)
    
    elif cmd == "add":
        if len(sys.argv) < 4:
            print("用法: python anp_cli.py add <id> <ad_url>")
            return
        await add_agent(sys.argv[2], sys.argv[3])
    
    elif cmd == "list":
        list_agents()
    
    elif cmd == "remove":
        if len(sys.argv) < 3:
            print("用法: python anp_cli.py remove <id>")
            return
        agent_id = sys.argv[2]
        agents_data = load_agents()
        agents_data["agents"] = [a for a in agents_data.get("agents", []) if a.get("id") != agent_id]
        save_agents(agents_data)
        print(f"已移除: {agent_id}")
    
    else:
        print(f"未知命令: {cmd}")
        show_help()

if __name__ == "__main__":
    asyncio.run(main())

```



---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### README.md

```markdown
# ANP Agent Skill

通过 ANP 协议调用远程 Agent（高德地图、快递查询、酒店预订等）。

## 一键部署

```bash
# 1. 克隆/下载本目录

# 2. 运行安装脚本
./setup.sh

# 3. 开始使用
python scripts/anp_cli.py list
```

## 使用方式

### 查看已注册的 Agent

```bash
python scripts/anp_cli.py list
```

### 调用 Agent

```bash
# 搜索地点
python scripts/anp_cli.py call amap maps_text_search '{"keywords":"咖啡厅","city":"北京"}'

# 查询天气
python scripts/anp_cli.py call amap maps_weather '{"city":"上海"}'

# 驾车路线规划
python scripts/anp_cli.py call amap maps_direction_driving '{"origin":"116.481028,39.989643","destination":"116.434446,39.90816"}'

# 周边搜索
python scripts/anp_cli.py call amap maps_around_search '{"location":"116.396652,40.003128","keywords":"酒店","radius":"3000"}'
```

### 连接新 Agent

```bash
# 查看 Agent 能力
python scripts/anp_cli.py connect "https://agent-connect.ai/mcp/agents/amap/ad.json"

# 添加到本地
python scripts/anp_cli.py add myagent "https://example.com/ad.json"

# 移除
python scripts/anp_cli.py remove myagent
```

## 已内置的 Agent

| ID | 名称 | 功能 |
|----|------|------|
| amap | 高德地图 | 地点搜索、路线规划、天气查询、周边搜索 |
| kuaidi | 快递查询 | 快递单号追踪 |
| hotel | 酒店预订 | 搜索酒店、查询房价 |
| juhe | 聚合查询 | 多种生活服务查询 |
| navigation | Agent导航 | 发现更多 Agent |

## 高德地图常用方法

| 方法 | 功能 | 参数 |
|------|------|------|
| maps_text_search | 搜索地点 | keywords, city |
| maps_weather | 查询天气 | city |
| maps_direction_driving | 驾车路线 | origin, destination |
| maps_around_search | 周边搜索 | location, keywords, radius |

## 身份配置（可选）

### 情况一：已有 DID 身份

如果你已有 ANP 网络的 DID 身份，将文件放到 `config/` 目录：

```
config/
├── did.json          # 你的 DID 文档
└── private-key.pem   # 你的私钥
```

### 情况二：没有 DID

运行 `setup.sh` 会自动生成本地临时身份，可以正常调用大部分 Agent。

如需注册正式 DID，访问：https://didhost.cc 或其他 DID 托管服务。

## 目录结构

```
anp-agent/
├── README.md           # 本文件
├── SKILL.md            # AI 助手技能描述
├── setup.sh            # 一键安装脚本
├── requirements.txt    # Python 依赖
├── config/
│   ├── agents.json     # 已注册的 Agent 列表
│   ├── did.json        # DID 身份文档
│   ├── private-key.pem # 私钥（自动生成或自己放）
│   └── .gitignore      # 防止私钥泄露
└── scripts/
    └── anp_cli.py      # 主程序
```

## 常见问题

### Q: 什么是 ANP？
ANP (Agent Network Protocol) 是 Agent 互联协议，让不同的 AI Agent 可以相互调用。

### Q: 什么是 DID？
DID (Decentralized Identifier) 是去中心化身份，用于 Agent 之间的身份验证。

### Q: 没有 DID 能用吗？
可以。setup.sh 会生成本地身份，大部分公开 Agent 都能调用。

### Q: 如何获取正式 DID？
访问 https://didhost.cc 注册，获得托管的 DID 身份。

## License

MIT

```