Composio 集成

Composio 是一个集成平台，提供对 500 多种工具的访问，涵盖 GitHub、Slack、Notion 等流行应用程序。它使 AI 代理能够通过统一的 API 与外部服务交互，处理身份验证、权限和事件驱动的工作流。

概述

集成详情

类	包	可序列化	JS 支持	版本
Composio	`composio-langchain`	❌	✅

工具特性

500+ 工具访问：为 GitHub、Slack、Gmail、Jira、Notion 等提供预构建集成
认证管理：处理 OAuth 流程、API 密钥和认证状态
事件驱动工作流：基于外部事件（如新的 Slack 消息、GitHub 问题等）触发代理
细粒度权限：按用户控制工具访问和数据暴露
自定义工具支持：添加专有 API 和内部工具

设置

集成位于 composio-langchain 包中。

pip install -U composio-langchain

凭证

您需要一个 Composio API 密钥。在 composio.dev 免费注册以获取您的 API 密钥。

设置 API 密钥

import getpass
import os

if not os.environ.get("COMPOSIO_API_KEY"):
    os.environ["COMPOSIO_API_KEY"] = getpass.getpass("Enter your Composio API key: ")

设置 LangSmith 以进行跟踪也很有帮助：

启用跟踪

# os.environ["LANGSMITH_API_KEY"] = getpass.getpass("Enter your LangSmith API key: ")
# os.environ["LANGSMITH_TRACING"] = "true"

实例化

使用 LangChain 提供程序初始化 Composio，并从特定工具包获取工具。每个工具包代表一个服务（例如 GitHub、Slack），包含多个工具（您可以执行的操作）。

初始化 Composio

from composio import Composio
from composio_langchain import LangchainProvider

# 使用 LangChain 提供程序初始化 Composio
composio = Composio(provider=LangchainProvider())

# 从特定工具包获取工具
# 您可以指定一个或多个工具包
tools = composio.tools.get(
    user_id="default",
    toolkits=["GITHUB"]
)

print(f"Loaded {len(tools)} tools from GitHub toolkit")

可用工具包

Composio 为各种服务提供工具包： 生产力：GitHub、Slack、Gmail、Jira、Notion、Asana、Trello、ClickUp 通信：Discord、Telegram、WhatsApp、Microsoft Teams 开发：GitLab、Bitbucket、Linear、Sentry 数据与分析：Google Sheets、Airtable、HubSpot、Salesforce 以及 100 多种更多…

调用

从多个工具包获取工具

您可以一次加载来自多个服务的工具：

# 从多个工具包获取工具
tools = composio.tools.get(
    user_id="default",
    toolkits=["GITHUB", "SLACK", "GMAIL"]
)

获取特定工具

您可以加载特定工具，而不是整个工具包：

# 按名称获取特定工具
tools = composio.tools.get(
    user_id="default",
    tools=["GITHUB_CREATE_ISSUE", "SLACK_SEND_MESSAGE"]
)

用户特定工具

Composio 支持多用户场景，具有用户特定的身份验证：

# 获取特定用户的工具
# 此用户必须首先验证其账户
tools = composio.tools.get(
    user_id="user_123",
    toolkits=["GITHUB"]
)

在代理中使用

以下是一个完整示例，使用 Composio 工具与 LangChain 代理交互 GitHub：

import os
import getpass

if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

# | output: false
# | echo: false

from langchain.chat_models import init_chat_model

llm = init_chat_model(model="gpt-5", model_provider="openai")

使用 Composio 工具的代理

from composio import Composio
from composio_langchain import LangchainProvider
from langchain import hub
from langchain.agents import AgentExecutor, create_openai_functions_agent

# 拉取提示模板
prompt = hub.pull("hwchase17/openai-functions-agent")

# 初始化 Composio
composio = Composio(provider=LangchainProvider())

# 获取 GitHub 工具
tools = composio.tools.get(user_id="default", toolkits=["GITHUB"])

# 定义任务
task = "Star a repo composiohq/composio on GitHub"

# 创建代理
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 使用 agent_executor 执行
agent_executor.invoke({"input": task})

事件驱动工作流

Composio 支持基于外部事件触发代理。当连接的应用程序中发生事件（如新的 GitHub 提交或 Slack 消息）时，触发器会自动将结构化有效负载发送到您的应用程序。

创建触发器

首先，为您要监控的事件创建触发器：

from composio import Composio

composio = Composio(api_key="your_api_key")
user_id = "user_123"

# 检查触发器所需的配置
trigger_type = composio.triggers.get_type("GITHUB_COMMIT_EVENT")
print(trigger_type.config)

# 使用所需配置创建触发器
trigger = composio.triggers.create(
    slug="GITHUB_COMMIT_EVENT",
    user_id=user_id,
    trigger_config={
        "owner": "composiohq",
        "repo": "composio"
    }
)

print(f"Trigger created: {trigger.trigger_id}")

订阅触发器（开发）

对于本地开发和原型设计，您可以直接订阅触发器：

from composio import Composio

composio = Composio(api_key="your_api_key")

# 订阅触发器事件
subscription = composio.triggers.subscribe()

# 定义事件处理程序
@subscription.handle(trigger_id="your_trigger_id")
def handle_github_commit(data):
    print(f"New commit detected: {data}")
    # 使用您的代理处理事件
    # ... 使用任务调用您的代理

