Composio 集成

通过 Composio 统一 API 平台为智能体提供安全访问 1,000+ 工具包和 20,000+ 工具，支持 OAuth 处理、事件驱动工作流和多用户支持

Composio 是一个集成平台，为智能体提供对 GitHub、Slack、Notion 等流行应用程序中 1,000+ 工具包和 20,000+ 工具的安全访问。智能体可以通过 MCP 或直接 API 使用 Composio 与外部服务交互，而 Composio 负责处理认证、权限和事件驱动的工作流。

composio-langchain 包将 Composio 工具包装为 LangChain 工具，并与 LangChain 当前 create_agent API 配合使用。

概述

集成细节

类	包	可序列化	支持 JS	版本
Composio	composio-langchain	否	是

查看 Composio JS 集成文档 了解 JavaScript 和 TypeScript 的使用方法。

工具特性

• 1,000+ 工具包和 20,000+ 工具：为 GitHub、Slack、Gmail、Jira、Notion 等提供预构建集成，可通过 MCP 或直接 API 访问。
• 认证管理：处理 OAuth 流程、API 密钥和认证状态。
• 事件驱动工作流：根据外部事件（如新的 Slack 消息或 GitHub 提交）触发智能体。
• 细粒度权限：按用户控制工具访问和数据暴露。
• 企业控制：支持治理、可审计性、SSO、组织级控制，以及企业安全审查所需的 SOC 2 / ISO 27001:2022 认证。
• 自定义工具支持：添加专有 API 和内部工具。

设置

集成位于 composio-langchain 包中。

pip

pip install -U composio-langchain langchain langchain-openai

凭证

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

import getpass
import os

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

建议设置 LangSmith 进行跟踪：

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

实例化

使用 LangChain 提供程序初始化 Composio。为智能体要访问的用户和工具包集创建会话：

from composio import Composio
from composio_langchain import LangchainProvider

composio = Composio(provider=LangchainProvider())
session = composio.create(user_id="user_123", toolkits=["GITHUB"])
tools = session.tools()

session.tools() 返回 Composio 路由工具，如 COMPOSIO_SEARCH_TOOLS、COMPOSIO_GET_TOOL_SCHEMAS 和 COMPOSIO_MANAGE_CONNECTIONS。智能体使用这些工具发现具体的工具包工具、检查模式、管理认证和执行操作。

可用工具包

Composio 为许多服务提供工具包：

类别	示例工具包
生产力	GitHub, Slack, Gmail, Jira, Notion, Asana, Trello, ClickUp
通信	Discord, Telegram, WhatsApp, Microsoft Teams
开发	GitLab, Bitbucket, Linear, Sentry
数据和分析	Google Sheets, Airtable, HubSpot, Salesforce

查看 Composio 工具包目录 获取完整列表。

调用

发现工具

使用 session.search(...) 在请求智能体执行任何操作之前检查匹配的具体工具包工具和连接状态：

search = session.search(query="get authenticated github user")

for result in search.results:
    print(result.primary_tool_slugs)

for status in search.toolkit_connection_statuses:
    print(status.toolkit, status.has_active_connection, status.status_message)

如果工具包未连接，Composio 将引导智能体使用 COMPOSIO_MANAGE_CONNECTIONS，或者您可以像 认证设置 中所示的那样创建授权链接。

低级工具加载

对于大多数智能体工作流，请优先使用 composio.create(...) 和 session.tools()。当您想自己加载直接工具列表时，也可以使用低级的 composio.tools.get(...) API：

github_tools = composio.tools.get(
    user_id="user_123",
    toolkits=["GITHUB"],
    limit=10,
)

read_scoped_tools = composio.tools.get(
    user_id="user_123",
    toolkits=["GITHUB"],
    scopes=["read"],
    limit=10,
)

specific_tool = composio.tools.get(
    user_id="user_123",
    tools=["GITHUB_GET_THE_AUTHENTICATED_USER"],
)

在智能体中使用

这是一个使用 Composio 工具和 LangChain 智能体的完整、只读示例。它要求模型检查可用的 GitHub 个人资料读取工具，而不会修改外部状态。

import getpass
import os

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

from composio import Composio
from composio_langchain import LangchainProvider
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

composio = Composio(provider=LangchainProvider())
session = composio.create(user_id="user_123", toolkits=["GITHUB"])
tools = session.tools()

model = ChatOpenAI(model="gpt-5.4-mini")
agent = create_agent(model=model, tools=tools)

result = agent.invoke(
    {
        "messages": [
            (
                "user",
                "List available Composio tool names for reading GitHub user "
                "profile data. Do not execute external account actions, do not "
                "ask me to authenticate, and do not mutate anything.",
            )
        ]
    }
)

print(result["messages"][-1].content)

认证设置

访问用户外部帐户的工具需要用户连接相应的工具包。

对于手动认证，从会话创建连接请求：

connection_request = session.authorize("github")
print(connection_request.redirect_url)

用户完成授权流程后，相同的 user_id 可以使用该已连接帐户。

对于聊天认证，允许智能体在工具包没有活动连接时调用 COMPOSIO_MANAGE_CONNECTIONS。然后，智能体可以引导用户连接所需的帐户，然后再执行工具包操作。

多用户场景

为每个应用程序用户使用稳定的 user_id。每个用户连接自己的帐户，Composio 使用与该 user_id 关联的凭证执行工具包操作。

session_alice = composio.create(user_id="alice", toolkits=["GITHUB"])
session_bob = composio.create(user_id="bob", toolkits=["GITHUB"])

alice_agent = create_agent(model=model, tools=session_alice.tools())
bob_agent = create_agent(model=model, tools=session_bob.tools())

事件驱动工作流

Composio 支持根据外部事件触发智能体。当连接的应用程序中发生事件（如新的 GitHub 提交或 Slack 消息）时，触发器可以将结构化有效负载发送到您的应用程序。

检查触发器配置

在创建触发器之前，检查所需的配置：

from composio import Composio

composio = Composio()

trigger_type = composio.triggers.get_type("GITHUB_COMMIT_EVENT")
print(trigger_type.config)

创建触发器

在相关帐户连接并您拥有所需配置后创建触发器：

trigger = composio.triggers.create(
    slug="GITHUB_COMMIT_EVENT",
    user_id="user_123",
    trigger_config={
        "owner": "composiohq",
        "repo": "composio",
    },
)

print(trigger.trigger_id)

Webhooks

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

from fastapi import FastAPI, Request

app = FastAPI()


@app.post("/webhook")
async def webhook_handler(request: Request):
    payload = await request.json()

    if payload.get("triggerSlug") == "GITHUB_COMMIT_EVENT":
        commit_data = payload.get("payload")
        # Invoke your agent with commit_data here.

    return {"status": "success"}

更多详情，请参阅 Composio 触发器文档。

自定义工具

Composio 允许您创建自定义工具，这些工具可以与内置工具一起使用。

from pydantic import BaseModel, Field


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


@composio.tools.custom_tool
def add_two_numbers(request: AddTwoNumbersInput) -> int:
    """Add two numbers."""
    return request.a + request.b


tools = session.tools()
tools.append(add_two_numbers)

API 参考

有关 Composio 功能和配置选项的详细文档，请参阅：

• Composio 文档
• Composio LangChain 提供程序指南
• 可用工具和操作

---

官方原文：src/oss/python/integrations/tools/composio.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。