"Connect Any Agent, Any Protocol"
Design Philosophy
Problem
The current Agent protocol ecosystem is fragmented:
- MCP: Anthropic's tool calling protocol
- A2A: Google's inter-Agent collaboration protocol
- ANP: Decentralized identity + Agent network protocol
- AITP: NEAR's interaction transaction protocol
- ...
Developers need to learn different SDKs and implement different calling logic for each protocol.
Solution
UniAgent provides a unified abstraction layer, with one set of APIs adapting to all protocols:
from uni_agent import UniAgent
agent = UniAgent()
# 调用 ANP Agent
agent.call("amap@anp", "maps_weather", {"city": "北京"})
# 调用 MCP Server
agent.call("filesystem@mcp", "read_file", {"path": "/tmp/a.txt"})
# 调用 A2A Agent
agent.call("assistant@a2a", "chat", {"message": "hello"})
# 调用 AITP Agent（带支付）
agent.call("shop@aitp", "purchase", {"item": "coffee", "amount": 10})
Architecture Design
┌─────────────────────────────────────────────────────────┐
│ UniAgent │
│ 统一调用接口 │
├─────────────────────────────────────────────────────────┤
│ call(agent_id, method, params) -> result │
│ discover(capability) -> List[Agent] │
│ connect(agent_id) -> Connection │
└────────────────────────┬────────────────────────────────┘
│
┌──────────┴──────────┐
│ Protocol Router │
│ 协议路由 & 适配 │
└──────────┬──────────┘
│
┌─────────┬───────────┼───────────┬─────────┐
▼ ▼ ▼ ▼ ▼
┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐
│ ANP │ │ MCP │ │ A2A │ │ AITP │ │ ... │
│Adapter│ │Adapter│ │Adapter│ │Adapter│ │Adapter│
└──────┘ └──────┘ └──────┘ └──────┘ └──────┘
Core Concepts
1. Agent ID Format
<agent_name>@<protocol>
示例：
- amap@anp # ANP 协议的高德地图 Agent
- filesystem@mcp # MCP 协议的文件系统 Server
- gemini@a2a # A2A 协议的 Gemini Agent
- shop@aitp # AITP 协议的商店 Agent
2. Unified Calling Interface
result = agent.call(
agent_id="amap@anp", # Agent 标识
method="maps_weather", # 方法名
params={"city": "北京"}, # 参数
timeout=30 # 可选超时
)
3. Capability Discovery
# 发现所有能提供天气服务的 Agent
agents = agent.discover("weather")
# 返回: [
# {"id": "amap@anp", "protocol": "anp", "methods": [...]},
# {"id": "weather@mcp", "protocol": "mcp", "methods": [...]}
# ]
4. Protocol Adapter Interface
class ProtocolAdapter(ABC):
"""协议适配器基类"""
@abstractmethod
def connect(self, agent_config: dict) -> Connection:
"""建立连接"""
pass
@abstractmethod
def call(self, connection: Connection, method: str, params: dict) -> dict:
"""调用方法"""
pass
@abstractmethod
def discover(self, capability: str) -> List[AgentInfo]:
"""发现 Agent"""
pass
@abstractmethod
def close(self, connection: Connection):
"""关闭连接"""
pass
Supported Protocols
| Protocol | Status | Adapter | Description |
|---|---|---|---|
| ANP | ✅ Implemented | adapters/anp.py |
Decentralized identity + Agent network |
| MCP | ✅ Implemented | adapters/mcp.py |
LLM tool calling |
| A2A | ✅ Implemented | adapters/a2a.py |
Inter-Agent collaboration |
| AITP | ✅ Implemented | adapters/aitp.py |
Interaction + transactions |
| Agent Protocol | ✅ Implemented | adapters/agent_protocol.py |
REST API |
| LMOS | ✅ Implemented | adapters/lmos.py |
Enterprise-grade platform |
Usage
CLI Invocation
# 调用 ANP Agent
python scripts/uni_cli.py call amap@anp maps_weather '{"city":"北京"}'
# 调用 MCP Server
python scripts/uni_cli.py call filesystem@mcp read_file '{"path":"/tmp/a.txt"}'
# 发现 Agent
python scripts/uni_cli.py discover weather
# 列出已注册 Agent
python scripts/uni_cli.py list
Python SDK
from uni_agent import UniAgent
# 初始化
agent = UniAgent(config_path="config/agents.yaml")
# 调用
result = agent.call("amap@anp", "maps_weather", {"city": "北京"})
print(result)
# 批量调用
results = agent.batch_call([
("amap@anp", "maps_weather", {"city": "北京"}),
("amap@anp", "maps_weather", {"city": "上海"}),
])
Configuration File
config/agents.yaml
agents:
# ANP Agents
- id: amap
protocol: anp
ad_url: https://agent-connect.ai/mcp/agents/amap/ad.json
- id: hotel
protocol: anp
ad_url: https://agent-connect.ai/agents/hotel-assistant/ad.json
# MCP Servers
- id: filesystem
protocol: mcp
command: npx
args: ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
- id: github
protocol: mcp
command: npx
args: ["-y", "@modelcontextprotocol/server-github"]
env:
GITHUB_TOKEN: "${GITHUB_TOKEN}"
# A2A Agents
- id: assistant
protocol: a2a
endpoint: https://example.com/.well-known/agent.json
config/identity.yaml
# 身份配置（跨协议通用）
identity:
# ANP DID 身份
anp:
did_document: config/did.json
private_key: config/private-key.pem
# A2A 认证
a2a:
auth_type: oauth2
client_id: "${A2A_CLIENT_ID}"
client_secret: "${A2A_CLIENT_SECRET}"
Directory Structure
uni-agent/
├── SKILL.md # 本文件
├── README.md # 使用文档
├── setup.sh # 一键安装
├── requirements.txt # Python 依赖
├── config/
│ ├── agents.yaml # Agent 注册表
│ ├── identity.yaml # 身份配置
│ └── .gitignore
├── adapters/
│ ├── __init__.py
│ ├── base.py # 适配器基类
│ ├── anp.py # ANP 适配器
│ ├── mcp.py # MCP 适配器
│ ├── a2a.py # A2A 适配器
│ └── aitp.py # AITP 适配器
├── scripts/
│ └── uni_cli.py # CLI 工具
└── docs/
├── architecture.md # 架构文档
└── adapters.md # 适配器开发指南
Extending with New Protocols
- Create adapter file
adapters/new_protocol.py - Inherit from
ProtocolAdapterbase class - Implement the
connect、call、discover、closemethod - In
adapters/__init__.pyregister
# adapters/new_protocol.py
from .base import ProtocolAdapter
class NewProtocolAdapter(ProtocolAdapter):
protocol_name = "new_protocol"
def connect(self, agent_config):
# 实现连接逻辑
pass
def call(self, connection, method, params):
# 实现调用逻辑
pass
# ...
Dependencies
pip install anp aiohttp mcp pyyaml