Google ADK

查看 Markdown查看 Markdown

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。

$

pip install dex-agents-google-adk

例子: agents/python/google-adk/README.md

Example

这个 example 就是每个 framework 页面共用的订单客服 Agent。

Flow type 是 GoogleADKOrderSupport（Python）。完整 example 位于 examples/python/agent-harnesses。

Agent 与 tool

Tool 是 ADK LlmAgent 上普通的 Python 函数。Tool policy 指定每个 tool。

google_adk_order_support.pyPython

1234567891011121314151617181920212223242526272829

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

Worker

一个 Worker 为 example 中所有 harness 的 session Flow 提供服务。Agent session 的 Step 是 async 的，因此 Python Registry 允许 async handler。每个启动 session 或调用其 RPC 的进程，都要构建同一个 session Flow。

app.pyPython

1234567891011121314151617181920212223242526272829303132

        # 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

Session route

每个 harness 都提供这同一组 route。这些 route 和 session token 的说明见 Session route。

http_routes.pyPython

1234567891011121314151617181920212223242526272829303132333435363738394041

    @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

在本地运行

启动 Dex 与 example

在第一个终端中，先渲染 session graph，再启动 Dex。Dex Web 启动时读取这个 graph，用它显示 session 和审批控件。

$$$$$

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

dexcli dev 会一直在前台运行。在第二个终端中启动 example：

$$

cd examples/python/agent-harnesses
uv run --frozen python -m agent_harness_examples

例子: 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，然后批准这次调用：

$$$$$$$

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

循环会反复读取 snapshot，直到 Agent 请求运行 cancel_order。这次调用先等待审批，批准后只运行一次。再读一次 snapshot 即可看到 Agent 的回答。

在 Dex Web 中审批

也可以不执行最后一条命令，改在 Dex Web 中审批，步骤见 在 Dex Web 中审批。

连接 Connector

Agent 可以把 Connector 的 operation 当作 tool 调用，见 把 Connector 用作 tool。这个 example 加入了 Slack tool slack_post_thread_reply，用来在 Slack thread 中回复。

提供 tool

安装 dex-agents 的 connectors extra，以及 Connector 包，例如 dex-connector-slack。安装命令见 Connector tools。

Worker 从项目配置中打开 Slack connection support-slack。connector_tool 把 operation 变成 tool，由 ConnectorToolSource 持有。没有这个 connection 时，agent 不会得到 Slack tool。

slack_connector_tools.pyPython

1234567891011121314151617181920212223242526272829303132

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

Worker 在启动时打开一次 connection，每个 agent 都使用它：

__main__.pyPython

123456789

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

Agent 把 tool source 和自己的 tool 一起传入：

google_adk_order_support.pyPython

123456

    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

调用如何运行

每次调用 slack_post_thread_reply 都要先经人批准，然后在自己的 CallSingleAttemptTool Step 中只运行一次。原因见 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。

试用时，先设置 DEX_CONNECTOR_CONFIG_FILE 再启动 example。然后让 agent 在 Slack thread 中回复，并给出 channel ID 和 thread timestamp：

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。

相关内容

• Agent Framework
• Agents reference
• Python adapter README