uni-agent

UniAgent — ユニバーサルエージェントプロトコルアダプター層

「あらゆるエージェント、あらゆるプロトコルに接続」

設計理念

課題

現在、エージェントプロトコルのエコシステムは分断されています。

• MCP：Anthropic のツール呼び出しプロトコル
• A2A：Google のエージェント間連携プロトコル
• ANP：分散型 ID ＋ エージェントネットワークプロトコル
• AITP：NEAR の対話・取引プロトコル
• ...

開発者は、プロトコルごとに異なる SDK を学び、異なる呼び出しロジックを実装する必要があります。

解決策

UniAgent は統一された抽象化レイヤーを提供し、1 つの API ですべてのプロトコルに対応します。

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})

アーキテクチャ設計

┌─────────────────────────────────────────────────────────┐
│                      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│
└──────┘ └──────┘   └──────┘   └──────┘  └──────┘

主な概念

1. エージェント ID の形式

<agent_name>@<protocol>

示例：
- amap@anp          # ANP 协议的高德地图 Agent
- filesystem@mcp   # MCP 协议的文件系统 Server
- gemini@a2a       # A2A 协议的 Gemini Agent
- shop@aitp        # AITP 协议的商店 Agent

2. 統一呼び出しインターフェース

result = agent.call(
    agent_id="amap@anp",      # Agent 标识
    method="maps_weather",     # 方法名
    params={"city": "北京"},   # 参数
    timeout=30                 # 可选超时
)

3. 機能の検出

# 发现所有能提供天气服务的 Agent
agents = agent.discover("weather")
# 返回: [
#   {"id": "amap@anp", "protocol": "anp", "methods": [...]},
#   {"id": "weather@mcp", "protocol": "mcp", "methods": [...]}
# ]

4. プロトコルアダプターインターフェース

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

対応プロトコル

プロトコル	状態	アダプター	説明
ANP	✅ 実装済み	adapters/anp.py	分散型 ID ＋ エージェントネットワーク
MCP	✅ 実装済み	adapters/mcp.py	LLM のツール呼び出し
A2A	✅ 実装済み	adapters/a2a.py	エージェント間連携
AITP	✅ 実装済み	adapters/aitp.py	対話 ＋ 取引
Agent Protocol	✅ 実装済み	adapters/agent_protocol.py	REST API
LMOS	✅ 実装済み	adapters/lmos.py	エンタープライズ向けプラットフォーム

使用方法

CLI からの呼び出し

# 调用 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": "上海"}),
])

設定ファイル

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}"

ディレクトリ構成

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          # 适配器开发指南

新しいプロトコルの拡張

1. アダプターファイル adapters/new_protocol.py を作成します
2. 基底クラス ProtocolAdapter を継承します
3. connect、call、discover、close メソッドを実装します
4. adapters/__init__.py に登録します

# 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
    
    # ...

依存関係

pip install anp aiohttp mcp pyyaml