企业官网建设流程全解析

Composio Python SDK 示例指南：从环境配置到工具路由器实战

【免费下载链接】composioComposio powers 1000+ toolkits, tool search, context management, authentication, and a sandboxed workbench to help you build AI agents that turn intent into action.项目地址: https://gitcode.com/GitHub_Trending/co/composio

Composio 的 Python 示例仓库（python/examples）是理解官方 SDK 各核心 API 的最直接入口。本文以 python/examples/README.md 为主线，系统讲解如何运行这些示例、如何通过一次性配置脚本准备认证资源，并逐一对工具执行、认证配置、连接账户、触发器、MCP 与工具路由器等示例做源码级拆解。读完本文，你将掌握 Composio Python SDK 的典型调用模式，以及如何在自己的项目中搭建可重复、可清理的示例运行环境。

示例目录概览

示例脚本大部分直接位于python/examples/目录下，按功能覆盖 SDK 的几大核心能力：

工具路由器（Tool Router）相关脚本则集中放在python/examples/tool_router/子目录中，包括 tools.py、preload.py、claude_agent.py、crewai_agent.py、langchain_agent.py、openai_agents.py、tool_router_mcp.py、session_update.py、authorize.py、files.py 等。

运行单个示例

标准运行流程

仓库使用uv管理 Python 工作区，README 给出的标准三步流程如下：

cd python uv sync out=$(node ../scripts/examples-provision.mjs) && eval "$out" uv run python examples/tools.py

• uv sync：同步工作区虚拟环境，安装python/composio及所有 provider 依赖；
• node ../scripts/examples-provision.mjs：对 Composio 项目做一次性"供给检查"（provisioning check），在 stdout 打印一批export COMPOSIO_EXAMPLES_*=...环境变量，eval "$out"把它们加载进当前 shell；
• uv run python examples/<file>.py：在虚拟环境中直接运行脚本，无需先激活环境。

uv run对子目录同样适用，例如uv run python examples/tool_router/tools.py。

两种 eval 写法为何不同

README 特别强调：必须先捕获再 eval（out=$(...) && eval "$out"），不要写成eval "$(...)"。原因在于eval的退出状态反映的是被求值文本的执行结果，而不是供给脚本本身的结果——如果供给运行失败，eval "$(...)"仍会"看起来成功"，从而掩盖失败。先捕获到变量、再 eval，供给脚本的退出码（非零即失败）会被&&正确传递。

供给脚本的核心逻辑见 scripts/examples-provision.mjs：它会依次调用GET /api/v3.1/auth_configs与GET /api/v3.1/connected_accounts（按用户过滤），将"报告"打印到 stderr，将可 eval 的export行打印到 stdout。该脚本是幂等的——已有资源会被验证通过，只有缺失的资源才会被创建，且永远不会打印任何凭据值。

供给脚本依赖的环境变量

供给脚本本身需要两个输入（见 scripts/examples-provision.mjs）：

• COMPOSIO_API_KEY——必需，缺失时脚本直接报错退出；
• COMPOSIO_EXAMPLES_USER_ID——可选，默认值为examples。

后端地址通过COMPOSIO_BASE_URL选择，默认指向 staging 环境，解析逻辑在 harness/backend-url.mjs。

需要人工参与的 OAuth 授权

示例所依赖的 OAuth 类工具包（gmail、googledrive、github、slack）需要一次性浏览器授权。当某工具包尚无活跃连接时，可加--initiate-missing让供给脚本代为发起连接请求：

node ../scripts/examples-provision.mjs --initiate-missing

脚本会为每个缺失连接调用POST /api/v3.1/connected_accounts，并把授权 URL 打印出来，只需在浏览器中访问一次即可。serpapi 属于 API-Key 认证，供给脚本会自动创建其认证配置，无需浏览器步骤。供给脚本会为 OAuth 工具包创建名为examples-<slug>的use_composio_managed_auth认证配置，为 serpapi 创建use_custom_auth/API_KEY方案配置，演示键值为刻意伪造的占位符examples-demo-key（见 scripts/examples-provision.mjs），并非真实密钥。

示例的环境变量约定

示例脚本统一从环境变量读取配置，缺失时抛出KeyError并指明缺失变量名。完整约定如下：