# 注意：对于生产环境，请改用 Webhook

Webhook（生产）

对于生产环境，在 Composio 仪表板中配置 Webhook：

from fastapi import FastAPI, Request
import json

app = FastAPI()

@app.post("/webhook")
async def webhook_handler(request: Request):
    # 获取 Webhook 有效负载
    payload = await request.json()

    print("Received trigger event:")
    print(json.dumps(payload, indent=2))

    # 使用您的代理处理事件
    if payload.get("triggerSlug") == "GITHUB_COMMIT_EVENT":
        commit_data = payload.get("payload")
        # ... 使用 commit_data 调用您的代理

    return {"status": "success"}

更多详情，请参阅 Composio Triggers 文档

身份验证设置

在使用需要身份验证的工具之前，用户需要连接他们的账户：

from composio import Composio

composio = Composio()

# 获取用户的认证 URL
auth_connection = composio.integrations.create(
    user_id="user_123",
    integration="github"
)

print(f"Authenticate at: {auth_connection.redirect_url}")

# 认证后，用户的连接账户将可用
# 工具将使用其凭据工作

多用户场景

对于具有多个用户的应用程序：

# 每个用户验证自己的账户
tools_user_1 = composio.tools.get(user_id="user_1", toolkits=["GITHUB"])
tools_user_2 = composio.tools.get(user_id="user_2", toolkits=["GITHUB"])

# 工具将使用相应用户的凭据
# 用户 1 的代理将在用户 1 的 GitHub 账户上操作
agent_1 = create_agent(llm, tools_user_1)

# 用户 2 的代理将在用户 2 的 GitHub 账户上操作
agent_2 = create_agent(llm, tools_user_2)

高级特性

自定义工具

Composio 允许您创建可与内置工具一起使用的自定义工具。有两种类型：

独立工具

不需要身份验证的简单工具：

from pydantic import BaseModel, Field
from composio import Composio

composio = Composio()

class AddTwoNumbersInput(BaseModel):
    a: int = Field(description="The first number to add")
    b: int = Field(description="The second number to add")

# 函数名称将用作工具 slug
@composio.tools.custom_tool
def add_two_numbers(request: AddTwoNumbersInput) -> int:
    """Add two numbers."""
    return request.a + request.b

# 与您的代理一起使用
tools = composio.tools.get(user_id="default", toolkits=["GITHUB"])
tools.append(add_two_numbers)

基于工具包的工具

需要身份验证并可以使用工具包凭据的工具：

from composio.types import ExecuteRequestFn

class GetIssueInfoInput(BaseModel):
    issue_number: int = Field(
        ..., description="The number of the issue to get information about"
    )

@composio.tools.custom_tool(toolkit="github")
def get_issue_info(
    request: GetIssueInfoInput,
    execute_request: ExecuteRequestFn,
    auth_credentials: dict,
) -> dict:
    """Get information about a GitHub issue."""
    response = execute_request(
        endpoint=f"/repos/composiohq/composio/issues/{request.issue_number}",
        method="GET",
        parameters=[
            {
                "name": "Accept",
                "value": "application/vnd.github.v3+json",
                "type": "header",
            },
            {
                "name": "Authorization",
                "value": f"Bearer {auth_credentials['access_token']}",
                "type": "header",
            },
        ],
    )
    return {"data": response.data}

执行自定义工具：

response = composio.tools.execute(
    user_id="default",
    slug="get_issue_info",  # 使用函数名称作为 slug
    arguments={"issue_number": 1},
)

更多详情，请参阅 Composio Custom Tools 文档

细粒度权限

控制工具可以执行的操作：

# 获取具有特定权限的工具
tools = composio.tools.get(
    user_id="default",
    toolkits=["GITHUB"],
    # 限制为只读操作
    permissions=["read"]
)

API 参考

有关所有 Composio 特性和配置的详细文档，请访问：

在 GitHub 上编辑此页面或提交问题。

通过 MCP 连接这些文档到 Claude、VSCode 等，以获取实时答案。

Popular Providers

Integrations by component

概述

集成详情

工具特性

设置

凭证

实例化

可用工具包

调用

从多个工具包获取工具

获取特定工具

用户特定工具

在代理中使用

事件驱动工作流

创建触发器

订阅触发器（开发）

Webhook（生产）

身份验证设置

多用户场景

高级特性

自定义工具

独立工具

基于工具包的工具

细粒度权限

API 参考

Popular Providers

Integrations by component

​概述

​集成详情

​工具特性

​设置

​凭证

​实例化

​可用工具包

​调用

​从多个工具包获取工具

​获取特定工具

​用户特定工具

​在代理中使用

​事件驱动工作流

​创建触发器

​订阅触发器（开发）

​Webhook（生产）

​身份验证设置

​多用户场景

​高级特性

​自定义工具

​独立工具

​基于工具包的工具

​细粒度权限

​API 参考

概述

集成详情

工具特性

设置

凭证

实例化

可用工具包

调用

从多个工具包获取工具

获取特定工具

用户特定工具

在代理中使用

事件驱动工作流

创建触发器

订阅触发器（开发）

Webhook（生产）

身份验证设置

多用户场景

高级特性

自定义工具

独立工具

基于工具包的工具

细粒度权限

API 参考