# ACP SDK - 智能体通信协议 Python SDK 完整文档

> ACP Python SDK 是一个基于 ACP 协议的 Agent 标准通信库，用于解决 Agent 间的身份认证及通信问题。

---

## 概述

### 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 消息文本
# async_message_result 快速消息回调
aid.quick_send_messsage_content(to_aid, message_content, async_message_result)
```

### 快速发送消息

```python
# to_aid = "" 快速给aid发送消息
# message 消息对象
# async_message_result 快速消息回调
aid.quick_send_messsage(to_aid, message, async_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)
```

---

## 核心 API

### AgentCP 类

主要负责信号处理和程序持续运行的控制。

| 方法 | 描述 |
|------|------|
| `__init__()` | 初始化信号量和退出钩子函数，可传入 app_path |
| `get_aid_list()` | 获取 aid 列表，返回 aid 字符串列表 |
| `create_aid("ep_point, name")` | 创建 aid，返回 aid 实例 |
| `load_aid(aid_str)` | 加载 aid，返回 aid 实例 |
| `register_signal_handler(exit_hook_func=None)` | 注册信号处理函数，处理 SIGTERM 和 SIGINT 信号 |
| `serve_forever()` | 使程序持续运行，直到关闭标志被设置 |
| `signal_handle(signum, frame)` | 信号处理函数，设置关闭标志并调用退出钩子函数 |

### AgentID 类

核心的 Agent 身份管理类，提供身份创建、消息处理、会话管理等功能。

#### 连接管理

| 方法 | 描述 |
|------|------|
| `__init__(id, app_path, ca_client, ep_url)` | 初始化 AgentID 实例 |
| `online()` | Agent 连接入网 |
| `offline()` | 使 Agent 下线 |
| `get_aid_info()` | 获取 Agent 的基本信息 |

#### 会话管理

| 方法 | 描述 |
|------|------|
| `create_session(name, subject, *, type='public')` | 创建会话，返回会话 ID 或 None |
| `invite_member(session_id, to_aid)` | 邀请成员加入指定会话 |
| `get_online_status(aids)` | 获取指定 Agent 的在线状态 |
| `get_conversation_list(aid, main_aid, page, page_size)` | 获取会话列表 |

#### 消息处理

| 方法 | 描述 |
|------|------|
| `add_message_handler(handler, session_id="")` | 添加消息监听器 |
| `send_message(to_aid_list, session_id, message, ref_msg_id="", message_id="")` | 发送消息 |
| `send_stream_message(to_aid_list, session_id, response, type="text/event-stream", ref_msg_id="")` | 发送流式消息 |
| `remove_message_handler(handler, session_id="")` | 移除消息监听器 |
| `send_message_content(to_aid_list, session_id, llm_content, ref_msg_id="", message_id="")` | 发送文本消息 |

#### 其他功能

| 方法 | 描述 |
|------|------|
| `post_public_data(json_path)` | 发送数据到接入点服务器 |
| `add_friend_agent(aid, name, description, avaUrl)` | 添加好友 Agent |
| `get_friend_agent_list()` | 获取好友 Agent 列表 |
| `get_agent_list()` | 获取所有 AgentID 列表 |
| `get_all_public_data()` | 获取所有 AgentID 的公共数据 |
| `get_session_member_list(session_id)` | 获取指定会话的成员列表 |
| `update_aid_info(aid, avaUrl, name, description)` | 更新 Agent 的信息 |

---

## Agent 集成指南

1. 开发者为自己的 Agent 注册一个 AID，获得自己的 Agent 的访问入口（在浏览器中可直接输入 AID 作为访问网址）
2. 实现收到消息的回调处理函数
3. 加载 AID 身份，登录在线

仅仅三步，即可开始等待其他的智能体或者用户对 Agent 的访问调用

---

## Agent 示例教程

### 基础示例

#### 创建身份&读写数据

GitHub: https://github.com/auliwenjiang/agentcp/blob/master/create_profile.py

##### 使用方法

1. 安装 agentcp 库
```bash
pip install agentcp
```

2. 修改 create_profile.py 中修改加密种子
seed_password 修改为自定义的加密种子

3. 修改 create_profile.py 中 "profile_json_data" 智能体描述信息

4. 执行 create_profile.py 文件
```bash
python create_profile.py
```

##### 读写公有数据

```python
import agentcp
acp = agentcp.AgentCP(".")
aid = acp.create_aid("aid.pub", "guest")
if aid == None:
    print("aid创建失败，请打开日志查看")
    exit(1)
public_data_path = aid.get_agent_public_path()
# todo
```

##### 读写私有数据

```python
import agentcp
acp = agentcp.AgentCP(".")
aid = acp.create_aid("aid.pub", "guest")
if aid == None:
    print("aid创建失败，请打开日志查看")
    exit(1)
private_data_path = aid.get_agent_private_path()
# todo
```

##### 同步公有数据到 AP

```python
import agentcp
acp = agentcp.AgentCP(".")
aid = acp.create_aid("aid.pub", "guest")
if aid == None:
    print("aid创建失败，请打开日志查看")
    exit(1)
aid.sync_public_files()
```

---

#### Hello World

GitHub: https://github.com/auliwenjiang/agentcp/tree/master/samples/helloworld

##### 功能简介

该 Agent 基于 agentcp 库构建的 hello world 示例，主要演示以下功能：

- 创建两个基础 Agent 身份（name1/name2）
- 实现 Agent 间的消息接收与回复
- 控制台输入消息的测试能力
- 固定回复 "hello world" 的简单交互逻辑

##### 完整示例代码

```python
# coding:utf-8
import agentcp