例如 tools.py 通过os.environ["COMPOSIO_EXAMPLES_GITHUB_CONNECTED_ACCOUNT_ID"]读取 GitHub 连接账户 id；connected_accounts.py 读取 Gmail 认证配置 id 与用户 id；fastapi_app.py 在启动时校验两个必需变量，并提前调用composio.auth_configs.get(...)做"快速失败"，避免带着失效的认证配置对外服务。

供给脚本的清理模式（--gc）

示例运行会在项目中遗留资源，可用--gc清理：

node ../scripts/examples-provision.mjs --gc --dry-run # 先预览 node ../scripts/examples-provision.mjs --gc # 真正删除

--gc会删除三类由示例创建的资源：

1. 从未达到ACTIVE状态的连接账户（OAuth 发起后未完成授权的死连接）；
2. 多余的 serpapi 演示账户（保留每个工具包最新一个活跃账户）；
3. 之前运行遗留的 MCP 配置——MCP 配置统一命名为examples-<label>-<unix-seconds>格式（如examples-lc-slack-1754923456），脚本用正则^examples-[a-z0-9-]+-\d{10}$精确匹配（见 scripts/examples-provision.mjs），既靠examples-前缀证明是示例创建，又靠 10 位秒级时间戳避免误伤手工创建的examples-scratch之类配置；API 对 MCP 名称有 30 字符上限，所以标签必须保持简短。

清理有几道安全防线：只删除 24 小时前创建的资源（避免误删并发运行中的资源）、只触碰示例自己创建的资源（连接账户必须绑定到examples-<slug>认证配置，任何项目自有的 gmail 配置都不会被误认为示例所有）、created_at缺失或不可解析时按"不够旧"处理（fail closed）。脚本文件头注释明确警告这是破坏性操作，应先在专用的一次性示例项目上配合--dry-run预览。

工具与工具包示例

工具查询与直接执行

tools.py 演示了工具发现与执行的核心调用：

from composio import Composio composio = Composio() # 按工具包获取全部工具 tools = composio.tools.get(user_id="default", toolkits=["GITHUB"]) # 按关键字搜索工具 tools = composio.tools.get(user_id="default", search="user") # 组合：工具包 + 搜索 tools = composio.tools.get(user_id="default", toolkits=["GITHUB"], search="star") # 直接执行工具 response = composio.tools.execute( user_id="default", slug="HACKERNEWS_GET_USER", arguments={"username": "pg"}, version="20251023_00", ) print(response)

其中composio.tools.execute是同步直接执行：传入工具 slug、参数与可选的version（工具版本号，示例中为20251023_00）。工具支持按工具包、按搜索词、或两者组合三种方式获取，user_id决定工具以哪个用户的上下文执行。

代理调用（Proxy）

同一文件中还演示了composio.tools.proxy——直接向第三方 API 发起代理请求，并借用已连接的账户身份：

