Google ADK

Gemini ADK エージェント — TwexAPI MCP でツイート検索、プロフィール、トレンド、レビュー付き X 書き込み。

TwexAPI MCP サーバー経由で Google ADK Twitter API エージェントを構築します。ツイート検索、プロフィール確認、トレンド読み取り、書き込みレビュー。tweet ID、カーソル、ルート名を耐久 JSON として保持。

Google ADK と TwexAPI を組み合わせる理由

ADK は Gemini ファースト。TwexAPI はリモート MCP ツールセットとして登場: explore でルート探索、twexapi_request で実行。

ランタイムが Gemini のとき ADK。Python SDK または Prefect はモデル不要のスケジュールジョブ向け。

前提条件

• Python 3.10 or later
• A TwexAPI API key
• A Google AI API key
• A Twitter cookie or auth_token for write actions — see

公開 X 読み取りに X Developer 認証情報は不要。TwexAPI で認証します。

インストール

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

TWEXAPI_API_KEY=YOUR_API_KEY
GOOGLE_API_KEY=...

TwexAPI MCP に接続

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 を送信。

完全例

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 繰り返し、ページ上限到達のいずれかでページネーション停止。

再開可能なエージェントハンドオフ

Agent MCP Handoff を参照。

エラー処理を構築

Error Handling と Rate Limits を参照。

Multi-agent setup

TwexAPI ツールはコレクターのみ。分析・執筆エージェントはツールなしで書き込み承認を明示的に。

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 アカウントを扱うとき動的ヘッダーを使用。

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,
)

計画エージェントには探索のみ公開:

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

パッケージバージョン

次のステップ

• MCP Tools
• Agent MCP Handoff
• Python SDK
• Advanced Twitter Search