Composio × Google ADK 集成指南:将 1000+ 工具桥接为 FunctionTool 的实战方案
【免费下载链接】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-google-adk是 Composio 官方为 Google ADK(Agent Development Kit)提供的适配层,它把 Composio 生态中的 1000+ 应用工具逐个包装成 Google ADK 的FunctionTool对象,让你的 Gemini 智能体可以直接以原生函数调用(function calling)的方式执行 GitHub、Gmail、Slack 等真实业务操作。本文以 python/providers/google_adk/README.md 为主线,深入讲解安装配置、Quickstart 代码、多轮会话复用,以及GoogleAdkProvider底层“schema → Python 签名/docstring → FunctionTool”的完整包装链路与源码实现,帮助你快速上手并理解其工作原理。
一、composio-google-adk 是什么
composio-google-adk是 Composio 为 Google ADK 框架提供的官方适配器。它解决的核心问题是:Google ADK 的Agent只能消费google.adk.tools.FunctionTool形式的函数声明,而 Composio 管理的 1000+ 应用工具(toolkits)并不天然具备该形态。通过本包,二者得以无缝对接:
- 工具形态转换:每个 Composio 工具被转换为
FunctionTool,ADK 可将其作为普通函数声明传递给 Gemini 模型; - 会话级工具作用域:每个 Composio 会话(session)绑定一个真实用户,工具执行时自动携带该用户的连接凭证;
- 类型安全:SDK 泛型自动推断为
Composio[FunctionTool, list[FunctionTool]],在 mypy 等类型检查器下可直接获得类型提示保障。
包的元信息可在 python/providers/google_adk/pyproject.toml 与 python/providers/google_adk/setup.py 中确认:包名composio-google-adk(导入名composio_google_adk),当前仓库版本 0.21.1,要求 Python >= 3.10,< 4,核心依赖为google-adk>=2.5.0与composio。
二、安装与密钥配置
按官方 README 安装三个包:
pip install composio composio-google-adk google-adk安装完成后,需要配置两类密钥(通过环境变量注入):
COMPOSIO_API_KEY:从 Composio Dashboard 的 Settings 页面获取,用于调用 Composio 后端获取工具列表与执行工具;GOOGLE_API_KEY:Google Gemini API 密钥,用于 ADK 驱动 Gemini 模型完成推理与函数调用。
SDK 层面对COMPOSIO_API_KEY的处理可在 python/composio/sdk.py 中看到:Composio.__init__会优先取构造参数中的api_key,否则回退到环境变量COMPOSIO_API_KEY,两者都缺失时抛出ApiKeyNotProvidedError。此外还支持COMPOSIO_BASE_URL等环境变量,便于自托管部署场景。
三、Quickstart:从零构建一个能执行 GitHub 操作的 ADK 智能体
3.1 完整代码(来自官方 README)
from composio import Composio from composio_google_adk import GoogleAdkProvider from google.adk.agents import Agent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types composio = Composio(provider=GoogleAdkProvider()) # Each Composio session is scoped to one of your users composio_session = composio.create(user_id="user_123") tools = composio_session.tools() agent = Agent( name="personal_assistant", model="gemini-2.0-flash", instruction="You are a helpful assistant. Use Composio tools to take action.", tools=tools, ) session_service = InMemorySessionService() session_service.create_session_sync( app_name="personal_assistant", user_id="user_123", session_id="1234", ) runner = Runner( agent=agent, app_name="personal_assistant", session_service=session_service, ) events = runner.run( user_id="user_123", session_id="1234", new_message=types.Content( role="user", parts=[types.Part(text="Star the repository composiohq/composio on GitHub")], ), ) for event in events: if event.is_final_response() and event.content and event.content.parts: print(event.content.parts[0].text)3.2 代码拆解
- 初始化 SDK 并指定适配器:
Composio(provider=GoogleAdkProvider())。传入 provider 后,SDK 的泛型参数自动推断为FunctionTool/list[FunctionTool](参见 python/tests/test_type_inference_google_adk.py,该文件被 mypy 静态分析以验证类型推断正确性)。 - 创建会话:
composio.create(user_id="user_123")。每个 Composio 会话对应一个最终用户,工具的鉴权与执行都绑定在该用户维度上。从源码看,composio.create与composio.use是composio.sessions(ToolRouter)的顶层快捷方式(python/composio/sdk.py)。 - 拉取会话级工具:
composio_session.tools()返回list[FunctionTool],即已按 ADK 格式包装好的工具集合(python/composio/core/models/tool_router_session.py)。 - 组装 ADK 智能体:将工具列表直接传入
Agent(..., tools=tools),模型选用gemini-2.0-flash,并在 instruction 中明确告知模型使用 Composio 工具采取行动。 - 建立会话与 Runner:
InMemorySessionService提供进程内会话存储,Runner负责驱动 agent 执行。 - 运行:
runner.run(...)以用户消息启动一轮对话;遍历返回的 events,取is_final_response()的最终文本打印。示例中,模型会识别“Star the repository composiohq/composio on GitHub”这一意图,通过FunctionTool调用 Composio 的 GitHub 工具完成真实操作。
仓库中还提供了一个同思路的可运行脚本 python/providers/google_adk/google_adk_demo.py,其区别在于通过composio.tools.get(user_id=user_id, toolkits=["GITHUB"])显式指定 GITHUB 工具包,且最终响应判断更完整(额外校验content.parts非空)。
四、多轮对话:用 session_id 复用会话
README 明确指出:多轮场景下不要再次调用create(),而应保存composio_session.session_id并用composio.use(session_id)复用。
# 第一轮:创建会话并保存 id composio_session = composio.create(user_id="user_123") session_id = composio_session.session_id # 后续轮次:复用已有会话 composio_session = composio.use(session_id=session_id) tools = composio_session.tools()这背后的语义是:create()会为指定user_id创建全新的工具会话上下文,重复创建既浪费后端资源,也无法延续前一轮对话中建立/授权的连接状态;use()则按会话 ID 恢复既有上下文,保留连接、工具配置与文件等会话内状态。底层实现参见 python/composio/core/models/tool_router.py,其中use()还支持传入custom_tools/custom_toolkits将 SDK 本地自定义工具绑定到会话中。
此外,create()本身支持丰富的会话配置(python/composio/core/models/tool_router.py),例如:
toolkits:启用/禁用工具包,如['github', 'slack']或{'enable': ['github']};tools:按工具包细粒度控制具体工具,如{'gmail': ['GMAIL_SEND_EMAIL', 'GMAIL_SEARCH']};tags:按标签(readOnlyHint、destructiveHint、idempotentHint、openWorldHint)过滤工具;connected_accounts:为工具包预绑定连接账户,如{'github': 'ca_xxx'};auth_configs:指定认证配置 ID,如{'github': 'ac_xxx'}。
这些参数均可用于create(user_id=..., **配置)精细控制会话内可用工具范围。
五、原理深挖:GoogleAdkProvider 如何把工具包成 FunctionTool
5.1 包装流程总览
GoogleAdkProvider继承自AgenticProvider[FunctionTool, list[FunctionTool]](python/providers/google_adk/composio_google_adk/provider.py),并注册了 provider 名称google_adk。基类AgenticProvider定义了两个抽象方法wrap_tool/wrap_tools(python/composio/core/provider/agentic.py),Google ADK 适配器对二者分别实现:wrap_tools只是对wrap_tool的批量调用。
单个工具的包装流程为:
- 取出工具输入 JSON Schema(
tool.input_parameters); alias_tool_input_schema将 Schema 属性名别名化为合法的 Python 参数名;- 依据属性 Schema 的描述生成带
Args:/Returns:段的 docstring; - 构造执行闭包
_execute,先restore_arguments还原原始参数名,再经normalize_tool_arguments防御性归一化后调用execute_tool; - 用
function_signature_from_jsonschema生成inspect.Parameter列表并覆写函数的__signature__、__annotations__与__doc__; - 最终以
FunctionTool(function)返回,供 ADK 使用。
5.2 Schema 别名化:兼容 Python 标识符
Composio 工具的输入 Schema 字段名未必是合法 Python 标识符(例如含连字符、或与 Python 关键字冲突如from)。alias_tool_input_schema(python/composio/utils/shared.py)会做深度拷贝、内联内部$ref引用,并将属性名转换为安全的 Python 参数名:
- Python 关键字保留历史后缀
_rs(如from→from_rs); - 其他无法作为标识符的名字转为 Pydantic 安全标识符;
- 长别名截断到 64 字符,以兼容 Anthropic 风格工具 Schema 校验器;
- 若两个属性映射到同一别名,抛出
InvalidSchemaError而非静默猜测。
包装器记录aliases反向映射,执行时用restore_arguments把模型传入的别名参数还原为后端真实字段名(python/composio/utils/shared.py)。
5.3 签名生成:JSON Schema → inspect.Parameter
function_signature_from_jsonschema(python/composio/utils/openapi.py)遍历 Schema 的properties,将oneOf/anyOf/allOf/type映射为对应的 Python 类型注解,构造inspect.Parameter(POSITIONAL_OR_KEYWORD类型)。参数默认值遵循以下规则:必填参数(出现在required中)或开启skip_default时,默认值置为inspect.Parameter.empty;否则使用 Schema 中的default。适配器类上声明了__schema_skip_defaults__ = True(python/providers/google_adk/composio_google_adk/provider.py),其语义在基类 python/composio/core/provider/base.py 中定义:即默认跳过 Schema 中的默认值,让 ADK/Gemini 显式传入每个参数,避免模型依赖隐含默认值导致行为不一致。
生成后,适配器用types.FunctionType复制执行闭包并设置函数名tool.slug,再通过setattr覆写__signature__与__annotations__(返回类型固定为dict),最后写入 docstring——这样 ADK 在构建FunctionTool时即可拿到完整的函数声明(名称、参数类型、文档),并原样传递给 Gemini 作为 function declaration。
5.4 执行与防御性参数归一化
包装出的函数在被 ADK 调用时执行_execute(**kwargs):
def _execute(**kwargs): kwargs = aliases.restore_arguments(kwargs) return execute_tool( slug=tool.slug, arguments=normalize_tool_arguments(kwargs) )normalize_tool_arguments(python/composio/utils/shared.py)是每个 provider 共用的参数归一化入口:None转为{}(模型对无参工具可能不传参数);字符串会被 JSON 解析(空串转{});无法解析为 dict 的输入(list、原始类型、非对象 JSON)抛出InvalidParams。这从底层防御了模型偶尔以 JSON 字符串形式输出参数导致执行失败的问题(仓库中对应 issue #2406 的修复)。执行结果遵循AgenticProviderExecuteFn协议(python/composio/core/provider/agentic.py),返回含data、error、successful三个键的字典。
5.5 类型推断保障
python/tests/test_type_inference_google_adk.py 通过assert_type(tools, list[FunctionTool])静态验证了四种获取工具的方式(toolkits、slug、tools列表、search)以及无显式注解时的自动推断,全部指向list[FunctionTool]。这意味着在接入 ADK 时,IDE 与类型检查器能够对工具列表给出可靠的类型提示,杜绝“工具传不进去”之类的隐式错误。
六、安装运行环境与版本前提
- Python 版本:
>=3.10,<4(见 python/providers/google_adk/pyproject.toml); - 依赖版本:
google-adk>=2.5.0、composio(当前仓库主 SDK); - 模型要求:示例使用
gemini-2.0-flash,需配置可用的GOOGLE_API_KEY; - 运行前提:联网访问 Composio 后端(或自托管
COMPOSIO_BASE_URL),且目标应用(如 GitHub)的用户连接已授权,工具执行才能携带有效凭证。
七、小结
composio-google-adk的价值在于用极少的胶水代码打通了 Google ADK 与 Composio 工具生态:创建用户会话 → 拉取FunctionTool工具列表 → 注入Agent→ 由 Runner 驱动执行。其底层GoogleAdkProvider通过 JSON Schema 别名化、Python 签名生成、docstring 构造与参数归一化四步,把每个 Composio 工具稳健地适配为 ADK 原生函数声明;配合session_id复用机制,可平滑支撑多轮对话与长时间运行的智能体应用。
如果想快速验证,可直接参考仓库中的 python/providers/google_adk/google_adk_demo.py 脚本,在配置好两个密钥后运行,体验“一句话让 Gemini 通过 Composio 完成真实 GitHub 操作”的完整链路。
【免费下载链接】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
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考