proxy_response = composio.tools.proxy( endpoint="/repos/composiohq/composio/issues/1", method="GET", connected_account_id=github_connected_account_id, # 借用 GitHub 连接 parameters=[ {"name": "Accept", "value": "application/vnd.github.v3+json", "type": "header"}, ], ) print(proxy_response)

这里connected_account_id来自环境变量（COMPOSIO_EXAMPLES_GITHUB_CONNECTED_ACCOUNT_ID），代理调用会把请求以该账户的 GitHub 身份发出，parameters可附加自定义请求头。

工具包查询与授权

toolkits.py 展示工具包 API：

toolkits = composio.toolkits.get() # 列出全部工具包 toolkit = composio.toolkits.get(slug="github") # 按 slug 查询 categories = composio.toolkits.list_categories() # 列出工具包分类 connection_request = composio.toolkits.authorize(user_id="123", toolkit="github")

composio.toolkits.authorize是面向用户发起授权的一行式封装，返回的连接请求对象可进一步用于获取授权 URL 或等待连接建立。

认证配置示例（auth_configs.py）

auth_configs.py 覆盖认证配置的完整生命周期，包括 Composio 托管认证（managed auth）与自定义认证（custom auth）两种模式：

from composio import Composio composio = Composio() # 列出全部认证配置 auth_configs = composio.auth_configs.list() # 托管认证：由 Composio 代为管理凭据 auth_config = composio.auth_configs.create( toolkit="github", options={"type": "use_composio_managed_auth"}, ) # 自定义认证：自带 OAuth2 应用凭据 auth_config = composio.auth_configs.create( toolkit="notion", options={ "name": "Notion Auth", "type": "use_custom_auth", "auth_scheme": "OAUTH2", "credentials": { "client_id": "1234567890", "client_secret": "1234567890", "oauth_redirect_uri": "https://backend.composio.dev/api/v1/auth-apps/add", }, }, ) # 创建前查询必填字段 required_fields = composio.toolkits.get_auth_config_creation_fields( toolkit="NOTION", auth_scheme="OAUTH2", ) # 按 id 获取 auth_config_retrieved = composio.auth_configs.get(auth_config.id) # 从工具包元数据读取认证输入字段 toolkit = composio.toolkits.get(slug="notion") required_input_fields = [ field.name for field in toolkit.auth_config_details or [] if field.mode ] # 更新认证配置（可修改凭据与工具访问策略） auth_config_updated = composio.auth_configs.update( auth_config.id, options={ "type": "custom", "credentials": {"client_id": "1234567890", "client_secret": "1234567890"}, "tool_access_config": { "tools_for_connected_account_creation": ["NOTION_ADD_PAGE_CONTENT"], }, }, ) # 启用 / 禁用 / 删除 composio.auth_configs.enable(auth_config.id) composio.auth_configs.disable(auth_config.id) composio.auth_configs.delete(auth_config.id)

几点值得注意的细节：

• auth_scheme在 Python 端以小写 snake_case 传递（OAUTH2、API_KEY），供给脚本调用 API 时则以authScheme形式出现（见 scripts/examples-provision.mjs）；
• tool_access_config.tools_for_connected_account_creation可限定哪些工具允许在创建连接账户时使用，用于按需暴露工具面；
• 示例中的client_id/client_secret均为演示占位值，实际使用需替换为真实应用凭据。

连接账户示例（connected_accounts.py）

connected_accounts.py 演示连接的创建与全生命周期管理，区分OAuth 连接与API-Key 连接两种路径：

from composio import Composio from composio.types import auth_scheme composio = Composio() gmail_auth_config_id = os.environ["COMPOSIO_EXAMPLES_GMAIL_AUTH_CONFIG_ID"] user_id = os.environ["COMPOSIO_EXAMPLES_USER_ID"] # 列出全部连接 connected_accounts = composio.connected_accounts.list() # OAuth 连接：发起请求 -> 用户浏览器授权 -> 等待激活 connection_request = composio.connected_accounts.initiate( user_id=user_id, auth_config_id=gmail_auth_config_id, allow_multiple=True, # 允许同一认证配置下存在多个连接 ) print(f"Visit this URL to authorize: {connection_request.redirect_url}") connected_account = connection_request.wait_for_connection() print(f"Connected account {connected_account.id} is {connected_account.status}") # API-Key 连接：无需浏览器，直接携带 key 发起 connection_request = composio.connected_accounts.initiate( user_id=user_id, auth_config_id=os.environ["COMPOSIO_EXAMPLES_APIKEY_AUTH_CONFIG_ID"], allow_multiple=True, config=auth_scheme.api_key( options={"generic_api_key": os.environ["COMPOSIO_EXAMPLES_APIKEY_PLACEHOLDER"]}, ), ) # 发起前查询必填字段 required_fields = composio.toolkits.get_connected_account_initiation_fields( toolkit="NOTION", auth_scheme="API_KEY", ) # 获取 / 禁用 / 启用 / 删除 connected_account_retrieved = composio.connected_accounts.get(connected_account.id) composio.connected_accounts.disable(connected_account.id) composio.connected_accounts.enable(connected_account.id) composio.connected_accounts.delete(connected_account.id)

连接示例为何不能自动跑完

README 特别说明了连接示例的特殊性：connected_accounts.py演示的是创建连接的完整过程——调用initiate()打印授权 URL、调用wait_for_connection()等待账户激活或超时放弃。这个等待必须有真人打开 URL 完成授权，因此该示例无法独立跑完，自动化运行会跳过它。相比之下，直接使用连接（通过 Gmail/GitHub/Slack 调用工具）的示例，依赖的是供给脚本已验证为活跃的常驻连接账户，可以全自动运行。

这种"创建连接需要人工、使用连接可自动化"的区分，是设计示例时对交互成本与可重复性的平衡，也解释了为什么供给脚本要维护一组常驻 ACTIVE 连接。

触发器示例（triggers.py）

triggers.py 覆盖触发器（trigger）与 Webhook 验签两大能力：

triggers = composio.triggers.list() # 全部触发器 active_triggers = composio.triggers.list_active() trigger_enums = composio.triggers.list_enum() trigger = composio.triggers.get_type(slug="GMAIL_NEW_GMAIL_MESSAGE") # 创建触发器实例：绑定连接账户 instance = composio.triggers.create( slug="GMAIL_NEW_GMAIL_MESSAGE", connected_account_id=gmail_connected_account_id, trigger_config={}, ) # 或只传 user_id，由后端解析活跃连接 instance = composio.triggers.create( slug="GMAIL_NEW_GMAIL_MESSAGE", user_id=user_id, trigger_config={}, ) # 禁用 / 启用 / 删除实例 composio.triggers.disable(trigger_id=instance.trigger_id) composio.triggers.enable(trigger_id=instance.trigger_id) composio.triggers.delete(trigger_id=instance.trigger_id)

Webhook 验签（Flask 示例）

示例文件内还以注释形式给出 Webhook 验签的 Flask 实现：从请求头读取webhook-id、webhook-signature、webhook-timestamp，调用composio.triggers.verify_webhook(...)校验签名，校验通过后可从结果中读取version（WebhookVersion：V1/V2/V3）、payload（归一化的 TriggerEvent）与raw_payload（原始负载）；签名不合法时抛WebhookSignatureVerificationError，对应返回 401。验签密钥来自环境变量COMPOSIO_WEBHOOK_SECRET。

订阅式触发

示例最后演示订阅模式：subscription = composio.triggers.subscribe()，用@subscription.handle(toolkit="GITHUB")、@subscription.handle(toolkit="SLACK")注册按工具包分发的回调函数，最后subscription.wait_forever()阻塞等待事件。这是轮询之外的另一种事件消费方式，适合长时间运行的常驻进程。

MCP 集成示例（mcp_example.py）

mcp_example.py 展示如何把 Composio 工具通过 MCP（Model Context Protocol）接入 LangChain Agent：

import asyncio, os, time from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent from composio import Composio composio = Composio() mcp_config = composio.mcp.create( # 命名遵循 examples-<label>-<unix-seconds>，供 --gc 识别 name=f"examples-lc-slack-{int(time.time())}", toolkits=[{"toolkit": "slack", "auth_config_id": slack_auth_config_id}], # 控制暴露的工具数量；LLM 提供方对单请求工具数有上限 allowed_tools=["SLACK_LIST_ALL_CHANNELS", "SLACK_SEARCH_MESSAGES"], ) mcp_server = mcp_config.generate(user_id=user_id) client = MultiServerMCPClient( { "composio": { "url": mcp_server["url"], "transport": "streamable_http", "headers": {"x-api-key": os.environ["COMPOSIO_API_KEY"]}, } } ) async def langchain_mcp(message: str): tools = await client.get_tools() agent = create_react_agent("openai:gpt-4.1", tools) response = await agent.ainvoke({"messages": message}) return response mcp_response = asyncio.run(langchain_mcp("Show me 20 most used slack channels"))

关键点：

• MCP 配置名采用examples-<label>-<unix-seconds>规范命名，这正是--gc清理 MCP 配置的匹配依据（见 scripts/examples-provision.mjs）；
• allowed_tools用于收窄暴露给 LLM 的工具集合，规避提供方单请求工具数量上限；
• MCP 端点通过x-api-key请求头用 Composio API 密钥认证，传输采用streamable_http传输层。

修饰器示例（modifiers.py）

modifiers.py 演示 Composio 的修饰器（modifier）机制——在工具执行链路中插入自定义逻辑的钩子：

from composio import ( Composio, after_execute, before_execute, before_file_upload, schema_modifier, ) from composio.types import Tool, ToolExecuteParams, ToolExecutionResponse composio = Composio() @before_execute(tools=["HACKERNEWS_GET_USER"]) def before_execute_modifier(tool, toolkit, params): # 执行前改写请求参数 print("before_execute_modifier", tool, toolkit) return params @before_file_upload(tools=["HACKERNEWS_GET_USER"]) def rewrite_upload_path(path, tool, toolkit): """每工具的文件路径策略钩子。""" return path @after_execute(tools=["HACKERNEWS_GET_USER"]) def after_execute_modifier(tool, toolkit, response): # 执行后改写响应，例如只保留指定字段 return {**response, "data": {"karama": response["data"]["karama"]}} # 执行时挂载修饰器 response = composio.tools.execute( user_id="default", slug="HACKERNEWS_GET_USER", arguments={"username": "pg"}, modifiers=[before_execute_modifier, rewrite_upload_path, after_execute_modifier], ) @schema_modifier(tools=["HACKERNEWS_GET_USER"]) def modify_schema(tool, toolkit, schema): return schema tools = composio.tools.get( user_id="default", slug="HACKERNEWS_GET_USER", modifiers=[modify_schema], )

修饰器四类职责清晰：before_execute改写请求参数、after_execute改写响应、before_file_upload实施文件上传路径策略、schema_modifier调整暴露给 LLM 的工具 schema。修饰器可同时作用于单个工具或整个工具包——示例后半部分用@before_execute(toolkits=["NOTION"])展示了如何为 Notion 工具包动态注入自定义认证头（从NOTION_API_KEY环境变量读取，写入custom_auth_params）。修饰器列表也可用于tools.get，在工具"暴露前"改写其 schema。

FastAPI 授权端点示例（fastapi_app.py）

fastapi_app.py 展示把 OAuth 授权流程嵌入 Web 应用的标准做法：

from fastapi import FastAPI from fastapi.responses import RedirectResponse from composio import Composio app = FastAPI() composio = Composio() @app.get("/authorize/{toolkit}") def authorize_app(toolkit: str): user_id = os.environ.get("COMPOSIO_EXAMPLES_USER_ID", "") auth_config_id = os.environ.get("COMPOSIO_EXAMPLES_GMAIL_AUTH_CONFIG_ID", "") connection_request = composio.connected_accounts.initiate( user_id=user_id, auth_config_id=auth_config_id, ) return RedirectResponse(url=connection_request.redirect_url) if __name__ == "__main__": import uvicorn for var in ("COMPOSIO_EXAMPLES_USER_ID", "COMPOSIO_EXAMPLES_GMAIL_AUTH_CONFIG_ID"): if not os.environ.get(var): raise SystemExit(f"{var} is required") # 启动前校验凭据与认证配置有效性 composio.auth_configs.get(os.environ["COMPOSIO_EXAMPLES_GMAIL_AUTH_CONFIG_ID"]) uvicorn.run(app, host="127.0.0.1", port=8000)

该示例的模式可直接迁移到真实产品：从自有用户体系取user_id，从配置取auth_config_id，initiate()拿到授权 URL 后 302 跳转。启动时的两个细节值得借鉴：一是缺失环境变量时SystemExit快速失败，二是在服务请求前先auth_configs.get()验证凭据与认证配置仍然有效，避免带着坏配置对外服务。

工具路由器示例（tool_router/ 子目录）

工具路由器（Tool Router）是示例集中较新的一组能力，核心思路是：为指定用户创建一个隔离的会话（session），会话内按需暴露工具，并可直接与各 LLM 框架对接。

与 OpenAI Agents 集成

tool_router/tools.py 演示基础用法：

from agents import Agent, Runner from composio_openai_agents import OpenAIAgentsProvider from composio import Composio, after_execute, before_execute composio = Composio(provider=OpenAIAgentsProvider()) # 为指定用户创建工具路由器会话，只暴露 gmail 工具包 session = composio.create(user_id="user_123", toolkits=["gmail"]) # 全局日志修饰器（空列表 = 作用于所有工具） @before_execute(tools=[]) def log_before_execute(tool, toolkit, params): print(f"🔧 Executing tool: {toolkit}.{tool}") return params @after_execute(tools=[]) def log_after_execute(tool, toolkit, response): print(f"✅ Completed tool: {toolkit}.{tool}") return response tools = session.tools(modifiers=[log_before_execute, log_after_execute]) agent = Agent( name="Gmail Assistant", instructions="You are a helpful assistant that helps users manage their Gmail accounts...", tools=tools, ) result = await Runner.run(starting_agent=agent, input="Fetch my last email from gmail and summarize it") print(result.final_output)

composio.create()返回的 session 提供session.tools()，产出已适配 OpenAI Agents 框架的工具对象；tools=[]的修饰器作用于会话内全部工具，适合统一打日志、审计或注入上下文。

预加载机制（preload）

tool_router/preload.py 演示三种直接暴露（preload）模式：通过preload={"tools": [...]}直接暴露 Composio 工具；通过@composio.experimental.tool(..., preload=True)暴露 SDK 自定义工具；以及带preload=False的嵌套自定义工具覆盖（在 preloaded 工具包内保持"仅可搜索"）。示例通过断言验证最终暴露的工具集合：

session = composio.create( user_id="preload-example-user", toolkits=["hackernews"], tools={"hackernews": {"enable": ["HACKERNEWS_GET_USER"]}}, preload={"tools": ["HACKERNEWS_GET_USER"]}, manage_connections=False, experimental={ "custom_tools": [lookup_internal_user, search_internal_users], "custom_toolkits": [internal_admin], }, ) tools = session.tools() tool_names = [tool.name for tool in tools] assert "HACKERNEWS_GET_USER" in tool_names assert "LOCAL_LOOKUP_INTERNAL_USER" in tool_names assert "LOCAL_INTERNAL_ADMIN_GET_ACCOUNT_HEALTH" in tool_names assert "LOCAL_SEARCH_INTERNAL_USERS" not in tool_names # 未 preload，保持搜索态 assert "LOCAL_INTERNAL_ADMIN_GET_ACCOUNT_AUDIT_LOG" not in tool_names # preload=False 覆盖

自定义工具名会带LOCAL_前缀以区分来源。示例运行方式：COMPOSIO_API_KEY=... OPENAI_API_KEY=... python examples/tool_router/preload.py。

与 Claude Agent SDK 集成

tool_router/claude_agent.py 展示通过 MCP 把会话工具接入 Claude：

from claude_agent_sdk import ClaudeAgentOptions, query from composio_claude_agent_sdk import ClaudeAgentSDKProvider from composio import Composio composio = Composio(provider=ClaudeAgentSDKProvider()) session = composio.create( user_id=os.environ["COMPOSIO_EXAMPLES_USER_ID"], mcp=True, # 让 session.mcp 可用 ) options = ClaudeAgentOptions( system_prompt="You are an expert Python developer", permission_mode="bypassPermissions", mcp_servers={ "composio": { "type": session.mcp.type, "url": session.mcp.url, "headers": session.mcp.headers, } }, ) async for message in query(prompt="Fetch my last email and summarize it.", options=options): print(message)

session.mcp直接提供 MCP 服务端三元组（type/url/headers），无需手工构造。工具路由器子目录中还包含 crewai_agent.py、langchain_agent.py、openai_agents.py 等框架适配示例，以及 session_update.py、files.py、authorize.py、tool_router_mcp.py 等专题示例，模式与上述一致：composio.create()建会话 →session.tools()/session.mcp取工具 → 交给对应 Agent 框架执行。

进阶阅读与验证

• 示例入口与运行约定：python/examples/README.md
• 供给/清理脚本源码：scripts/examples-provision.mjs（幂等检查、--initiate-missing、--gc/--dry-run的完整实现）
• Python SDK 核心客户端：python/composio/sdk.py 与 python/composio/client
• 类型与辅助构造：python/composio/types.py（含auth_scheme.api_key等辅助函数）
• SDK 单元测试（验证示例所用 API 行为）：python/tests/test_auth_configs.py、python/tests/test_connected_accounts.py、python/tests/test_triggers.py、python/tests/test_tool_router.py、python/tests/test_files.py

这些测试文件与示例脚本相互印证：示例展示的是"面向用户的典型调用"，测试则覆盖了参数序列化、认证方案、schema 语义等边界行为，两者结合可以更完整地理解 SDK 的契约。

【免费下载链接】composioComposio powers 1000+ toolkits, tool search, context management, authentication, and a sandboxed workbench to help you build AI agents that turn intent into action.项目地址: https://gitcode.com/GitHub_Trending/co/composio