Composio
Composio - AI 代理工具执行平台
Composio 将您的 AI 代理连接到 500 多个应用程序 — Gmail、Slack、GitHub、Notion、Google Workspace、Microsoft(Outlook、Teams)、X/Twitter、Figma、网络搜索、浏览器自动化、Meta 应用、TikTok 等 — 实现无缝的跨应用自动化。
使用本指南来发现工具、管理连接并在任何受支持的应用程序中执行操作。

API 基础地址: https://backend.composio.dev/api/v3
认证:所有请求都需要x-api-key: YOUR_API_KEY
设置(一次性)
1. 获取您的 API 密钥
前往platform.composio.dev并使用 Google 或其他账户登录。导航到您的项目设置并复制您的 API 密钥。
export COMPOSIO_API_KEY="your_api_key_here"
export COMPOSIO_BASE="https://backend.composio.dev/api/v3"
就这样。您现在可以调用任何 Composio 工具了。
核心操作
每个工具都通过一个端点调用 — 工具标识符(slug)放在 URL 中:
POST /api/v3/tools/execute/{TOOL_SLUG}
curl -X POST "$COMPOSIO_BASE/tools/execute/TOOL_SLUG_HERE" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": { ... }
}'
每个请求的可选字段:
| 字段 | 描述 |
|---|---|
参数 | 工具特定的输入参数(见下方各工具) |
user_id | 您最终用户的标识符(用于多用户关联账户) |
connected_account_id | 覆盖要使用的关联账户 |
COMPOSIO_SEARCH_TOOLS — 发现工具
请始终首先调用此功能。搜索500多个应用,为您的任务找到合适的工具。返回工具模式、执行计划、连接状态和注意事项。
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_SEARCH_TOOLS" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"queries": [
{
"use_case": "send an email to someone",
"known_fields": "recipient_name: John"
},
{
"use_case": "create a meeting invite",
"known_fields": "meeting_date: tomorrow"
}
],
"session": {
"generate_id": true
}
}
}'
| 参数 | 必需 | 描述 |
|---|---|---|
queries | 是 | 搜索查询数组(1-7项) |
queries[].use_case | 是 | 任务的规范化英文描述。如已知应用名称,请包含。此处不包含个人标识符。 |
queries[].known_fields | 否 | 已知输入的逗号分隔键值对(例如,"channel_name: general, timezone: UTC") |
session.generate_id | 是(首次调用) | 设置为true以使新工作流获取会话ID |
session.id | 是(后续调用) | 复用首次调用返回的会话ID |
model | 否 | 您的LLM模型名称(例如,"claude-4.5-sonnet")用于优化规划 |
响应:
{
"data": {
"results": [
{
"index": 1,
"use_case": "send an email to someone",
"primary_tool_slugs": ["GMAIL_SEND_EMAIL"],
"related_tool_slugs": ["GMAIL_CREATE_EMAIL_DRAFT"],
"toolkits": ["gmail"],
"recommended_plan_steps": ["Step 1: ...", "Step 2: ..."],
"known_pitfalls": ["Always set user_id to 'me'"],
"reference_workbench_snippets": [...]
}
],
"tool_schemas": {
"GMAIL_SEND_EMAIL": {
"toolkit": "gmail",
"tool_slug": "GMAIL_SEND_EMAIL",
"description": "Send an email",
"input_schema": { ... }
}
},
"toolkit_connection_statuses": [
{
"toolkit": "gmail",
"has_active_connection": false,
"status_message": "No active connection. Initiate via COMPOSIO_MANAGE_CONNECTIONS."
}
],
"time_info": {
"current_time_utc": "2025-01-15T10:30:00Z",
"current_time_utc_epoch_seconds": 1736935800
},
"session": {
"id": "abcd",
"instructions": "Pass this session ID in all subsequent calls"
}
},
"successful": true
}
拆分指南:
- 1个查询 = 1个工具操作。包含隐藏的先决条件(例如,在“更新问题”之前“获取问题”)。
- 如果用户指定了应用名称,请在每个子查询中包含应用名称。
- 将非英语提示翻译为英语,同时保留标识符。
搜索后:
- 在执行前,请审阅
recommended_plan_steps及known_pitfalls。 - 如果一个工具包含
schemaRef而非input_schema,请首先调用COMPOSIO_GET_TOOL_SCHEMAS。 - 如果某个工具包的
has_active_connection状态为false,请在执行其工具前调用COMPOSIO_MANAGE_CONNECTIONS。
COMPOSIO_MANAGE_CONNECTIONS — 连接至应用程序
为工具包创建或检查 OAuth/API 连接。返回用于用户身份验证的授权链接。
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_MANAGE_CONNECTIONS" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"toolkits": ["gmail", "slack"],
"session_id": "abcd"
}
}'
| 参数 | 必需 | 说明 |
|---|---|---|
工具包 | 是 | 来自 SEARCH_TOOLS 响应的工具包名称(例如,["gmail", "github"])。切勿编造名称。 |
重新初始化所有 | 否 | 即使已处于活动状态也强制重新连接(默认:false) |
会话ID | 否 | 来自 SEARCH_TOOLS 响应的会话 ID |
连接工作流程:
{
"data": {
"message": "1 active, 1 initiated",
"results": {
"gmail": {
"toolkit": "gmail",
"status": "initiated",
"redirect_url": "https://accounts.google.com/o/oauth2/...",
"instruction": "Click the link to authenticate"
},
"slack": {
"toolkit": "slack",
"status": "active",
"has_active_connection": true,
"connected_account_id": "ca_xxx",
"current_user_info": { "email": "user@example.com" }
}
},
"summary": {
"total_toolkits": 2,
"active_connections": 1,
"initiated_connections": 1,
"failed_connections": 0
}
},
"successful": true
}
调用 SEARCH_TOOLS — 检查
- toolkit_connection_statuses
如果 - has_active_connection
为 false,则调用 MANAGE_CONNECTIONS如果状态为 - "initiated"
,则提供redirect_urlredirect_url用于用户身份验证 - 用户完成身份验证后,连接即变为活动状态
- 仅在确认连接活动后执行工具
COMPOSIO_GET_TOOL_SCHEMAS — 获取完整输入模式
检索工具的完整输入模式。当SEARCH_TOOLS返回schemaRef而非完整input_schema时调用此功能schemaRef而非完整input_schema时调用此功能
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_GET_TOOL_SCHEMAS" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"tool_slugs": ["GMAIL_SEND_EMAIL", "SLACK_SEND_MESSAGE"],
"session_id": "abcd"
}
}'
| 参数 | 必需 | 说明 |
|---|---|---|
tool_slugs | 是 | 来自SEARCH_TOOLS的工具标识数组 |
session_id | 否 | 会话ID |
响应:
{
"data": {
"success": true,
"tool_schemas": {
"GMAIL_SEND_EMAIL": {
"toolkit": "gmail",
"tool_slug": "GMAIL_SEND_EMAIL",
"description": "Send an email via Gmail",
"input_schema": {
"properties": {
"to": { "type": "string", "description": "Recipient email" },
"subject": { "type": "string" },
"body": { "type": "string" }
},
"required": ["to", "subject", "body"]
}
}
},
"not_found": []
},
"successful": true
}
COMPOSIO_MULTI_EXECUTE_TOOL — 执行工具
并行执行一个或多个工具。这是运行通过SEARCH_TOOLS发现的工具的方式
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_MULTI_EXECUTE_TOOL" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"tools": [
{
"tool_slug": "GMAIL_SEND_EMAIL",
"arguments": {
"to": "john@example.com",
"subject": "Hello",
"body": "Welcome to the team!"
}
},
{
"tool_slug": "SLACK_SEND_MESSAGE",
"arguments": {
"channel": "#general",
"text": "New member joined!"
}
}
],
"sync_response_to_workbench": false,
"session_id": "abcd"
}
}'
| 参数 | 必需 | 描述 |
|---|---|---|
工具 | 是 | 工具执行数组(1-50项) |
tools[].tool_slug | 是 | 来自SEARCH_TOOLS的有效标识符 |
tools[].arguments | 是 | 完全匹配工具输入模式的参数 |
sync_response_to_workbench | 是 | 设置true如果响应可能较大或需要用于后续脚本编写 |
session_id | 否 | 会话ID |
响应:
{
"data": {
"results": [
{
"tool_slug": "GMAIL_SEND_EMAIL",
"index": 0,
"response": {
"data": { "id": "msg_123", "threadId": "thread_456" },
"successful": true
},
"error": null
},
{
"tool_slug": "SLACK_SEND_MESSAGE",
"index": 1,
"response": {
"data": { "ok": true, "ts": "1234567890.123456" },
"successful": true
},
"error": null
}
],
"total_count": 2,
"success_count": 2,
"error_count": 0,
"remote_file_info": null
},
"successful": true
}
规则:
- 仅批量处理逻辑上独立的工具(无顺序依赖)。
- 切勿编造工具标识符或参数字段——始终使用SEARCH_TOOLS返回的内容。
- 执行前确保存在活跃连接。
- 若响应数据较大,可能会保存至远程文件——请使用
remote_file_info.file_path配合远程工作台或Bash工具进行处理。
COMPOSIO_REMOTE_WORKBENCH — 运行Python代码
在持久化的远程Jupyter沙箱中执行Python代码。适用于处理大型工具输出、批量操作以及编写多工具链脚本。
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_REMOTE_WORKBENCH" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"code_to_execute": "result, error = run_composio_tool(\"GMAIL_FETCH_EMAILS\", {\"max_results\": 5, \"user_id\": \"me\"})\nif error:\n print(\"Error:\", error)\nelse:\n emails = result.get(\"data\", {})\n print(\"Fetched:\", len(emails.get(\"messages\", [])))",
"thought": "Fetching recent emails for bulk processing",
"session_id": "abcd"
}
}'
| 参数 | 必填 | 说明 |
|---|---|---|
code_to_execute | 是 | 需在沙箱中运行的Python代码 |
thought | 否 | 单句目标描述 |
timeout | 否 | 执行超时时间(单位:秒,范围:1-780,默认值:600) |
session_id | 否 | 会话ID |
响应:
{
"data": {
"results": "Fetched: 5",
"stdout": "Fetched: 5\n",
"stderr": "",
"error": "",
"sandbox_id_suffix": "a1b2"
},
"successful": true
}
如果输出超过40,000个字符,它将被保存到文件中,并且results_file_path/stdout_file_path会被返回。
预加载的辅助函数(请勿导入或重新声明这些函数):
| 函数 | 签名 | 描述 |
|---|---|---|
run_composio_tool | (tool_slug: str, arguments: dict) -> tuple[dict, str] | 执行一个Composio应用工具。返回(response, error)。 |
invoke_llm | (query: str) -> tuple[str, str] | 调用LLM进行分析、总结、提取。最大输入200k字符。返回(response, error)。 |
proxy_execute | (method, endpoint, toolkit, query_params?, body?, headers?) -> tuple[any, str] | 当没有Composio工具存在时,直接调用已连接工具包的API。返回(response, error). |
web_search | (query: str) -> tuple[str, str] | 通过Exa AI搜索网络。返回(results, error). |
upload_local_file | (*file_paths) -> tuple[dict, str] | 将沙盒文件上传至云存储。返回({"s3_url": ...}, error). |
smart_file_extract | (sandbox_file_path: str) -> tuple[str, str] | 从PDF、图像等文件中提取文本。返回(text, error). |
所有辅助函数均返回(result, error)— 在使用结果前务必检查错误。
编码规则:
- 将工作拆分为小步骤;将中间结果保存到
/tmp/文件中。 - 状态在多次执行间持续存在(变量、导入项、文件)。
- 硬性超时限制为4分钟——对于批量操作,请使用
ThreadPoolExecutor。 - 在使用结果前,务必检查辅助函数返回的错误。
- 对于总结和分析,请使用
invoke_llm——它能比临时过滤提供更好的结果。 - 请勿通过
run_composio_tool调用COMPOSIO_*元工具——仅将其用于应用工具(例如,GMAIL_SEND_EMAIL)。
示例——批量邮件处理:
import concurrent.futures
def process_email(email_id):
result, error = run_composio_tool("GMAIL_GET_EMAIL", {
"message_id": email_id, "user_id": "me"
})
if error:
return {"id": email_id, "error": error}
subject = result.get("data", {}).get("subject", "")
return {"id": email_id, "subject": subject}
# Fetch email list
emails, err = run_composio_tool("GMAIL_FETCH_EMAILS", {
"max_results": 50, "user_id": "me"
})
if not err:
ids = [m["id"] for m in emails.get("data", {}).get("messages", [])]
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as ex:
results = list(ex.map(process_email, ids))
print(f"Processed {len(results)} emails")
COMPOSIO_REMOTE_BASH_TOOL —— 运行Bash命令
在与工作台相同的持久沙箱中执行bash命令。
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_REMOTE_BASH_TOOL" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"command": "cat /home/user/.code_out/response.json | jq \".results[] | .tool_slug\"",
"session_id": "abcd"
}
}'
| 参数 | 必需 | 描述 |
|---|---|---|
命令 | 是 | 要执行的 Bash 命令 |
会话 ID | 否 | 会话 ID |
响应:
{
"data": {
"stdout": "\"GMAIL_SEND_EMAIL\"\n\"SLACK_SEND_MESSAGE\"\n",
"stdoutLines": 2,
"stderr": "",
"stderrLines": 0,
"sandbox_id_suffix": "a1b2"
},
"successful": true
}
使用案例:
- 处理保存到远程文件的大型工具响应(通过
jq、awk、sed、grep) - 沙盒中的文件系统操作
- 默认从
/home/user运行命令 - 5分钟超时,每个流最多输出40,000个字符
典型工作流程
1. COMPOSIO_SEARCH_TOOLS
| Find tools for your task
| Check connection statuses
| Review execution plan and pitfalls
2. COMPOSIO_GET_TOOL_SCHEMAS (if needed)
| Get full input schemas for tools with schemaRef
3. COMPOSIO_MANAGE_CONNECTIONS (if needed)
| Initiate connections for toolkits without active connections
| Present auth URL to user -> user completes OAuth
4. COMPOSIO_MULTI_EXECUTE_TOOL
| Execute tools with schema-compliant arguments
| Process inline results or note remote file paths
5. COMPOSIO_REMOTE_WORKBENCH / COMPOSIO_REMOTE_BASH_TOOL (if needed)
| Process large outputs saved to remote files
| Run bulk operations or multi-tool scripts
| Upload artifacts via upload_local_file
示例——完整的“发送邮件”工作流程:
# Step 1: Search for email tools
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_SEARCH_TOOLS" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"queries": [{"use_case": "send an email via gmail", "known_fields": "recipient: john@example.com"}],
"session": {"generate_id": true}
}
}'
# -> Returns GMAIL_SEND_EMAIL tool with schema
# -> Check toolkit_connection_statuses for gmail
# -> Save session.id from response (e.g., "abcd")
# Step 2: Connect gmail (if not active)
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_MANAGE_CONNECTIONS" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"toolkits": ["gmail"],
"session_id": "abcd"
}
}'
# -> If status is "initiated", present redirect_url to user for OAuth
# Step 3: Execute the tool
curl -X POST "$COMPOSIO_BASE/tools/execute/COMPOSIO_MULTI_EXECUTE_TOOL" \
-H "x-api-key: $COMPOSIO_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"arguments": {
"tools": [{
"tool_slug": "GMAIL_SEND_EMAIL",
"arguments": {
"to": "john@example.com",
"subject": "Welcome!",
"body": "Welcome to the team, John!",
"user_id": "me"
}
}],
"sync_response_to_workbench": false,
"session_id": "abcd"
}
}'
最佳实践
会话管理
- 在首次调用SEARCH_TOOLS时传递
session: {generate_id: true}以获取会话ID。 - 在同一工作流中,将该
会话ID传递给所有后续的工具调用。 - 当用户转向不同任务时,请生成新的会话。
工具发现
- 始终首先调用SEARCH_TOOLS——切勿猜测工具标识符或参数字段。
- 当因错误或需求变化而需要额外工具时,请重新运行SEARCH_TOOLS。
- 在执行前,请审阅
recommended_plan_steps和known_pitfalls。
连接
- 切勿在没有活动连接的情况下执行工具包工具。
- 使用SEARCH_TOOLS返回的确切工具包名称——切勿自行编造名称。
- 如果连接失败,向用户展示认证URL。
执行
- 仅使用符合模式的参数——每次调用前检查input_schema。
- 将独立的工具批量处理为单个MULTI_EXECUTE调用。
- 对于大型响应,设置
sync_response_to_workbench: true并在工作台中处理。
工作台
- 工作台仅用于远程文件处理或批量脚本——不用于已内联可见的数据。
- 保持代码简洁;将长操作拆分为多个工作台调用。
- 使用
ThreadPoolExecutor在4分钟超时内实现并行处理。
快速参考
| 操作 | 工具标识 | 关键参数 |
|---|---|---|
| 发现工具 | COMPOSIO_SEARCH_TOOLS | 查询,session |
| 获取输入模式 | COMPOSIO_GET_TOOL_SCHEMAS | 工具标识符 |
| 连接到应用程序 | COMPOSIO_MANAGE_CONNECTIONS | 工具包,重新初始化全部 |
| 执行工具 | COMPOSIO_MULTI_EXECUTE_TOOL | 工具,同步响应至工作台 |
| 运行 Python 代码 | COMPOSIO_REMOTE_WORKBENCH | 待执行代码,思考 |
| 运行 bash 命令 | COMPOSIO_REMOTE_BASH_TOOL | 命令 |
所有工具的端点:
POST /api/v3/tools/execute/{TOOL_SLUG}
常见错误:
| 错误 | 含义 |
|---|---|
| 401 | API密钥无效或缺失 |
| 403 | 禁止访问 — 权限不足 |
| 404 | 工具或资源未找到 |
| 422 | 参数无效(请检查输入模式) |
| 429 | 请求频率受限 — 请等待并重试 |
| 500 | 服务器内部错误 |
频率限制:请遵循429响应,采用指数退避策略。
代理备注
测试过程中发现的陷阱和非显而易见的行为。
- 某些工具返回的是
schemaRef而非input_schema。当您看到"hasFullSchema": false时,您必须调用COMPOSIO_GET_TOOL_SCHEMAS在执行该工具之前。 - 响应嵌套:工具执行结果位于
response.data.results[].response.data— 两层data。请仔细解析。 - 工作台状态在多次调用间持续存在
。变量、导入和/tmp/ - 目录中的文件在同一沙箱会话的执行之间会保留。工作台代码中的字符串格式化
:通过JSON发送代码时,避免使用带有嵌套引号的f-strings。应使用字符串拼接来防止引号问题。invoke_llm


微信扫一扫,打赏作者吧~