--- title: "Google ADK" description: "Google ADK（Agent Development Kit）为 Gemini 和其他模型构建 Agent 与多 Agent 树。Dex adapter 把 ADK Agent 放进 durable session 中运行。它支持 Python。" canonical_url: "https://docs.superdurable.io/zh-Hans/agent-frameworks/google-adk/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/agent-frameworks/google-adk.mdx" locale: "zh-Hans" --- # Google ADK Google ADK（Agent Development Kit）为 Gemini 和其他模型构建 Agent 与多 Agent 树。Dex adapter 把 ADK Agent 放进 durable session 中运行。它支持 Python。 ## 何时使用 Agent 已经用 Google ADK 编写时，使用它，包括 sequential、loop 和 parallel Agent。保留 Agent 树和它的 tool，由 Dex 管理 session。 ## Session 如何运行 Adapter 给 ADK App 加上一个 plugin。对每个 tool call，plugin 都会向 ADK 请求确认，因此 ADK 会在 tool 运行之前暂停。每个 **RunModel** Step 调用一次模型，并保存这个 turn 的 ADK event。 每个 tool call 在自己的 Step 中并行运行：read 或 idempotent write 用 **CallTool**，其他写操作用 **CallSingleAttemptTool**。需要审批的调用先在 **AwaitToolApproval** 中等待。Tool Step 运行 tool 一次。下一个 **RunModel** 确认暂停的调用，plugin 返回记录下来的结果，而不会再次运行 tool。 查询订单的 turn 依次运行 **RunModel**、**CallTool**、**ResumeTurn** 和 **RunModel**。 ## 安装 把 adapter 与它的 framework 一起安装。每个 adapter 都会带上 Dex Agents core。 **Python** ```bash pip install dex-agents-google-adk ``` 例子: [agents/python/google-adk/README.md](https://github.com/superdurable/dex/tree/main/agents/python/google-adk/README.md) ## Example 这个 example 就是每个 framework 页面共用的[订单客服 Agent](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#the-examples)。 Flow type 是 **GoogleADKOrderSupport**（Python）。完整 example 位于 [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses)。 ### Agent 与 tool Tool 是 ADK **LlmAgent** 上普通的 Python 函数。Tool policy 指定每个 tool。 **Python** ```python def build_google_adk_order_support_flow( order_desk: OrderDesk, model: str | BaseLlm, slack_connection: slack.Connection | None ) -> AgentSessionFlow: """Builds the session Flow; ``model`` is a Gemini model name, such as "gemini-2.5-flash".""" # The tools are closures, so they call the OrderDesk passed in here. def look_up_order(order_id: str) -> str: """Look up one order by its order ID and return its items and status.""" return order_desk.describe_order(order_id) def cancel_order(order_id: str, reason: str) -> str: """Cancel one order that has not shipped yet.""" return order_desk.cancel_order(order_id, reason) agent = LlmAgent( name="order_support", model=model, instruction=ORDER_SUPPORT_INSTRUCTIONS, tools=[look_up_order, cancel_order], ) return adk_agent_session_flow( agent, flow_type=GOOGLE_ADK_ORDER_SUPPORT_FLOW_TYPE, tool_sources=build_slack_tool_sources(slack_connection), tool_policies={ "look_up_order": ToolPolicy(ToolEffect.READ), "cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE), }, ) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/google_adk_order_support.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/google_adk_order_support.py) ### Worker 一个 Worker 为 example 中所有 harness 的 session Flow 提供服务。Agent session 的 Step 是 async 的，因此 Python Registry 允许 async handler。每个启动 session 或调用其 RPC 的进程，都要构建同一个 session Flow。 **Python** ```python # Agent session Steps are async, so the Registry allows async handlers. self.registry = Registry( tuple(example.session_flow for example in self.examples), allow_async_handlers=True ) config.blob_cache_dir.mkdir(parents=True, exist_ok=True) self.blob_cache = open_blob_cache(BlobCacheConfig(str(config.blob_cache_dir), 1 << 30)) self.worker = AsyncWorker( self.registry, self.blob_cache, WorkerOptions( server_address=config.server_address, pull=PullWorkerOptions(max_concurrent_tasks=MAX_CONCURRENT_TASKS), ), ) self.client = AsyncClient( self.registry, self.blob_cache, ClientOptions(server_address=config.server_address) ) # ... def _new_google_adk_example( config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None ) -> AgentHarnessExample: order_desk = OrderDesk() model = _google_adk_model(config.google_adk_model) flow = build_google_adk_order_support_flow(order_desk, model, slack_connection) return AgentHarnessExample("google-adk", flow, order_desk) # ... def _google_adk_model(model: str) -> str | BaseLlm: return GoogleADKDemoModel() if is_demo_model(model) else model ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/app.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/app.py) ### Session route 每个 harness 都提供这同一组 route。这些 route 和 session token 的说明见 [Session route](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#session-routes)。 **Python** ```python @blueprint.post("/sessions") async def start_session() -> Response: # The server picks the session ID, so a caller cannot claim an existing session. session_id = f"{harness_name}-order-support-{uuid.uuid4()}" await sessions.start_session(session_id) session_token = sign_session_id(app_state.config.session_key, session_id) return jsonify(session_id=session_id, session_token=session_token) @blueprint.post("/messages") async def send_user_message() -> Response: body, session_id = await session_request() accepted = await sessions.send_user_message(session_id, _required_string(body, "text")) return jsonify(accepted=accepted) @blueprint.post("/interrupts/resolve") async def resolve_interrupt() -> Response: body, session_id = await session_request() interrupt_id = _required_string(body, "interrupt_id") if not isinstance(approved := body.get("approved"), bool): raise BadRequest("approved must be a boolean") rejection_message = body.get("rejection_message") if rejection_message is not None and not isinstance(rejection_message, str): raise BadRequest("rejection_message must be a string") accepted = await sessions.resolve_interrupt( session_id, interrupt_id, approved=approved, rejection_message=rejection_message ) return jsonify(accepted=accepted) @blueprint.get("/snapshot") async def get_agent_snapshot() -> Response: session_id = request.args.get("session_id", "") if not session_id: raise BadRequest("session_id is required") authorize_session(session_id) try: snapshot = await read_agent_event_stream_snapshot( app_state.client, example.session_flow, session_id ) except FlowNotActiveOrNotFoundError as error: raise NotFound("the agent session is missing or no longer retained") from error return jsonify(snapshot.to_json()) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/http_routes.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/http_routes.py) ## 在本地运行 ### 启动 Dex 与 example 在第一个终端中，先渲染 session graph，再启动 Dex。Dex Web 启动时读取这个 graph，用它显示 session 和审批控件。 **Python** ```bash cd examples/python/agent-harnesses uv sync --locked mkdir -p definitions uv run --frozen dex-agents-session-graph --flow-type GoogleADKOrderSupport \ --permission order-support.operate --out definitions/google-adk.json dexcli dev --flow-rendering-dir ./definitions ``` 例子: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) **dexcli dev** 会一直在前台运行。在第二个终端中启动 example： **Python** ```bash cd examples/python/agent-harnesses uv run --frozen python -m agent_harness_examples ``` 例子: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) 应用监听 **127.0.0.1:8081**。Dex Web 位于 **127.0.0.1:8802**。 没有设置 **DEX_GOOGLE_ADK_MODEL** 时，Agent 使用 **mock/order-support**，这是一个不需要 key 的本地 demo model。收到 "look up order 1001" 时它调用 **look_up_order**，收到 "cancel order 1002" 时它调用 **cancel_order**。 | 变量 | 默认值 | 含义 | | ------------------------- | ---------------------- | ------------------------------------ | | **DEX_GOOGLE_ADK_MODEL** | **mock/order-support** | Gemini 模型名称，例如 **gemini-2.5-flash**。 | | **GOOGLE_API_KEY** | 无 | Gemini API key，使用真实模型时需要。 | | **DEX_AGENT_SESSION_KEY** | 随机 key | 签发 session token 的 key。 | ### 调用 session 启动一个 session，让 Agent 取消订单 1002，然后批准这次调用： **Python** ```bash base=http://127.0.0.1:8081/products/agent-harnesses/google-adk session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' -d '{}') id=$(echo "$session" | jq -r .session_id) token=$(echo "$session" | jq -r .session_token) curl -s -X POST "$base/messages" -H 'Content-Type: application/json' \ -H "X-Agent-Session-Token: $token" -d "{\"session_id\": \"$id\", \"text\": \"Cancel order 1002\"}" until interrupt=$(curl -s "$base/snapshot?session_id=$id" -H "X-Agent-Session-Token: $token" \ | jq -re '.snapshot.pending_interrupts[0].interrupt_id'); do sleep 1; done curl -s -X POST "$base/interrupts/resolve" -H 'Content-Type: application/json' \ -H "X-Agent-Session-Token: $token" \ -d "{\"session_id\": \"$id\", \"interrupt_id\": \"$interrupt\", \"approved\": true}" ``` 例子: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) 循环会反复读取 snapshot，直到 Agent 请求运行 **cancel_order**。这次调用先等待审批，批准后只运行一次。再读一次 snapshot 即可看到 Agent 的回答。 ### 在 Dex Web 中审批 也可以不执行最后一条命令，改在 Dex Web 中审批，步骤见 [在 Dex Web 中审批](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#approve-in-dex-web)。 ## 连接 Connector Agent 可以把 [Connector](https://docs.superdurable.io/zh-Hans/concepts/connector.md) 的 operation 当作 tool 调用，见 [把 Connector 用作 tool](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#connectors-as-tools)。这个 example 加入了 Slack tool **slack_post_thread_reply**，用来在 Slack thread 中回复。 ### 提供 tool 安装 **dex-agents** 的 **connectors** extra，以及 Connector 包，例如 **dex-connector-slack**。安装命令见 [Connector tools](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#install)。 Worker 从项目配置中打开 Slack connection **support-slack**。**connector_tool** 把 operation 变成 tool，由 **ConnectorToolSource** 持有。没有这个 connection 时，agent 不会得到 Slack tool。 **Python** ```python def build_slack_tool_sources(connection: slack.Connection | None) -> list[ToolSource]: """Returns the Slack tool source, or no source when Slack is not configured.""" if connection is None: return [] return [ ConnectorToolSource([ # The manifest does not ask for approval; this agent asks before every post. connector_tool( connection.post_thread_reply(), connection.reference, requires_approval=True ), ]), ] # ... async def open_slack_project_connection( environment: Mapping[str, str] = os.environ, ) -> SlackProjectConnection | None: # ... # An incomplete or invalid configuration raises here, so the Worker fails at startup. try: project = await load_from_environment(environment) except FileNotFoundError: _log_slack_not_set_up("the connector configuration file does not exist") return None # ... try: connection = slack.new_project_connection(project, SLACK_CONNECTION_NAME) except BaseException: await project.close() raise return SlackProjectConnection(project, connection) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py) Worker 在启动时打开一次 connection，每个 agent 都使用它： **Python** ```python async def main() -> None: config = AgentHarnessExamplesConfig.from_env() slack_setup = await open_slack_project_connection() try: app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None) # ... finally: if slack_setup is not None: await slack_setup.aclose() ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/\_\_main\_\_.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/__main__.py) Agent 把 tool source 和自己的 tool 一起传入： **Python** ```python return adk_agent_session_flow( agent, flow_type=GOOGLE_ADK_ORDER_SUPPORT_FLOW_TYPE, tool_sources=build_slack_tool_sources(slack_connection), # ... ) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/google_adk_order_support.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/google_adk_order_support.py) ### 调用如何运行 每次调用 **slack_post_thread_reply** 都要先经人批准，然后在自己的 **CallSingleAttemptTool** Step 中只运行一次。原因见 [Slack Connector tool](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#slack-connector-tool)。 Adapter 给 agent 树中每个 **LlmAgent** 的 tool 加上一个 **SourceToolset**。它只在 runner 的 **RunModel** 内提供 Connector tool。在其他地方，agent 照常运行。Agent 的 **before_tool_callback** 链仍然最先看到每次调用。 ### 配置 connection Worker 从哪里找到 connection **support-slack**、Slack app 需要什么，见 [Connection](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#connector-tools-connection)。 试用时，先设置 **DEX_CONNECTOR_CONFIG_FILE** 再启动 example。然后让 agent 在 Slack thread 中回复，并给出 channel ID 和 thread timestamp： ```text Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday. ``` 这次发帖和 **cancel_order** 一样，会先等待审批。 ## 限制 - App 的根必须是 **BaseAgent**。不支持以 ADK **Workflow** 为根。 - Tool 在 ADK run 之外运行，使用全新的 **ToolContext**。它对 state 和 artifact 的修改会被丢弃。请通过参数传入它需要的内容。 - 在自己的 Step 中请求确认或 credential 的 tool 会失败。 - 不调用 tool 就继续执行的 Agent，会在同一个 **RunModel** 中调用模型。 - **before_tool_callback** 在请求该调用的 **RunModel** 中运行，这个 Step 重试时会再次运行。 - 除 **BuiltInCodeExecutor** 外的 code executor 会被拒绝。请改为从 tool 中运行这类代码。 - **app:** 与 **user:** state key 只存在于当前 Dex session 中，不会跨 session 共享。 Adapter README 列出了全部限制。Server 的限制参见 [Limits for long sessions](https://docs.superdurable.io/zh-Hans/reference/agents/session-limits.md)。 ## 相关内容 - [Agent Framework](https://docs.superdurable.io/zh-Hans/agent-frameworks.md) - [Agents reference](https://docs.superdurable.io/zh-Hans/reference/agents.md) - [Python adapter README](https://github.com/superdurable/dex/tree/main/agents/python/google-adk)