结构A2A-代理到代理通信框架
一个模型无关、代理无关的消息传递系统，用于使用Redis Streams、Pub/Sub和MCP作为接口协调AI代理。
概述
结构A2A使任何AI代理都可以通过统一的消息总线发现、通信和委派任务给其他代理。基于模型上下文协议（MCP），它提供：
- 异步消息传递 -用于持久任务队列的Redis流
- 实时事件 -广播和通知的发布/订阅
- 代理注册表 -PostgreSQL支持的发现服务
- 内置工具 -20多种文件I/O、HTTP、数学、文本等工具
- ACL安全 -每个代理的细粒度权限
建筑
┌─────────────────────────────────────────────────────────────────┐
│ Agent Layer │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Percy │ │ Coder │ │ Vision │ │ Memory │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
└───────┼──────────────┼──────────────┼──────────────┼──────────┘
│ │ │ │
└──────────────┴──────────────┴──────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Fabric A2A Gateway │
│ ┌────────────────┐ ┌────────────────┐ ┌────────────────┐ │
│ │ MCP Server │ │ Message Bus │ │ Registry │ │
│ │ (HTTP/WS) │ │ (Redis Streams)│ │ (PostgreSQL) │ │
│ └────────────────┘ └────────────────┘ └────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
┌──────────────────────┼──────────────────────┐
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Redis Streams│ │ Redis Pub/Sub│ │ PostgreSQL │
│ (Task Queues)│ │ (Events) │ │ (Registry) │
└──────────────┘ └──────────────┘ └──────────────┘组件
消息总线(fabric_message_bus.py)
使用Redis Streams和Pub/Sub进行异步A2A通信：
from fabric_message_bus import FabricMessageBus, MessagePriority
bus = FabricMessageBus(redis_url="redis://localhost:6379")
# Send task to agent
await bus.send_task(
from_agent="coder",
to_agent="percy",
task_type="code_review",
payload={"pr_id": "123", "files": ["main.py"]},
priority=MessagePriority.HIGH
)
# Receive messages
messages = await bus.receive_messages("percy", count=5, block_ms=10000)
# Publish event
await bus.publish("analytics.insights", {"pattern": "unusual_traffic"})MCP服务器(server.py)
暴露MCP工具的HTTP/WebSocket服务器：
fabric.call-将任务委派给代理fabric.agent.list-列出注册代理人fabric.agent.describe-获取代理详细信息fabric.tool.list-列出可用工具fabric.tool.call-执行内置工具fabric.message.*-异步消息传递操作
代理注册表(database/postgres_registry.py)
PostgreSQL支持的代理发现：
- 代理人注册和健康监测
- 基于能力的路由
- 信任层管理
- 通话记录和指标
工具系统(tools/)
基于插件的工具基础架构：
tools/
├── base.py # BaseTool class and registry
├── builtin_tools.py # Legacy compatibility layer
└── plugins/
├── builtin_io.py # File I/O
├── builtin_web.py # HTTP requests
├── builtin_math.py # Calculator, statistics
├── builtin_text.py # Regex, transform, diff
├── builtin_system.py # Execute, env vars
├── builtin_data.py # JSON, CSV, validation
├── builtin_security.py # Hash, base64
├── builtin_encode.py # URL encoding
└── builtin_docs.py # Markdown processingRedis配置
ACL权限(config/redis/users.acl)
每个代理的访问控制：
# Percy Agent - Reasoning
user percy on >percy_secret ~agent:percy:* ~shared:* +@stream +@pubsub +@read &shared:* &agent.percy:*
# Coder Agent - Code generation
user coder on >coder_secret ~agent:coder:* ~shared:* +@stream +@pubsub +@read &shared:* &agent.coder:*
# Fabric MCP Server - Full access
user fabric_mcp on >mcp_secret ~agent:* +@stream +@pubsub +@read &*流模式
agent:{agent_id}:inbox-每个代理的任务队列shared:{topic}-共享状态密钥
发布/子频道
agent.{agent_id}.new_message-每个代理的通知shared:*-共享活动主题
快速开始
先决条件
- Python 3.11+
- Redis 7.0+
- PostgreSQL 14+
安装
cd fabric
pip install -r requirements.txt配置
# Set environment variables
export REDIS_URL="redis://localhost:6379"
export DATABASE_URL="postgresql://user:pass@localhost:5432/fabric"
export FABRIC_ADMIN_KEY="fab_admin_$(python3 -c 'import secrets; print(secrets.token_hex(32))')"跑步
# Start MCP server (HTTP)
python server.py --transport http --port 8000
# Start with stdio transport
python server.py --transport stdioAPI密钥引导
cd fabric
python3 bootstrap.pybootstrap.py 初始化 fabric_api_keys 表格并打印您的第一张 fab_sk_live_... 密钥加上开发测试密钥。
MCP工具参考
Agent通信
{
"name": "fabric.call",
"arguments": {
"agent_id": "percy",
"capability": "reason",
"task": "Analyze the pros and cons of microservices",
"context": {"domain": "software architecture"},
"stream": false,
"timeout_ms": 60000
}
}工具执行
{
"name": "fabric.tool.call",
"arguments": {
"tool_id": "io.read_file",
"capability": "read",
"parameters": {"path": "./file.txt"}
}
}异步消息传递
{
"name": "fabric.message.send",
"arguments": {
"from_agent": "coder",
"to_agent": "percy",
"message_type": "task",
"payload": {"task_type": "code_review", "pr_id": "123"}
}
}代理商注册
代理通过MCP协议注册：
curl -X POST http://localhost:8000/mcp/register_agent \
-H "Authorization: Bearer $FABRIC_ADMIN_KEY" \
-H "Content-Type: application/json" \
-d '{
"agent_id": "my-agent",
"display_name": "My Agent",
"version": "1.0.0",
"capabilities": [
{"name": "process", "description": "Process data"}
],
"endpoint": {"transport": "http", "uri": "http://localhost:9000/mcp"}
}'项目结构
fabric/
├── server.py # MCP HTTP/WS server
├── server_new.py # Updated server implementation
├── fabric_message_bus.py # Redis Streams/Pub/Sub messaging
├── config/
│ └── redis/
│ └── users.acl # Redis ACL configuration
├── database/
│ ├── models.py # SQLAlchemy models
│ └── postgres_registry.py # PostgreSQL registry backend
├── tools/
│ ├── base.py # BaseTool class
│ ├── builtin_tools.py # Tool exports
│ └── plugins/ # Tool implementations
├── sdk/
│ └── python/ # Python SDK
└── observability/ # Metrics and monitoring安全
Redis ACL
代理由ACL隔离：
- 关键模式：
~agent:{agent_id}:*-仅限自己的流 - 通道模式：
&shared:*-仅限共享主题 - 命令：
+@stream（XADD/xleadgroup），+@pubsub（发布/订阅）
认证
- PSK促进发展
- 生产代理护照（Ed25519签名）
监控
健康终点
curl http://localhost:8000/health指标
Prometheus指标可在以下网址获得 /metrics:
fabric_calls_total-代理人拨打的电话总数fabric_messages_sent-通过消息总线发送消息fabric_agent_online-在线代理商数量
许可证
麻省理工学院