---
title: "Google ADK"
description: "通过 TwexAPI MCP 构建 Gemini ADK AI Agent，实现推文搜索、资料查询、趋势读取与经审核的 X 写操作。"
---

通过 TwexAPI 的 MCP 服务器构建 Google ADK Twitter API AI Agent。搜索推文、查看资料、读取趋势并审核写操作。将 tweet ID、游标与路由名持久化为 durable JSON。

## 为何将 Google ADK 与 TwexAPI 搭配使用？

ADK 以 Gemini 为先。TwexAPI 作为远程 MCP 工具集出现：`explore` 发现路由，`twexapi_request` 执行。

| AI Agent任务 | TwexAPI 路由 | 为下一步保留 |
| --- | --- | --- |
| 搜索推文 | `POST /twitter/advanced_search/page` | 查询、tweet ID、作者、`created_at`、游标 |
| 查看资料 | `GET /twitter/{screen_name}/about` | 用户 ID、用户名、简介、粉丝数 |
| 读取趋势 | `GET /twitter/global-trending/tweets` | 国家、主题、推文行 |
| 发帖或回复 | `POST /twitter/tweets/create` | Tweet ID、路由、人工审批、cookie 确认 |

运行时为 Gemini 时用 ADK。无需模型的定时作业请用 [Python SDK](/sdks/python) 或 [Prefect](/guides/prefect)。

## 前置条件

- Python 3.10 或更高版本
- [TwexAPI API 密钥](https://twexapi.io/dashboard)
- Google AI API 密钥
- 写操作所需的 Twitter cookie 或 `auth_token` — 见

公开 X 读取无需 X Developer 凭证，使用 TwexAPI 认证即可。

## 安装

```bash
python -m pip install "google-adk>=1.0" python-dotenv
```

```txt .env
TWEXAPI_API_KEY=YOUR_API_KEY
GOOGLE_API_KEY=...
```

## 连接 TwexAPI MCP

```python
import os

from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

twexapi_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://api.twexapi.io/mcp",
        headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    )
)
```

未认证 MCP 请求返回 `401`。首次请求发送 `x-api-key`。

## 完整示例

```python
import asyncio
import os
from pathlib import Path
from typing import Literal

from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.runners import InMemoryRunner
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams
from google.genai import types
from pydantic import BaseModel


class TweetRow(BaseModel):
    tweet_id: str
    text: str
    author_username: str | None = None
    created_at: str | None = None


class TweetSearchHandoff(BaseModel):
    query: str
    route_used: str
    tweets: list[TweetRow]
    has_more: bool
    next_cursor: str | None = None
    stop_reason: Literal[
        "complete",
        "requested_limit",
        "cursor_stalled",
        "page_cap",
    ]


async def main() -> None:
    load_dotenv()

    twexapi_toolset = McpToolset(
        connection_params=StreamableHTTPConnectionParams(
            url="https://api.twexapi.io/mcp",
            headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
        )
    )

    agent = LlmAgent(
        model="gemini-2.5-flash",
        name="twexapi_agent",
        instruction=(
            "Use TwexAPI for Twitter API requests. Call explore before twexapi_request. "
            "Preserve exact IDs and cursors. Never invent missing tweet fields. "
            "Ask for confirmation before read_only: false actions. "
            "Return only valid JSON for TweetSearchHandoff."
        ),
        tools=[twexapi_toolset],
    )

    runner = InMemoryRunner(agent=agent, app_name="twexapi_app")
    session = await runner.session_service.create_session(
        app_name="twexapi_app",
        user_id="user-1",
    )

    response_parts: list[str] = []
    async for event in runner.run_async(
        user_id="user-1",
        session_id=session.id,
        new_message=types.Content(
            role="user",
            parts=[
                types.Part(
                    text=(
                        "Search 25 recent tweets about Google ADK MCP. "
                        "Return query, route_used, tweet rows, has_more, "
                        "next_cursor, and stop_reason as JSON."
                    )
                )
            ],
        ),
    ):
        if event.content and event.content.parts:
            response_parts.extend(
                part.text for part in event.content.parts if part.text
            )

    handoff = TweetSearchHandoff.model_validate_json("".join(response_parts))
    Path("twexapi-adk-handoff.json").write_text(
        handoff.model_dump_json(indent=2),
        encoding="utf-8",
    )
    await twexapi_toolset.close()


asyncio.run(main())
```

若模型用 Markdown 围栏包裹 JSON，`model_validate_json` 前需剥离。在 ADK 会话外持久化文件。

## 保留 MCP 响应契约

每页复用相同查询与筛选。游标视为 opaque。

满足请求总量、`has_more` 为 false、`next_cursor` 重复或达到页面上限时停止分页。

## 保持可恢复的AI Agent handoff

<CardGroup cols={2}>
  <Card title="推文分页数据" icon="message-square">
    存储 `tweet_id`、`text`、`author_username`、`created_at`、`has_more`、`next_cursor` 与原始查询。
  </Card>
  <Card title="用户资料数据" icon="user-round">
    存储 `user_id`、`username`、`name`、`description` 与粉丝数。
  </Card>
  <Card title="趋势话题榜单" icon="radio">
    存储国家、主题、内容标签与 tweet ID。
  </Card>
  <Card title="写入操作 (发推/点赞/关注)" icon="send">
    存储路由、预览文本与审批。cookie 放在密钥存储。见。
  </Card>
</CardGroup>

见 [Agent MCP Handoff](/mcp/agent-handoff)。

## 构建错误处理

| 状态 | 含义 | AI Agent决策 |
| --- | --- | --- |
| `400` | 无效路由或参数 | 重试前修正请求 |
| `401` | API 密钥缺失或无效 | 停止并更换凭证 |
| `403` | 访问被拒或额度不足 | 暂停写操作；查 [Get Balance](/api-reference/balance-endpoints/get-balance-api-balance-get) |
| `429` | 达到速率限制 | 退避后从同一游标继续 |
| `5xx` | 临时服务故障 | 对安全读取应用有界退避 |

见 [Error Handling](/guides/error-handling) 与 [Rate Limits](/guides/rate-limits)。

## 多AI Agent设置

仅向采集者提供 TwexAPI 工具。分析与写作AI Agent无工具，使写审批保持显式。

```python
researcher = LlmAgent(
    model="gemini-2.5-flash",
    name="researcher",
    instruction="Collect X/Twitter data through TwexAPI MCP and return compact JSON.",
    tools=[twexapi_toolset],
)

analyst = LlmAgent(
    model="gemini-2.5-flash",
    name="analyst",
    instruction="Analyze structured tweet rows. Do not call external tools.",
)
```

## 动态 header 与工具过滤

一个 ADK 应用服务多个 TwexAPI 账号时使用动态 header。

```python
def get_headers(context):
    return {"x-api-key": context.state["twexapi_api_key"]}


twexapi_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://api.twexapi.io/mcp",
    ),
    header_provider=get_headers,
)
```

规划AI Agent仅暴露发现：

```python
planning_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://api.twexapi.io/mcp",
        headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    ),
    tool_filter=["explore"],
)
```

## 包版本

| 包 | 支持范围 |
| --- | --- |
| Python | `>=3.10` |
| `google-adk` | `>=1.0` |
| `pydantic` | `>=2.7` |

## 下一步

- [MCP Tools](/mcp/tools)
- [Agent MCP Handoff](/mcp/agent-handoff)
-
- [Python SDK](/sdks/python)
- [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post)