Google ADK MCP Twitter API tutorial for Python

python -m pip install --upgrade \
  "google-adk[mcp]>=2.6,<2.7" \
  "pydantic>=2.12,<3" \
  python-dotenv

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

from dotenv import load_dotenv
from google.adk import Agent
from google.adk.runners import InMemoryRunner
from google.adk.tools.mcp_tool.mcp_session_manager import (
    StreamableHTTPConnectionParams,
)
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.genai import types
from pydantic import BaseModel


class TweetRow(BaseModel):
    tweet_id: str
    text: str
    author_username: str | None = None
    created: int | None = None
    url: str | None = None


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


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

    xquik_tools = McpToolset(
        connection_params=StreamableHTTPConnectionParams(
            url="https://xquik.com/mcp",
            headers={"x-api-key": os.environ["XQUIK_API_KEY"]},
            timeout=15,
            sse_read_timeout=120,
        ),
        tool_filter=["explore", "xquik"],
    )

    agent = Agent(
        model="gemini-3.5-flash",
        name="xquik_tweet_search_agent",
        instruction=(
            "Use Xquik for Twitter API requests. "
            "Use GET /api/v1/x/tweets/search. "
            "Preserve exact tweet IDs, created timestamps, and cursors. "
            "Never invent missing tweet fields. "
            "Stop when the requested limit is reached. "
            "Stop if the server repeats a cursor."
        ),
        tools=[xquik_tools],
        output_schema=TweetSearchHandoff,
    )

    async with InMemoryRunner(
        agent=agent,
        app_name="xquik_adk_app",
    ) as runner:
        session = await runner.session_service.create_session(
            app_name="xquik_adk_app",
            user_id="user-1",
        )

        final_text = ""
        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 50 latest tweets about Google ADK MCP. "
                            "Return compact validated JSON."
                        )
                    )
                ],
            ),
        ):
            if (
                event.is_final_response()
                and event.content
                and event.content.parts
            ):
                final_text = "".join(
                    part.text or "" for part in event.content.parts
                )

        handoff = TweetSearchHandoff.model_validate_json(final_text)
        Path("xquik-adk-handoff.json").write_text(
            handoff.model_dump_json(indent=2),
            encoding="utf-8",
        )


asyncio.run(main())

import os

from google.adk import Agent
from google.adk.tools.mcp_tool.mcp_session_manager import (
    StreamableHTTPConnectionParams,
)
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset


def connection() -> StreamableHTTPConnectionParams:
    return StreamableHTTPConnectionParams(
        url="https://xquik.com/mcp",
        headers={"x-api-key": os.environ["XQUIK_API_KEY"]},
    )


research_tools = McpToolset(
    connection_params=connection(),
    tool_filter=["explore", "xquik"],
)

approved_write_tools = McpToolset(
    connection_params=connection(),
    tool_filter=["xquik"],
    require_confirmation=True,
)

researcher = Agent(
    name="tweet_researcher",
    model="gemini-3.5-flash",
    instruction=(
        "Search tweets and profiles. Preserve IDs, created timestamps, "
        "has_more, and next_cursor. Never perform X write actions."
    ),
    tools=[research_tools],
    output_key="tweet_research",
)

publisher = Agent(
    name="approved_x_publisher",
    model="gemini-3.5-flash",
    instruction=(
        "Use only reviewed tweet text and the selected X account. "
        "Never change approved text. Never retry a pending write."
    ),
    tools=[approved_write_tools],
)

coordinator = Agent(
    name="twitter_campaign_coordinator",
    model="gemini-3.5-flash",
    instruction=(
        "Delegate tweet research to tweet_researcher. "
        "Delegate approved X actions to approved_x_publisher."
    ),
    sub_agents=[researcher, publisher],
)

discovery_tools = McpToolset(
    connection_params=connection(),
    tool_filter=["explore"],
)

from google.adk.agents.readonly_context import ReadonlyContext


def tenant_headers(context: ReadonlyContext) -> dict[str, str]:
    tenant_id = str(context.state["tenant_id"])
    api_key = secret_store.get_xquik_key(tenant_id)
    return {"x-api-key": api_key}


tenant_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://xquik.com/mcp",
    ),
    header_provider=tenant_headers,
    tool_filter=["explore", "xquik"],
)