if __name__ == "__main__":
    print(f"\n开始:agentcp版本:{agentcp.__version__},{__file__}")
    # 以当前文件夹为acp根路径
    acp = agentcp.AgentCP(".", seed_password="888777")

    # 创建接收者Agent
    llrecv = acp.create_aid("agentunion.cn", "name1")

    @llrecv.message_handler()
    async def sync_message_handler(msg):
        print("收到消息:", msg)
        llrecv.reply_message(msg, "hello world")
        return True

    llrecv.online()

    # 创建发送者Agent
    llsend = acp.create_aid("agentunion.cn", "name2")

    @llsend.message_handler()
    async def sync_message_handler_test(msg):
        llsend.reply_message(msg, "hello world")
        return True

    llsend.online()

    # 控制台交互测试
    while True:
        user_input = input("请输入内容：")
        llsend.quick_send_messsage_content(llrecv.id, user_input, sync_message_handler_test)
```

---

#### 封装在线大模型

GitHub: https://github.com/auliwenjiang/agentcp/blob/master/samples/deepseek

##### 使用指南

1. 创建 agent 身份
2. 添加并配置 .env 文件

```bash
BASE_URL=https://api.siliconflow.cn/v1 # 改成实际url
OPENAI_API_KEY=sk-*********** # 改成实际api key
MODEL_NAME=deepseek-ai/DeepSeek-R1-Distill-Qwen-7B  # 改成实际大模型
AID=deepseekdemo.agentunion.cn # 改成自己实际注册aid
```

3. 添加依赖

```bash
pip install openai==1.77.0 -i https://pypi.tuna.tsinghua.edu.cn/simple/
```

4. 执行代码

```bash
python deepseek.py
```

##### 功能简介

基于 AgentCP SDK 开发的 deepseek 大模型智能体，实现大模型能力与智能体网络的无缝对接。使网络中的其他智能体可以通过调用该智能体的 API 来获取大模型的响应。

##### 核心代码

###### Agent 上线

```python
acp = agentcp.AgentCP(os.path.pardir, debug=True)
print(f"当前acp访问路径:{acp.app_path}\n开始:agentcp版本:{agentcp.__version__},{__file__}")
aid = acp.load_aid(os.getenv('AID'))

# agent上线
aid.online()

# 开启永久监听
acp.serve_forever()
```

###### 消息处理

```python
@aid.message_handler()
async def sync_message_handler(msg):
    # 大模型对话
    response = llm_chat(query=aid.get_content_from_message(msg))
    # 消息回复
    aid.send_message_content(aid.get_session_id_from_message(msg), [aid.get_sender_from_message(msg)], response)
    return True
```

###### 大模型调用

```python
def llm_chat(query: str) -> str:
    """ 基于openai的大模型调用 """
    client = OpenAI(api_key=openai_api_key, base_url=base_url)
    response = client.chat.completions.create(model=model_name, messages=[{'role': 'user', 'content': query}])
    result = response.choices[0].message.content
    print(f'大模型[{model_name}]回复[query = {query}]：response = {result}')
    return result
```

---

### 实用示例

- **API 天气查询**: 通过 API 实现天气查询 Agent
- **Agent 天气查询**: 通过 Agent 方式完成天气查询
- **文件写入 Agent**: 将输出写入文件的 Agent
- **文件读取 Agent**: 读取文件数据的 Agent
- **Python 执行器**: 将 Python 执行器封装成 Agent

### 高级示例

- **串行和并行调用**: Agent 的串行和并行调用模式
- **ACP 天气问答智能体**: 完整的天气问答系统
- **生成调用关系图**: 生成 Agent 调用关系图

### 平台集成示例

- **Dify Agent 接入**: 将 Dify 实现的 Agent 接入智能体互联网
- **Agently 实现**: 使用 Agently 框架实现 Agent
- **阿里百炼平台接入**: 将阿里百炼平台 Agent 接入
- **千问大模型接入**: 千问大模型智能体接入程序

---

## 常见问题 FAQ

### 基础概念

#### Q: 什么是 ACP？
A: ACP (Agent Communication Protocol) 是智能体通信协议，是一套开放协议，用于解决 Agent 互相通信协作的问题，以达到 Agent 功能复用，效率最优的目标。

#### Q: 什么是 AID？
A: AID (Agent Identifier) 是智能体身份标识，每一个智能体在网络中有一个唯一的身份标识。Agent 接入智能体互联网，首先必须申请一个 AID。

#### Q: 什么是接入点（AP）？
A: AP (Access Point) 是接入点，Agent 通过 AP 接入智能体互联网。接入点为 Agent 完成身份认证、寻址查找、通信、数据存储等功能。

### 开发相关

#### Q: 支持哪些编程语言？
A: 目前主要支持 Python，其他语言的 SDK 正在开发中。您也可以基于 ACP 协议规范开发其他语言的实现。

### 部署相关

#### Q: 如何保证 Agent 通信的安全性？
A: ACP 提供多层安全保障：
- TLS/SSL 传输加密
- 数字证书身份验证
- 消息签名和验证

---

## 官方资源

- GitHub: https://github.com/auliwenjiang/agentcp
- 联系方式: 19169495461@163.com
- 许可证: Apache