Composio 工具集成¶

pip install composio-core

# 带框架支持
pip install composio-openai      # OpenAI
pip install composio-langchain   # LangChain
pip install composio-crewai      # CrewAI
pip install composio-autogen     # AutoGen

# CLI工具
pip install composio-cli

# 登录Composio
composio login

# 查看可用App
composio apps

# 连接GitHub(会弹出OAuth页面)
composio add github

# 连接Gmail
composio add gmail

# 查看已连接的服务
composio connections

from composio_openai import ComposioToolSet, App, Action

# 初始化工具集
toolset = ComposioToolSet(api_key="your-composio-api-key")

# 或使用环境变量
# export COMPOSIO_API_KEY="..."
toolset = ComposioToolSet()

from openai import OpenAI
from composio_openai import ComposioToolSet, App

openai_client = OpenAI()
toolset = ComposioToolSet()

# 获取GitHub相关工具
tools = toolset.get_tools(apps=[App.GITHUB])

# 让AI使用工具
response = openai_client.chat.completions.create(
    model="gpt-4o-mini",
    tools=tools,
    messages=[
        {"role": "user", "content": "在我的repo 'my-project'中创建一个issue，标题是'添加单元测试'，内容是'需要为核心模块添加pytest单元测试'"}
    ]
)

# 执行工具调用
result = toolset.handle_tool_call(response)
print(result)

from composio_openai import Action

# 只使用特定操作（更精确）
tools = toolset.get_tools(actions=[
    Action.GITHUB_CREATE_ISSUE,
    Action.GITHUB_LIST_REPOS,
    Action.GITHUB_CREATE_PR
])

response = openai_client.chat.completions.create(
    model="gpt-4o-mini",
    tools=tools,
    messages=[
        {"role": "user", "content": "列出我的GitHub仓库"}
    ]
)

result = toolset.handle_tool_call(response)

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate
from composio_langchain import ComposioToolSet, App

toolset = ComposioToolSet()
tools = toolset.get_tools(apps=[App.GITHUB, App.SLACK])

llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个项目管理助手，可以操作GitHub和Slack"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_openai_functions_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = executor.invoke({
    "input": "在Slack的#dev频道发送消息：今天的代码审查已完成"
})

from composio_openai import ComposioToolSet

# 为不同用户创建工具集
toolset = ComposioToolSet(entity_id="user-alice")
alice_tools = toolset.get_tools(apps=[App.GMAIL])

toolset = ComposioToolSet(entity_id="user-bob")
bob_tools = toolset.get_tools(apps=[App.GMAIL])

# 每个用户使用自己的OAuth连接

from composio_openai import ComposioToolSet, App

toolset = ComposioToolSet()

# 监听GitHub新issue事件
listener = toolset.create_trigger_listener()

@listener.callback(filters={"trigger_name": "github_issue_opened"})
def handle_new_issue(event):
    print(f"新Issue: {event.payload['title']}")
    # 自动用AI分析和分类issue
    # ...

listener.listen()

from composio import Action, action

@action(toolname="my_calculator")
def calculate(expression: str) -> str:
    """
    计算数学表达式

    :param expression: 数学表达式，如 "2 + 3 * 4"
    :return: 计算结果
    """
    try:
        result = eval(expression)  # 实际使用应更安全
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

# 注册自定义工具
toolset = ComposioToolSet()
tools = toolset.get_tools(actions=[calculate])

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

toolset = ComposioToolSet()

# 获取工具
github_tools = toolset.get_tools(apps=[App.GITHUB])
slack_tools = toolset.get_tools(apps=[App.SLACK])

# 创建Agent
developer = Agent(
    role="开发助手",
    goal="管理GitHub仓库和代码审查",
    tools=github_tools,
    llm="gpt-4o-mini"
)

communicator = Agent(
    role="沟通助手",
    goal="在Slack上同步项目状态",
    tools=slack_tools,
    llm="gpt-4o-mini"
)

# 创建任务
task1 = Task(
    description="检查最近的PR并总结状态",
    agent=developer
)

task2 = Task(
    description="将PR状态总结发送到Slack #dev频道",
    agent=communicator
)

crew = Crew(agents=[developer, communicator], tasks=[task1, task2])
result = crew.kickoff()

from composio_openai import ComposioToolSet

toolset = ComposioToolSet()

# 执行操作并处理错误
try:
    result = toolset.execute_action(
        action=Action.GITHUB_CREATE_ISSUE,
        params={
            "owner": "my-org",
            "repo": "my-project",
            "title": "Bug fix needed",
            "body": "详细描述..."
        },
        entity_id="user-001"
    )

    if result["success"]:
        print(f"Issue创建成功: {result['data']['url']}")
    else:
        print(f"失败: {result['error']}")

except Exception as e:
    print(f"执行错误: {e}")

# 使用tags过滤工具（减少token消耗）
tools = toolset.get_tools(
    apps=[App.GITHUB],
    tags=["issues"]  # 只获取issue相关的工具
)

# 使用use_case描述来智能选择工具
tools = toolset.get_tools(
    apps=[App.GITHUB, App.SLACK, App.GMAIL],
    use_case="我需要创建GitHub issue并通过Slack通知团队"
)

# 重新连接
composio remove github
composio add github

# 检查连接状态
composio connections
composio whoami

# 使用API Key直接连接（无需OAuth弹窗）
toolset = ComposioToolSet()
connection = toolset.initiate_connection(
    app=App.GITHUB,
    entity_id="server-bot",
    auth_config={"api_key": "ghp_xxx"}  # 使用token
)