Google ADK

Build Gemini ADK agents for tweet search, profiles, trends, and reviewed X writes through TwexAPI MCP.

Build a Google ADK Twitter API agent through TwexAPI’s MCP server. Search tweets, inspect profiles, read trends,. Preserve tweet IDs, cursors, and route names as durable JSON.

Why use Google ADK with TwexAPI?

ADK is Gemini-first. TwexAPI appears as a remote MCP toolset: explore discovers routes, twexapi_request executes them.

Use ADK when the runtime is Gemini. Use the Python SDK or Prefect for scheduled jobs that need no model.

Prerequisites

• Python 3.10 or later
• A TwexAPI API key
• A Google AI API key

Public X reads need no X Developer credentials. Authenticate with TwexAPI.

Install

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

TWEXAPI_API_KEY=YOUR_API_KEY
GOOGLE_API_KEY=...

Connect 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"]},
    )
)

Unauthenticated MCP requests return 401. Send x-api-key on the first request.

Full example

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

If the model wraps JSON in Markdown fences, strip them before model_validate_json. Persist the file outside the ADK session.

Preserve the MCP response contract

Reuse the same query and filters on every page. Treat each cursor as opaque.

Stop pagination when the requested total is met, has_more is false, next_cursor repeats, or the page cap is reached.

Keep a resumable agent handoff

See Agent MCP Handoff.

Build error handling

See Error Handling and Rate Limits.

Multi-agent setup

Give TwexAPI tools only to the collector. Keep analysis and writing agents tool-free so write approvals stay explicit.

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

Use dynamic headers when one ADK app serves multiple TwexAPI accounts.

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

Expose only discovery to planning agents:

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

Package versions

Next steps

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