Composio 完全指南：为 AI Agent 赋予 1000+ 工具能力的开源框架

# 使用 pip
pip install composio

# 或使用 poetry
poetry add composio

pip install composio_openai_agents openai-agents

# 使用 npm
npm install @composio/core

# 使用 yarn
yarn add @composio/core

# 使用 pnpm
pnpm add @composio/core

npm install @composio/openai-agents @openai/agents

export COMPOSIO_API_KEY="your-api-key-here"

import asyncio
from agents import Agent, Runner
from composio import Composio
from composio_openai_agents import OpenAIAgentsProvider

# 初始化 Composio 客户端
composio = Composio(provider=OpenAIAgentsProvider())

# 获取用户 ID（用于认证隔离）
user_id = "user@example.com"

# 获取 Hacker News 工具
tools = composio.tools.get(
    user_id=user_id,
    toolkits=["HACKERNEWS"]
)

# 创建 Agent
agent = Agent(
    name="HackerNews Agent",
    instructions="你是一个 Hacker News 助手，可以帮助用户了解最新的技术动态。",
    tools=tools,
)

# 运行 Agent
async def main():
    result = await Runner.run(
        starting_agent=agent,
        input="今天 Hacker News 上最热门的技术帖子是什么？",
    )
    print(result.final_output)

asyncio.run(main())

import { Composio } from '@composio/core';
import { OpenAIAgentsProvider } from '@composio/openai-agents';
import { Agent, run } from '@openai/agents';

// 初始化 SDK
const composio = new Composio({
  provider: new OpenAIAgentsProvider(),
});

const userId = 'user@example.com';

// 获取工具
const tools = await composio.tools.get(userId, {
  toolkits: ['HACKERNEWS'],
});

// 创建 Agent
const agent = new Agent({
  name: 'HackerNews assistant',
  tools: tools,
});

// 运行
const result = await run(agent, 'What is the latest hackernews post about?');
console.log(JSON.stringify(result.finalOutput, null, 2));

from composio import Composio
from composio_openai_agents import OpenAIAgentsProvider

# 初始化
composio = Composio(provider=OpenAIAgentsProvider())
user_id = "dev-team@company.com"

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

# 创建 DevOps Agent
agent = Agent(
    name="DevOps Assistant",
    instructions="""
    你是一个 DevOps 助手，可以：
    1. 查询 GitHub 仓库的 Issue 状态
    2. 创建新的 Issue 或 Pull Request
    3. 向 Slack 频道发送通知

    当发现重要的 Issue 更新时，自动通知团队。
    """,
    tools=tools,
)

async def monitor_issues():
    """监控关键 Issue 并通知团队"""
    result = await Runner.run(
        starting_agent=agent,
        input="""
        检查 composiohq/composio 仓库中标签为 'bug' 且超过 7 天未回复的 Issue，
        将列表发送到 #engineering Slack 频道。
        """,
    )
    print(result.final_output)

# 只需一行代码触发认证流程
connection = composio.connections.initiate(
    user_id=user_id,
    app_name="gmail",
)
# 用户完成 OAuth 授权后，Composio 自动管理令牌

from composio_langchain import ComposioToolSet

toolset = ComposioToolSet()
tools = toolset.get_tools(apps=["github", "slack"])

from langchain.agents import initialize_agent
agent = initialize_agent(
    tools=tools,
    llm=your_llm,
    agent_type="zero-shot-react-description"
)

from crewai import Agent, Task, Crew
from composio_crewai import ComposioToolSet

toolset = ComposioToolSet()
github_tools = toolset.get_tools(apps=["github"])

researcher = Agent(
    role='GitHub Researcher',
    goal='分析 GitHub 趋势项目',
    tools=github_tools,
    backstory='你是一位开源项目分析师',
)

from composio_autogen import ComposioToolSet

toolset = ComposioToolSet()
tools = toolset.get_tools(apps=["notion", "gmail"])

# 将工具传递给 AutoGen Agent
assistant = ConversableAgent(
    "assistant",
    llm_config={"config_list": config_list},
    function_map=tools.get_function_map(),
)

from composio import Composio
from composio_openai_agents import OpenAIAgentsProvider

# 初始化
composio = Composio(provider=OpenAIAgentsProvider())
user_id = "support-bot@company.com"

# 获取客服相关工具
tools = composio.tools.get(
    user_id=user_id,
    toolkits=["NOTION", "JIRA", "GMAIL"],
)

# 创建客服 Agent
support_agent = Agent(
    name="Customer Support Agent",
    instructions="""
    你是公司的智能客服助手。工作流程：

    1. 首先在 Notion 知识库中搜索用户问题的答案
    2. 如果找到答案，通过 Gmail 发送给用户
    3. 如果找不到答案，在 Jira 中创建支持工单，
       分配给技术支持团队，并通过 Gmail 告知用户工单号

    保持友好、专业的语气。
    """,
    tools=tools,
)

# 处理用户查询
async def handle_customer_query(customer_email: str, question: str):
    prompt = f"""
    客户邮箱：{customer_email}
    问题：{question}

    请按照工作流程处理。
    """

    result = await Runner.run(
        starting_agent=support_agent,
        input=prompt,
    )
    return result.final_output

# 同时查询多个数据源
tools = composio.tools.get(
    user_id=user_id,
    toolkits=["GITHUB", "GITLAB", "BITBUCKET"],
)
# Agent 可以并行查询三个平台的项目信息

tools = composio.tools.get(
    user_id=user_id,
    toolkits=["HACKERNEWS"],
    cache_ttl=300,  # 缓存 5 分钟
)

# 开发场景：只加载代码相关工具
dev_tools = composio.tools.get(user_id, toolkits=["GITHUB", "GITLAB"])

# 办公场景：只加载生产力工具
office_tools = composio.tools.get(user_id, toolkits=["GMAIL", "NOTION", "SLACK"])

try:
    result = await Runner.run(agent, input)
except ToolExecutionError as e:
    print(f"工具执行失败: {e}")
    # 降级策略：使用备用工具或返回友好提示