ACP SDK快速入门
Agent Communication Protocol(智能体通信协议,简称ACP)
ACP是一个开放协议，用于解决Agent互相通信协作的问题
ACP定义了agent的数据规范、agent之间的通信规范以及agent之间的授权规范
ACP Python SDK
ACP Python SDK是一个基于ACP协议的Agent标准通信库，用于解决Agent间的身份认证及通信问题。
ACP Python SDK提供了一系列API，用于创建AID、连接入网、构建会话，收发消息等。
ACP Python SDK支持多Agent协作，异步消息处理，支持内网穿透，支持Agent访问的负载均衡
功能特性
- 🔐 Agent之间采用对等网络通信
- 🛡️ 基于https安全通信及PKI体系的安全身份认证、连接管理
- 🔄 异步消息处理，支持全链路流式输入输出
- 🤖 多 Agent 协作模式支持
- 📦 简洁易用的 API 设计
- 📊 支持Agent的高并发场景
- 📚 支持多种消息类型，包括文本、图片、文件等
- 🔗 支持内网部署，全网服务
- 🌐 异构兼容：标准化通信接口支持不同架构的Agent互联
安装
bash
pip install agentcp快速入门
初始化ACP实例
python
from agentcp import AgentCP
# 创建 AgentCP 实例
# - agent_data_path: acp路径，必须外部指定,"."为当前目录
# - seed_password: 加密种子，用于私有证书加密
# - debug: 是否开启调试模式，默认为False
# - 注意：日志默认输出在控制台&当前路径下log.txt文件中
agent_data_path = "."
acp = AgentCP(agent_data_path, seed_password="123456", debug=True)创建新身份
python
# 创建新身份
# - ap: 接入点URL，指定Agent网络的接入点（如："agentunion.cn"）
# - name: Agent的身份标识，用于在该接入点上唯一标识该Agent
# - 创建身份成功，返回aid对象，创建身份失败，抛出异常，可获取失败原因
# - ps:下面两行代码将创建一个临时的aid标识,用于临时演示
# - 正式的aid标识可以在浏览器中像二级域名一样直接访问
name = "guest"
aid = acp.create_aid("agentunion.cn", name)获取身份列表
python
# 获取身份列表
list = acp.get_aid_list()加载现有身份
python
# - load_success: 加载成功返回aid对象,加载失败返回None
aid = acp.load_aid("yourname.agentunion.cn")设置消息监听器
方式1：通过装饰器方式
python
@aid.message_handler()
async def sync_message_handler(msg):
#print(f"收到消息数据: {msg}")
return True方式2：通过方法灵活设置
python
async def sync_message_handler(msg):
#print(f"收到消息数据: {msg}")
return True
aid.add_message_handler(sync_message_handler)方式3：绑定session_id和方法监听器
python
async def sync_message_handler(msg):
#print(f"收到消息数据: {msg}")
return True
aid.add_message_handler(sync_message_handler, session_id=session_id)移除消息监听器
python
async def sync_message_handler(msg):
return True
aid.remove_message_handler(sync_message_handler, session_id=session_id)连接到网络
python
# aid上线，开始监听消息
aid.online()快速回复消息
python
# msg 收到的消息dict
# message 发送的消息对象或者消息文本
aid.reply_message(msg, message)快速发送文本消息
python
# to_aid = "" 快速给aid发送消息
# message_content 消息文本
# asnyc_message_result 快速消息回调
aid.quick_send_messsage_content(to_aid, message_content, asnyc_message_result)快速发送消息
python
# to_aid = "" 快速给aid发送消息
# message 消息对象
# asnyc_message_result 快速消息回调
aid.quick_send_messsage(to_aid, message, asnyc_message_result)创建会话
python
# 创建会话
session_id = aid.create_session(
name="会话",
subject="主题"
)在会话中发送文本消息
python
# to_aid_list = [] 指定多人接收处理
# session_id 会话id
# llm_content 大模型处理结果
aid.send_message_content(to_aid_list, session_id, llm_content)在会话中发送消息
python
# 在会话中发送消息
aid.send_message(
session_id=session_id,
to_aid_list=["member1.agentunion.cn"],
message={"type": "text", "content": "你好！"}
)在会话中发送流式消息
python
# to_aid_list = [] 指定多人接收处理
# session_id 会话id
# response 大模型调用流式response
# type默认为text/event-stream
await aid.send_stream_message(to_aid_list, session_id, response, type)