CAMEL 的 ACIToolkit 实战指南:用自然语言驱动 ACI 600+ 外部应用集成
【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel
导读
ACIToolkit 是 CAMEL 框架中面向 ACI(Agent Client Interface)平台的标准工具包(Toolkit),它把 ACI 提供的 600+ 应用集成能力封装为可供ChatAgent直接调用的函数工具(FunctionTool)。本文以 ACIToolkit API 参考 为主体,结合 源码实现、单元测试 与 官方示例,系统讲解环境准备、初始化、应用发现/配置、账户链接、函数检索与执行等全部 API,并给出一个「Agent 用一句自然语言为 GitHub 仓库加 Star」的完整可运行方案。读完本文,你将掌握如何把 ACI 的工具生态无缝接入 CAMEL Agent。
1. 背景:ACIToolkit 在 CAMEL 工具体系中的定位
CAMEL 通过camel/toolkits/目录管理数十个面向具体服务的工具包(如 GitHub、Gmail、Notion、Stripe 等)。ACIToolkit 是其中面向 ACI 平台的统一入口,其核心定位是:让 CAMEL Agent 通过统一接口发现、配置、链接并执行 ACI 生态中各类第三方应用的函数,而无需为每个应用单独实现工具包。
从源码看,ACIToolkit 继承自 BaseToolkit,并在类级别声明了两个关键装饰器(见 aci_toolkit.py 头部):
@api_keys_required( [ (None, 'ACI_API_KEY'), ] ) class ACIToolkit(BaseToolkit): r"""A toolkit for interacting with the ACI API."""@api_keys_required(定义见 camel/utils/commons.py):实例化时校验ACI_API_KEY环境变量是否存在,缺失则抛出ValueError;@dependencies_required('aci')(__init__上,定义见 camel/utils/commons.py):校验 Python 侧是否已安装aciSDK,缺失则抛出ImportError。
因此,使用 ACIToolkit 的前置条件非常明确:安装aci依赖 + 配置ACI_API_KEY环境变量,二者缺一不可。测试文件中同样体现了这一点——test_aci_toolkit_init在未设置ACI_API_KEY时会被skipif跳过(见 test_aci_toolkit.py)。
另外,继承BaseToolkit还带来两个通用能力:
- 超时控制:
BaseToolkit.__init_subclass__会自动为所有可调用方法包装with_timeout,超时值通过构造参数timeout传入(base.py); - MCP 服务器:
run_mcp_server(mode)支持以stdio/sse/streamable-http模式将工具包暴露为 MCP 服务(base.py)。
2. 环境准备:安装依赖与配置密钥
2.1 安装依赖
ACIToolkit 内部通过from aci import ACI实例化客户端,因此需要安装 ACI 官方 Python SDK(CAMEL 的camel-ai[all]安装方式同样可用,见 ACI Cookbook):
pip install aci # 或安装 CAMEL 全量依赖 pip install "camel-ai[all]"2.2 配置环境变量
参考 示例代码 与 ACI Cookbook,需要设置三类密钥:
| 环境变量 | 说明 |
|---|---|
ACI_API_KEY | ACI 平台 API Key,用于身份认证,在 ACI 控制台(platform.aci.dev)申请 |
ACI_BASE_URL | ACI API 的基础 URL(可选,默认由aciSDK 决定) |
LINKED_ACCOUNT_OWNER(或LINKED_ACCOUNT_OWNER_ID) | 已链接账户的属主 ID(例如"johndoe"),用于代表终端用户执行函数 |
建议使用dotenv从.env文件加载:
import os from dotenv import load_dotenv load_dotenv() ACI_API_KEY = os.getenv("ACI_API_KEY") LINKED_ACCOUNT_OWNER = os.getenv("LINKED_ACCOUNT_OWNER")说明:
linked_account_owner_id与「账户链接」机制强相关——需要先在 ACI 控制台完成相应应用的账户授权,执行函数时才能以该属主身份操作。
3. 初始化 ACIToolkit
3.1 构造参数
根据 API 参考 与 源码,构造函数签名如下:
def __init__( self, api_key: Optional[str] = None, # ACI API Key(缺省时读取 ACI_API_KEY) base_url: Optional[str] = None, # ACI API 基础 URL(缺省时读取 ACI_BASE_URL) linked_account_owner_id: Optional[str] = None, # 链接账户属主 ID,如 "johndoe" timeout: Optional[float] = None, # 请求超时时间 ) -> None四个参数全部可空:
api_key/base_url:显式传入优先,否则回退到环境变量ACI_API_KEY/ACI_BASE_URL;linked_account_owner_id:默认None,在执行函数前建议显式指定,否则调用方需在execute_function中单独传入;timeout:透传给BaseToolkit,用于自动超时包装。
3.2 初始化行为
__init__中仅做三件事(源码 L61-L68):
from aci import ACI super().__init__(timeout) self._api_key = api_key or os.getenv("ACI_API_KEY") self._base_url = base_url or os.getenv("ACI_BASE_URL") self.client = ACI(api_key=self._api_key, base_url=self._base_url) self.linked_account_owner_id = linked_account_owner_id测试 test_aci_toolkit.py 验证了两种初始化路径:
- 默认参数下,
_api_key应等于os.getenv("ACI_API_KEY")、linked_account_owner_id为None; - 显式传入
api_key、base_url、linked_account_owner_id时,三者均被正确保存。
4. 核心 API 详解
ACIToolkit 共暴露 15 个方法(含 1 个异步变体),按职责可分为三组:应用发现与配置管理、账户链接管理、函数检索与执行。以下逐一讲解(签名与默认值均以 API 参考为准)。
4.1 应用发现与配置管理
search_tool —— 按意图搜索应用
def search_tool( self, intent: Optional[str] = None, # 意图描述,结果按与该意图的相关性排序 allowed_app_only: bool = True, # 仅返回当前 api_key 被允许访问的应用 include_functions: bool = False, # 是否在结果中附带函数名与描述 categories: Optional[List[str]] = None, # 按分类过滤,默认空列表 limit: Optional[int] = 10, # 返回结果上限 offset: Optional[int] = 0, # 分页偏移 ) -> Optional[List[AppBasic]]成功返回List[AppBasic],异常时记录日志并返回错误字符串(下同)。测试 test_search_tool 验证其内部调用client.apps.search(...),参数一一对应(注意 CAMEL 侧参数名allowed_app_only与 SDK 侧allowed_apps_only的差异)。
list_configured_apps —— 列出已配置应用
def list_configured_apps( self, app_names: Optional[List[str]] = None, # 按应用名过滤 limit: Optional[int] = 10, offset: Optional[int] = 0, ) -> Union[List[AppConfiguration], str]内部调用client.app_configurations.list(...)(源码 L136-L143)。
configure_app —— 配置应用认证方式
def configure_app(self, app_name: str) -> Union[Dict, str]这是一个自动判定认证方式的智能方法(源码 L145-L170):
app_details = self.get_app_details(app_name) if app_details and app_details.security_schemes[0] == "api_key": security_scheme = SecurityScheme.API_KEY elif app_details and app_details.security_schemes[0] == "oauth2": security_scheme = SecurityScheme.OAUTH2 else: security_scheme = SecurityScheme.NO_AUTH configuration = self.client.app_configurations.create( app_name=app_name, security_scheme=security_scheme )即:先查询应用详情,根据其首个安全方案(api_key/oauth2/ 其他)自动选择SecurityScheme.API_KEY/OAUTH2/NO_AUTH,再创建配置。测试 test_configure_app 以security_schemes = ["api_key"]的场景验证了该判定逻辑。
get_app_configuration / delete_app / get_app_details
def get_app_configuration(self, app_name: str) -> Union[AppConfiguration, str] # 查询指定应用配置 def delete_app(self, app_name: str) -> Optional[str] # 删除应用配置,成功返回 None def get_app_details(self, app_name: str) -> AppDetails # 获取应用详情(含安全方案等元数据)注意get_app_details没有try/except包装(源码 L244-L254),异常会直接抛出;delete_app成功时返回None而非消息。
4.2 账户链接管理
这部分管理「已授权账户」,是执行函数前必须完成的一步。
| 方法 | 签名 | 说明 |
|---|---|---|
link_account | link_account(app_name: str) -> Union[LinkedAccount, str] | 为已配置应用链接账户;若应用认证方案为API_KEY,会携带self._api_key调用 SDK(源码 L207-L242) |
get_linked_accounts | get_linked_accounts(app_name: str) -> Union[List[LinkedAccount], str] | 列出某应用下全部已链接账户 |
enable_linked_account | enable_linked_account(linked_account_id: str) -> Union[LinkedAccount, str] | 启用指定链接账户 |
disable_linked_account | disable_linked_account(linked_account_id: str) -> Union[LinkedAccount, str] | 禁用指定链接账户 |
delete_linked_account | delete_linked_account(linked_account_id: str) -> str | 删除链接账户,成功返回"linked_account_id: {id} deleted successfully"(源码 L330-L332) |
enable/disable/delete的返回值行为与调用参数在 test_aci_toolkit.py 中均有断言验证。
4.3 函数检索与执行
这是整个工具包的能力核心,让 Agent 能「按意图找函数 → 拿函数定义 → 执行函数」。
search_function —— 按意图搜索函数
def search_function( self, app_names: Optional[List[str]] = None, # 限定应用范围 intent: Optional[str] = None, # 搜索意图 allowed_apps_only: bool = True, # 仅返回允许访问应用中的函数 limit: Optional[int] = 10, offset: Optional[int] = 0, ) -> List[Dict]内部调用client.functions.search(...)(源码 L373-L379)。
function_definition —— 获取函数定义
def function_definition(self, func_name: str) -> Dict返回包含函数名、描述、参数 Schema(type/function/parameters)的字典,直接调用client.functions.get_definition(func_name)(源码 L346)。返回结构可参见测试中的 Mock(test_aci_toolkit.py)。
execute_function —— 执行函数调用
def execute_function( self, function_name: str, # 要执行的函数名 function_arguments: Dict, # 函数参数字典 linked_account_owner_id: str, # 终端用户(账户属主)ID,须先在 ACI 控制台链接同属主账户 allowed_apps_only: bool = False, # 仅使用 api_key 被允许的函数/应用 ) -> Dict内部调用client.handle_function_call(...)(源码 L404-L410)。测试 test_execute_function 确认四个参数原样透传给 SDK。
aexecute_function —— 异步执行函数调用
async def aexecute_function( self, function_name: str, function_arguments: Dict, linked_account_owner_id: str, allowed_apps_only: bool = False, ) -> Dict通过asyncio.to_thread把同步的handle_function_call放到线程池中执行,避免阻塞事件循环(源码 L412-L442)。测试 test_aexecute_function 使用pytest.mark.asyncio验证其异步行为。
5. get_tools():把 ACI 能力注入 ChatAgent
get_tools()是每个 Toolkit 的通用出口,返回List[FunctionTool]。ACIToolkit 的实现(源码 L444-L501)分两步:
第一步:注册 15 个管理类工具。将search_tool、list_configured_apps、configure_app、get_app_configuration、delete_app、link_account、get_app_details、get_linked_accounts、enable_linked_account、disable_linked_account、delete_linked_account、function_definition、search_function、execute_function、aexecute_function全部包装为FunctionTool。
第二步:动态注入已配置应用的真实函数。流程如下(源码 L451-L500):
_configure_app = [app.app_name for app in self.list_configured_apps() or []] _all_function = self.search_function(app_names=_configure_app) for function in _all_function: schema = self.client.functions.get_definition(function['function']['name']) def dummy_func(*, schema=schema, **kwargs): return self.execute_function( function_name=schema['function']['name'], function_arguments=kwargs, linked_account_owner_id=self.linked_account_owner_id, ) async def async_dummy_func(*, schema=schema, **kwargs): return await self.aexecute_function( function_name=schema['function']['name'], function_arguments=kwargs, linked_account_owner_id=self.linked_account_owner_id, ) dummy_func.async_call = async_dummy_func # 为同步函数附加异步入口 tool = FunctionTool(func=dummy_func, openai_tool_schema=schema) tools.append(tool)关键点:
- 闭包捕获:每个
dummy_func通过默认参数schema=schema绑定各自的函数定义,避免循环变量共享; - Schema 透传:以 ACI 返回的原始 OpenAI 兼容 Schema 直接构造
FunctionTool,LLM 据此生成参数; - 同步/异步双入口:
dummy_func.async_call = async_dummy_func让同一工具既能被同步调用也能被异步调用; - 数量可预期:
get_tools()返回 = 15 个管理工具 + 已配置应用函数数;测试 test_get_tools 在 Mock 出 1 个函数时断言结果为 16 个工具。
6. 完整实战:一句自然语言操作 GitHub
下面复现 examples/toolkits/aci_toolkit.py 的完整流程——让 CAMEL Agent 用自然语言star the repo camel-ai/camel为 GitHub 仓库加 Star。
6.1 完整代码
import os from dotenv import load_dotenv from camel.agents import ChatAgent from camel.models import ModelFactory from camel.toolkits import ACIToolkit from camel.types import ModelPlatformType, ModelType load_dotenv() LINKED_ACCOUNT_OWNER = os.getenv("LINKED_ACCOUNT_OWNER") if LINKED_ACCOUNT_OWNER is None: raise ValueError("LINKED_ACCOUNT_OWNER environment variable is not set.") # 创建 ACIToolkit(带 GitHub 应用权限) aci_toolkit = ACIToolkit(linked_account_owner_id=LINKED_ACCOUNT_OWNER) # 创建默认模型 model = ModelFactory.create( model_platform=ModelPlatformType.DEFAULT, model_type=ModelType.DEFAULT, ) # 创建 ChatAgent,并注入 ACI 工具 chat_agent = ChatAgent( model=model, tools=aci_toolkit.get_tools(), # 显式启用 GitHub 应用工具 ) # 执行自然语言指令 response = chat_agent.step("star the repo camel-ai/camel") print(response)6.2 运行链路解析
aci_toolkit.get_tools()动态加载 GitHub 相关函数(例如GITHUB__STAR_REPOSITORY);ChatAgent.step("star the repo camel-ai/camel")中,LLM 依据函数 Schema 生成工具调用;- 框架调用对应的
dummy_func,其内部通过execute_function调用 ACI 的handle_function_call; - ACI 以
linked_account_owner_id对应的已授权账户身份完成 GitHub 操作。
示例文件末尾给出了真实运行输出(examples/toolkits/aci_toolkit.py),其中tool_calls记录了:
ToolCallingRecord( tool_name='GITHUB__STAR_REPOSITORY', args={'path': {'repo': 'camel', 'owner': 'camel-ai'}}, result={'success': True, 'data': {}}, ... )Agent 最终回复:「The repositorycamel-ai/camelhas been successfully starred!」——整条链路从自然语言到真实第三方操作完全打通。
6.3 交互式查询变体
examples/usecases/aci_mcp/aci_toolkit_camel.py 提供了一个交互式版本:从环境读取LINKED_ACCOUNT_OWNER_ID,用 Gemini 模型(ModelPlatformType.GEMINI+ModelType.GEMINI_2_5_PRO)构建 Agent,支持用户输入任意查询后调用 ACI 工具并打印响应,可作为多模型场景下的参考模板。
7. 进阶:把 ACIToolkit 暴露为 MCP 服务器
由于ACIToolkit继承自BaseToolkit,它天然具备 run_mcp_server 能力,可将其工具以 MCP 协议暴露给任意 MCP 客户端:
from camel.toolkits import ACIToolkit toolkit = ACIToolkit(linked_account_owner_id="johndoe") toolkit.run_mcp_server(mode="stdio") # 或 "sse" / "streamable-http"相关 cookbook 位于 docs/cookbooks/mcp/camel_aci_mcp_cookbook.ipynb,展示了「CAMEL Toolkit 作为 MCP 服务器」的完整用法。
8. 行为契约与测试验证
ACIToolkit 的单元测试集中在 test/toolkits/test_aci_toolkit.py,从中可以总结出清晰的行为契约:
- 统一异常处理:除
get_app_details与function_definition外,绝大多数方法用try/except包裹 SDK 调用,异常时通过logger.error记录并返回错误字符串,而非抛出异常——这让工具在 Agent 循环中「失败可观测、可重试」; - 环境依赖:真实初始化测试(
test_aci_toolkit_init)要求ACI_API_KEY已设置,否则跳过; - SDK 参数透传:CAMEL 侧参数与 SDK 侧参数一一映射(如
allowed_app_only→allowed_apps_only),测试用assert_called_once_with严格校验; - 工具数量:
get_tools()= 固定 15 个管理工具 + 动态注入的已配置应用函数。
运行测试(需先设置ACI_API_KEY):
pytest test/toolkits/test_aci_toolkit.py9. 常见问题与使用建议
- 初始化报
ValueError: Missing required API key:未设置ACI_API_KEY环境变量。请先到 ACI 控制台申请 Key 并export ACI_API_KEY=...; - 初始化报
ImportError: Missing required modules: aci:未安装 ACI Python SDK,执行pip install aci; - 执行函数返回权限错误:检查是否在 ACI 控制台完成了对应应用的授权,以及
linked_account_owner_id是否与控制台中的账户属主一致;必要时把allowed_apps_only设为True只使用 api_key 明确允许的函数; - 工具数量不符合预期:
get_tools()只注入「已配置应用」的函数,先调用configure_app(app_name)并link_account(app_name)完成配置与授权,函数才会出现在工具列表中; - 需要异步场景:在异步 Agent 或高并发场景优先使用
aexecute_function,避免同步调用阻塞事件循环。
10. 总结
ACIToolkit 是 CAMEL 与 ACI 生态之间的桥梁:通过「应用搜索 → 应用配置 → 账户链接 → 函数检索 → 函数执行」五步标准化流程,把 600+ 第三方应用能力以统一的 FunctionTool 形式暴露给 LLM。其 15 个 API 覆盖了从资源发现到最终执行的全生命周期,配合get_tools()的动态注入机制,开发者只需十余行代码即可让 Agent 用自然语言操作真实世界的外部应用。
【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考