--- title: "Google ADK" description: "Gemini ADK エージェント — TwexAPI MCP でツイート検索、プロフィール、トレンド、レビュー付き X 書き込み。" --- TwexAPI MCP サーバー経由で Google ADK Twitter API エージェントを構築します。ツイート検索、プロフィール確認、トレンド読み取り、書き込みレビュー。tweet ID、カーソル、ルート名を耐久 JSON として保持。 ## Google ADK と TwexAPI を組み合わせる理由 ADK は Gemini ファースト。TwexAPI はリモート MCP ツールセットとして登場: `explore` でルート探索、`twexapi_request` で実行。 | エージェントタスク | TwexAPI route | 次ステップで保持 | | --- | --- | --- | | ツイート検索 | `POST /twitter/advanced_search/page` | Query, tweet IDs, authors, `created_at`, cursor | | プロフィール確認 | `GET /twitter/{screen_name}/about` | User ID, username, biography, follower count | | トレンド読み取り | `GET /twitter/global-trending/tweets` | Country, topic, tweet rows | | 投稿または返信 | `POST /twitter/tweets/create` | Tweet ID, route, human approval, cookie confirmation | ランタイムが Gemini のとき ADK。[Python SDK](/sdks/python) または [Prefect](/guides/prefect) はモデル不要のスケジュールジョブ向け。 ## 前提条件 - Python 3.10 or later - A [TwexAPI API key](https://twexapi.io/dashboard) - A Google AI API key - A Twitter cookie or `auth_token` for write actions — see 公開 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()) ``` モデルが JSON を Markdown フェンスで包む場合、`model_validate_json` 前に除去。ファイルは ADK セッション外に永続化。 ## MCP レスポンス契約を保持 各ページで同じクエリとフィルタを再利用。各カーソルは不透明として扱う。 要求総数到達、`has_more` false、`next_cursor` 繰り返し、ページ上限到達のいずれかでページネーション停止。 ## 再開可能なエージェントハンドオフ Store `tweet_id`, `text`, `author_username`, `created_at`, `has_more`, `next_cursor`, and the original query. Store `user_id`, `username`, `name`, `description`, and follower counts. Store country, topic, content tag, and tweet IDs. Store route, preview text, and approval. Keep cookies in a secret store. See. [Agent MCP Handoff](/mcp/agent-handoff) を参照。 ## エラー処理を構築 | ステータス | 意味 | エージェント判断 | | --- | --- | --- | | `400` | 無効なルートまたはパラメータ | リトライ前にリクエスト修正 | | `401` | API キー欠落または無効 | 停止し認証情報を差し替え | | `403` | アクセス拒否またはクレジット | Pause writes; check [Get Balance](/api-reference/balance-endpoints/get-balance-api-balance-get) | | `429` | レート制限 | Back off, then resume the same cursor | | `5xx` | 一時的サーバー失敗 | 安全な読み取りに上限付きバックオフ | [Error Handling](/guides/error-handling) と [Rate Limits](/guides/rate-limits) を参照。 ## Multi-agent setup TwexAPI ツールはコレクターのみ。分析・執筆エージェントはツールなしで書き込み承認を明示的に。 ```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.", ) ``` ## Dynamic headers and tool filtering 1 ADK アプリが複数 TwexAPI アカウントを扱うとき動的ヘッダーを使用。 ```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, ) ``` 計画エージェントには探索のみ公開: ```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)