Connect Apps Plugin - Composio Claude Agents SDK

pip install composio composio-claude-agent-sdk claude-agent-sdk

import asyncio
from composio import Composio
from composio_claude_agent_sdk import ClaudeAgentSDKProvider
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, create_sdk_mcp_server

# 初始化 Composio
composio = Composio(
    api_key="YOUR_API_KEY",  # 从 Composio 获取
    provider=ClaudeAgentSDKProvider()
)

external_user_id = "unique-user-id"

# 创建工具路由会话
session = composio.create(
    user_id=external_user_id,
)

# 获取会话中的工具（原生）
tools = session.tools()
custom_server = create_sdk_mcp_server(name="composio", version="1.0.0", tools=tools)

# 使用 MCP 工具查询 Claude
async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful assistant",
        permission_mode="bypassPermissions",
        mcp_servers={
            "composio": custom_server,
        },
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Send an email to user@example.com with the subject 'Hello'")
        async for msg in client.receive_response():
            print(msg)

asyncio.run(main())

# 创建工具路由会话（禁用自动连接管理）
session = composio.create(
    user_id=external_user_id,
    manage_connections=False,
)

# 手动授权用户使用特定 API（例如 Gmail）
connection_request = session.authorize(
    toolkit='gmail',
    callback_url='https://your-app.com/callback-path'
)

# 重定向用户到 OAuth 流程
print(f"请访问以下 URL 授权应用: {connection_request.redirect_url}")

# 等待用户授权应用
connected_account = connection_request.wait_for_connection()
print(f"连接成功建立！连接账户 ID: {connected_account.id}")

await client.query("""
通过 Gmail 发送交易报告：
- 今日盈亏: +$1,250
- 胜率: 65%
- 交易次数: 8
收件人: trader@example.com
""")

await client.query("""
将最新的策略参数更新推送到 GitHub：
- 文件: config/strategy_params.yaml
- 更新: basket_tp 从 50 改为 55
- 提交信息: "调整 basket_tp 参数至 55"
""")

await client.query("""
在 #trading-alerts 频道发送风险警报：
- 账户回撤超过 15%
- 当前回撤: 17.2%
- 建议立即降仓
""")

await client.query("""
将今日交易日志上传到 Google Drive：
- 文件: logs/trading_2025-02-23.csv
- 文件夹: /trading_logs/2025/
""")

await client.query("""
在 Notion 中创建今日交易日报：
- 页面标题: 2025-02-23 交易日报
- 包含: 盈亏明细、持仓状态、明日计划
- 数据库: Trading Reports
""")

# Composio API 密钥
export COMPOSIO_API_KEY="ak_your_api_key"

# 可选：外部用户 ID（默认从环境或会话获取）
export COMPOSIO_USER_ID="your-unique-user-id"

try:
    await client.query("Send email to user@example.com")
except ConnectionError as e:
    print(f"连接未建立: {e}")
    session.authorize(toolkit='gmail')

# 检查现有连接
connections = session.list_connections()

# 删除特定连接
session.delete_connection(connection_id="conn_123")

# 为同一应用创建多个连接
gmail_personal = session.authorize(toolkit='gmail', label='personal')
gmail_work = session.authorize(toolkit='gmail', label='work')

cp skills/connect-apps-plugin/SKILL.md .trae/skills/connect-apps-plugin/SKILL.md

# 快速启动模板 - 全局外部应用连接
from composio import Composio
from composio_claude_agent_sdk import ClaudeAgentSDKProvider
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, create_sdk_mcp_server

async def run_connected_agent(prompt: str, toolkits: list = None):
    """运行带外部应用连接的 Agent"""

    # 初始化
    composio = Composio(api_key=os.getenv("COMPOSIO_API_KEY"),
                        provider=ClaudeAgentSDKProvider())
    session = composio.create(user_id="default_user")

    # 获取工具
    tools = session.tools(toolkits=toolkits)
    mcp_server = create_sdk_mcp_server(name="composio", tools=tools)

    # 运行 Agent
    options = ClaudeAgentOptions(
        system_prompt="You are a trading bot assistant with external app access.",
        permission_mode="selectivePermissions",
        mcp_servers={"composio": mcp_server},
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query(prompt)
        async for msg in client.receive_response():
            yield msg

# 使用示例
async for response in run_connected_agent(
    "Send Slack message: Trade completed with +$500 profit",
    toolkits=["slack"]
):
    print(response)