使用 DataHub Agent Context 构建 Google ADK 自主数据智能体【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 的 Agent Context Kit 提供了将企业数据上下文数据所有权、血缘、文档、质量信号等直接注入 AI 智能体的能力。本文聚焦 Google ADKAgent Development Kit这一官方集成路径讲解如何在 Google ADK 中通过 Python 工具直嵌或 MCP Server 两种方式接入 DataHub读完即可基于当前仓库中的示例代码搭建一个能检索数据集、追踪血缘、查询文档乃至写入元数据的自主数据智能体。集成总览两种接入方式在 Google ADK 中消费 DataHub 上下文官方提供两条互补路径见 google-adk.mdPython 工具直嵌通过datahub-agent-context包的build_google_adk_tools()把 DataHub 能力包装成普通 Python 函数直接传给 ADKAgent的tools参数零 MCP 基础设施依赖MCP Server 连接让 ADK 通过内置的McpToolset连接 DataHub 的 MCP serverDataHub Cloud 托管端点或自建端点由 MCP 协议统一发现工具。两种方式共享同一套底层 MCP 工具实现位于 datahub-agent-context/src/datahub_agent_context/mcp_tools/因此能力集一致区别仅在于工具如何被 ADK 发现与调用。前置条件开始之前需要准备Python 3.10 及以上版本Google ADKpip install google-adk一个可访问的 DataHub 实例以及 个人访问令牌一个 Google API KeyGemini Developer API或Google Cloud 凭据Vertex AI。安装pip install datahub-agent-context[google-adk][google-adk]是可选依赖组。从当前仓库的 requirements.txt 可以看到该示例锁定的版本约束为google-adk1.0.0,2.0.0并因 CVE-2025-43859 对httpcore1.0.9与h110.16做了下限约束。本地开发可改用pip install -e datahub-agent-context[google-adk]方式安装。快速开始Python 工具直嵌第一步创建 DataHub 客户端from datahub.sdk.main_client import DataHubClient client DataHubClient.from_env()DataHubClient.from_env()会从环境变量读取连接配置GMS 地址与令牌。若需显式指定可参考 simple_search.py 的写法import os from datahub.sdk.main_client import DataHubClient datahub_gms_url os.getenv(DATAHUB_GMS_URL) if datahub_gms_url is None: client DataHubClient.from_env() else: client DataHubClient(serverdatahub_gms_url, tokenos.getenv(DATAHUB_GMS_TOKEN))对应环境变量为DATAHUB_GMS_URL默认http://localhost:8080与DATAHUB_GMS_TOKEN。第二步构建工具集from datahub_agent_context.google_adk_tools import build_google_adk_tools # 默认只读设置 include_mutationsTrue 可启用写操作 tools build_google_adk_tools(client, include_mutationsFalse)从 builder.py 的源码可以确认默认只读工具清单工具作用search按关键词搜索数据集、仪表盘等实体get_entities获取实体完整详情schema、所有权、文档、标签list_schema_fields列出数据集的字段列get_lineage追踪上游/下游血缘get_lineage_paths_between查询两个实体之间的血缘路径get_dataset_queries获取与数据集关联的 SQL 查询get_dataset_assertions获取数据集的质量断言assertionslist_incidents列出事件incidentssearch_documents/grep_documents搜索知识库文章与文档get_me获取当前用户信息当include_mutationsTrue时追加写操作工具update_description更新描述、set_domains/remove_domains管理域、add_owners/remove_owners管理所有者、add_tags/remove_tags管理标签、add_glossary_terms/remove_glossary_terms关联术语表术语、add_structured_properties/remove_structured_properties结构化属性、save_document保存文档、raise_incident/resolve_incident创建/解决事件。这些包装函数的实现原理见 utils.pycreate_context_wrapper利用contextvars在执行函数前将DataHubClient注入上下文执行后重置同时把工具抛出的ItemNotFoundError转换为结构化字典返回保证 LLM 拿到的是可读消息而非未处理异常。另有一个 Cloud 专属的 build_google_adk_cloud_tools 函数可启用Ask DataHubAI 助手工具ask_datahub_chat/get_datahub_chat仅适用于 DataHub Cloud 实例OSS 实例请勿启用。第三步注入 ADK Agentfrom google.adk.agents import Agent agent Agent( modelgemini-2.5-flash, namedatahub_agent, descriptionA data discovery assistant with access to DataHub., instructionUse the available tools to search for datasets, get entity details, and trace lineage. Always include URNs in your answers., toolstools, )instruction提示词是决定智能体是否真正调用工具的关键——建议在其中明确指定应使用的工具类别如先搜索、再取详情、再追血缘并约定回答格式如始终在回答中包含 URN这一经验同样体现在 basic_agent.py 的SYSTEM_PROMPT中。完整可运行的最小示例仓库中的 simple_search.py 是绝对最小可运行版本完整展示了从建客户端、建工具、建 Agent 到用Runner流式执行一次查询的闭环import asyncio from google.adk.agents import Agent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types from datahub.sdk.main_client import DataHubClient from datahub_agent_context.google_adk_tools import build_google_adk_tools client DataHubClient.from_env() tools build_google_adk_tools(client, include_mutationsFalse) agent Agent( modelgemini-2.5-flash, namedatahub_agent, instructionYou help users find datasets in DataHub. Provide clear, concise answers., toolstools, ) async def main() - None: session_service InMemorySessionService() session await session_service.create_session(app_namedatahub_simple_search, user_iduser) runner Runner(agentagent, app_namedatahub_simple_search, session_servicesession_service) async for event in runner.run_async( user_iduser, session_idsession.id, new_messagetypes.Content(roleuser, parts[types.Part(textFind datasets about users)]), ): if event.is_final_response() and event.content and event.content.parts: print(fAgent: {event.content.parts[0].text}) asyncio.run(main())进阶示例DataHub BigQuery 数据分析智能体basic_agent.py 演示了更完整的数据分析师智能体同时挂载 DataHub 工具含写操作与 Google ADK 自带的BigQueryToolset形成先到 DataHub 发现和确认表结构 → 再写 SQL 查真实数据 → 最后综合解释结果的工作流。其中通过google.auth.default()探测 GCP 凭据未找到时静默降级、只保留 DataHub 工具BigQuery 工具以WriteMode.BLOCKED配置为只读使用InMemorySessionService维持跨轮次会话上下文流式事件循环中实时打印工具调用名与参数便于观察智能体行为。通过 MCP Server 连接不想直嵌 Python 工具时可让 ADK 通过McpToolset连接 DataHub 的 MCP serverfrom google.adk.tools.mcp_tool import McpToolset from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams toolset McpToolset( connection_paramsStreamableHTTPConnectionParams( urlhttps://tenant.acryl.io/integrations/ai/mcp ), headers{Authorization: fBearer {YOUR_TOKEN}}, ) # 在当前任务内主动初始化确保 AsyncExitStack 归属于本任务 await toolset.get_tools() agent Agent( modelgemini-2.5-flash, namedatahub_agent, instructionYou help users find datasets in DataHub., tools[toolset], )用完记得await toolset.close()。自建 DataHub CoreOSS时MCP 端点默认为http://localhost:8080/mcp示例 simple_mcp.py 展示了用DATAHUB_MCP_SERVER_URL环境变量覆盖默认值、并把DataHubClient的 token 直接放进Authorization头的写法且把await toolset.get_tools()与toolset.close()放进try/finally保证资源释放。关于 AsyncExitStack 的要点ADK 的McpToolset通过异步上下文管理器持有 MCP 会话。若不在当前任务内主动调用await toolset.get_tools()ADK 会在派生的任务中创建会话随后调用close()可能抛出Attempted to exit cancel scope in a different task错误。正确做法是在拥有 toolset 的同一 async 任务中初始化并在finally块中关闭。使用 Vertex AI 替代 Gemini Developer API默认情况下 ADK 通过GOOGLE_API_KEY使用 Gemini Developer API。若改用 Vertex AI不要设置GOOGLE_API_KEY——ADK 会自动回退到 Application Default CredentialsADC确保已执行过gcloud auth application-default login。工具上下文注入原理无论走哪条接入路径DataHub Agent Context 的 Python 工具都依赖contextvars实现客户端注入create_context_wrapper在执行工具函数前set_client(client)、结束后reset_client(token)工具内部通过get_datahub_client()获取客户端。这保证了对同一批 MCP 工具函数的复用——它们既能在 MCP 服务端按标准协议调用也能被包装成普通函数直接交给 ADK这是两种接入方式、一套工具实现的底层机制见 utils.py 与 context.py。常见问题排查工具执行报错检查 DataHub 连接client.config与令牌权限确认 token 是否具备对应实体的读取/写入范围。智能体不调用工具强化instruction提示词明确列出应使用的工具与回答格式或换用工具调用能力更强的模型Gemini 2.0 及以上。出现AsyncExitStack/ 任务错误在与 toolset 相同的 async 任务中调用await toolset.get_tools()并在finally块中await toolset.close()。导入报错执行pip install datahub-agent-context[google-adk] google-adk确保可选依赖组与 ADK 本体都已安装。延伸阅读Agent Context Kit 总览了解可构建的智能体类型Text-to-SQL 数据分析、数据质量、数据治理/合规与其他平台的接入指南MCP Server 指南MCP 端点的鉴权与自建部署方式个人访问令牌申请 DataHub API 访问凭据可运行示例 basic_agent.py · simple_search.py · simple_mcp.py。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 使用 DataHub Agent Context 构建 Google ADK 自主数据智能体
使用 DataHub Agent Context 构建 Google ADK 自主数据智能体【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 的 Agent Context Kit 提供了将企业数据上下文数据所有权、血缘、文档、质量信号等直接注入 AI 智能体的能力。本文聚焦 Google ADKAgent Development Kit这一官方集成路径讲解如何在 Google ADK 中通过 Python 工具直嵌或 MCP Server 两种方式接入 DataHub读完即可基于当前仓库中的示例代码搭建一个能检索数据集、追踪血缘、查询文档乃至写入元数据的自主数据智能体。集成总览两种接入方式在 Google ADK 中消费 DataHub 上下文官方提供两条互补路径见 google-adk.mdPython 工具直嵌通过datahub-agent-context包的build_google_adk_tools()把 DataHub 能力包装成普通 Python 函数直接传给 ADKAgent的tools参数零 MCP 基础设施依赖MCP Server 连接让 ADK 通过内置的McpToolset连接 DataHub 的 MCP serverDataHub Cloud 托管端点或自建端点由 MCP 协议统一发现工具。两种方式共享同一套底层 MCP 工具实现位于 datahub-agent-context/src/datahub_agent_context/mcp_tools/因此能力集一致区别仅在于工具如何被 ADK 发现与调用。前置条件开始之前需要准备Python 3.10 及以上版本Google ADKpip install google-adk一个可访问的 DataHub 实例以及 个人访问令牌一个 Google API KeyGemini Developer API或Google Cloud 凭据Vertex AI。安装pip install datahub-agent-context[google-adk][google-adk]是可选依赖组。从当前仓库的 requirements.txt 可以看到该示例锁定的版本约束为google-adk1.0.0,2.0.0并因 CVE-2025-43859 对httpcore1.0.9与h110.16做了下限约束。本地开发可改用pip install -e datahub-agent-context[google-adk]方式安装。快速开始Python 工具直嵌第一步创建 DataHub 客户端from datahub.sdk.main_client import DataHubClient client DataHubClient.from_env()DataHubClient.from_env()会从环境变量读取连接配置GMS 地址与令牌。若需显式指定可参考 simple_search.py 的写法import os from datahub.sdk.main_client import DataHubClient datahub_gms_url os.getenv(DATAHUB_GMS_URL) if datahub_gms_url is None: client DataHubClient.from_env() else: client DataHubClient(serverdatahub_gms_url, tokenos.getenv(DATAHUB_GMS_TOKEN))对应环境变量为DATAHUB_GMS_URL默认http://localhost:8080与DATAHUB_GMS_TOKEN。第二步构建工具集from datahub_agent_context.google_adk_tools import build_google_adk_tools # 默认只读设置 include_mutationsTrue 可启用写操作 tools build_google_adk_tools(client, include_mutationsFalse)从 builder.py 的源码可以确认默认只读工具清单工具作用search按关键词搜索数据集、仪表盘等实体get_entities获取实体完整详情schema、所有权、文档、标签list_schema_fields列出数据集的字段列get_lineage追踪上游/下游血缘get_lineage_paths_between查询两个实体之间的血缘路径get_dataset_queries获取与数据集关联的 SQL 查询get_dataset_assertions获取数据集的质量断言assertionslist_incidents列出事件incidentssearch_documents/grep_documents搜索知识库文章与文档get_me获取当前用户信息当include_mutationsTrue时追加写操作工具update_description更新描述、set_domains/remove_domains管理域、add_owners/remove_owners管理所有者、add_tags/remove_tags管理标签、add_glossary_terms/remove_glossary_terms关联术语表术语、add_structured_properties/remove_structured_properties结构化属性、save_document保存文档、raise_incident/resolve_incident创建/解决事件。这些包装函数的实现原理见 utils.pycreate_context_wrapper利用contextvars在执行函数前将DataHubClient注入上下文执行后重置同时把工具抛出的ItemNotFoundError转换为结构化字典返回保证 LLM 拿到的是可读消息而非未处理异常。另有一个 Cloud 专属的 build_google_adk_cloud_tools 函数可启用Ask DataHubAI 助手工具ask_datahub_chat/get_datahub_chat仅适用于 DataHub Cloud 实例OSS 实例请勿启用。第三步注入 ADK Agentfrom google.adk.agents import Agent agent Agent( modelgemini-2.5-flash, namedatahub_agent, descriptionA data discovery assistant with access to DataHub., instructionUse the available tools to search for datasets, get entity details, and trace lineage. Always include URNs in your answers., toolstools, )instruction提示词是决定智能体是否真正调用工具的关键——建议在其中明确指定应使用的工具类别如先搜索、再取详情、再追血缘并约定回答格式如始终在回答中包含 URN这一经验同样体现在 basic_agent.py 的SYSTEM_PROMPT中。完整可运行的最小示例仓库中的 simple_search.py 是绝对最小可运行版本完整展示了从建客户端、建工具、建 Agent 到用Runner流式执行一次查询的闭环import asyncio from google.adk.agents import Agent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types from datahub.sdk.main_client import DataHubClient from datahub_agent_context.google_adk_tools import build_google_adk_tools client DataHubClient.from_env() tools build_google_adk_tools(client, include_mutationsFalse) agent Agent( modelgemini-2.5-flash, namedatahub_agent, instructionYou help users find datasets in DataHub. Provide clear, concise answers., toolstools, ) async def main() - None: session_service InMemorySessionService() session await session_service.create_session(app_namedatahub_simple_search, user_iduser) runner Runner(agentagent, app_namedatahub_simple_search, session_servicesession_service) async for event in runner.run_async( user_iduser, session_idsession.id, new_messagetypes.Content(roleuser, parts[types.Part(textFind datasets about users)]), ): if event.is_final_response() and event.content and event.content.parts: print(fAgent: {event.content.parts[0].text}) asyncio.run(main())进阶示例DataHub BigQuery 数据分析智能体basic_agent.py 演示了更完整的数据分析师智能体同时挂载 DataHub 工具含写操作与 Google ADK 自带的BigQueryToolset形成先到 DataHub 发现和确认表结构 → 再写 SQL 查真实数据 → 最后综合解释结果的工作流。其中通过google.auth.default()探测 GCP 凭据未找到时静默降级、只保留 DataHub 工具BigQuery 工具以WriteMode.BLOCKED配置为只读使用InMemorySessionService维持跨轮次会话上下文流式事件循环中实时打印工具调用名与参数便于观察智能体行为。通过 MCP Server 连接不想直嵌 Python 工具时可让 ADK 通过McpToolset连接 DataHub 的 MCP serverfrom google.adk.tools.mcp_tool import McpToolset from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams toolset McpToolset( connection_paramsStreamableHTTPConnectionParams( urlhttps://tenant.acryl.io/integrations/ai/mcp ), headers{Authorization: fBearer {YOUR_TOKEN}}, ) # 在当前任务内主动初始化确保 AsyncExitStack 归属于本任务 await toolset.get_tools() agent Agent( modelgemini-2.5-flash, namedatahub_agent, instructionYou help users find datasets in DataHub., tools[toolset], )用完记得await toolset.close()。自建 DataHub CoreOSS时MCP 端点默认为http://localhost:8080/mcp示例 simple_mcp.py 展示了用DATAHUB_MCP_SERVER_URL环境变量覆盖默认值、并把DataHubClient的 token 直接放进Authorization头的写法且把await toolset.get_tools()与toolset.close()放进try/finally保证资源释放。关于 AsyncExitStack 的要点ADK 的McpToolset通过异步上下文管理器持有 MCP 会话。若不在当前任务内主动调用await toolset.get_tools()ADK 会在派生的任务中创建会话随后调用close()可能抛出Attempted to exit cancel scope in a different task错误。正确做法是在拥有 toolset 的同一 async 任务中初始化并在finally块中关闭。使用 Vertex AI 替代 Gemini Developer API默认情况下 ADK 通过GOOGLE_API_KEY使用 Gemini Developer API。若改用 Vertex AI不要设置GOOGLE_API_KEY——ADK 会自动回退到 Application Default CredentialsADC确保已执行过gcloud auth application-default login。工具上下文注入原理无论走哪条接入路径DataHub Agent Context 的 Python 工具都依赖contextvars实现客户端注入create_context_wrapper在执行工具函数前set_client(client)、结束后reset_client(token)工具内部通过get_datahub_client()获取客户端。这保证了对同一批 MCP 工具函数的复用——它们既能在 MCP 服务端按标准协议调用也能被包装成普通函数直接交给 ADK这是两种接入方式、一套工具实现的底层机制见 utils.py 与 context.py。常见问题排查工具执行报错检查 DataHub 连接client.config与令牌权限确认 token 是否具备对应实体的读取/写入范围。智能体不调用工具强化instruction提示词明确列出应使用的工具与回答格式或换用工具调用能力更强的模型Gemini 2.0 及以上。出现AsyncExitStack/ 任务错误在与 toolset 相同的 async 任务中调用await toolset.get_tools()并在finally块中await toolset.close()。导入报错执行pip install datahub-agent-context[google-adk] google-adk确保可选依赖组与 ADK 本体都已安装。延伸阅读Agent Context Kit 总览了解可构建的智能体类型Text-to-SQL 数据分析、数据质量、数据治理/合规与其他平台的接入指南MCP Server 指南MCP 端点的鉴权与自建部署方式个人访问令牌申请 DataHub API 访问凭据可运行示例 basic_agent.py · simple_search.py · simple_mcp.py。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考