Google ADK

Construisez des agents Gemini ADK pour la recherche de tweets, les profils, les tendances et les écritures X revues via TwexAPI MCP.

Construisez un agent Google ADK pour l’API Twitter via le serveur MCP de TwexAPI. Recherchez des tweets, inspectez des profils, lisez les tendances et révisez les actions d’écriture. Conservez les ID de tweets, les curseurs et les noms de routes en JSON durable.

Pourquoi utiliser Google ADK avec TwexAPI ?

ADK est orienté Gemini. TwexAPI apparaît comme un jeu d’outils MCP distant : explore découvre les routes, twexapi_request les exécute.

Utilisez ADK lorsque le runtime est Gemini. Utilisez le SDK Python ou Prefect pour les tâches planifiées sans modèle.

Prérequis

• Python 3.10 ou plus récent
• Une clé API TwexAPI
• Une clé API Google AI
• Un cookie Twitter ou auth_token pour les actions d’écriture — voir

Les lectures X publiques ne nécessitent pas d’identifiants X Developer. Authentifiez-vous avec TwexAPI.

Installation

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

TWEXAPI_API_KEY=YOUR_API_KEY
GOOGLE_API_KEY=...

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

Les requêtes MCP non authentifiées renvoient 401. Envoyez x-api-key sur la première requête.

Exemple complet

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

Si le modèle enveloppe le JSON dans des clôtures Markdown, supprimez-les avant model_validate_json. Persistez le fichier en dehors de la session ADK.

Préserver le contrat de réponse MCP

Réutilisez la même requête et les mêmes filtres sur chaque page. Traitez chaque curseur comme opaque.

Arrêtez la pagination lorsque le total demandé est atteint, has_more est false, next_cursor se répète ou le plafond de pages est atteint.

Conserver un transfert d’agent reprenable

Consultez Transfert MCP pour agents.

Gestion des erreurs

Consultez Gestion des erreurs et Limites de débit.

Configuration multi-agents

Donnez les outils TwexAPI uniquement au collecteur. Gardez les agents d’analyse et d’écriture sans outils pour que les approbations d’écriture restent explicites.

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

En-têtes dynamiques et filtrage d’outils

Utilisez des en-têtes dynamiques lorsqu’une application ADK sert plusieurs comptes 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,
)

Exposez uniquement la découverte aux agents de planification :

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

Versions des packages

Étapes suivantes

• Outils MCP
• Transfert MCP pour agents
• SDK Python
• Advanced Twitter Search