Google ADK

TwexAPI MCP aracılığıyla tweet araması, profiller, trendler ve incelenen X yazıları için Gemini ADK aracıları oluşturun.

TwexAPI’nin MCP sunucusu aracılığıyla bir Google ADK Twitter API aracısı oluşturun. Tweet’leri arayın, profilleri inceleyin, eğilimleri okuyun ve yazma işlemlerini inceleyin. Tweet kimliklerini, imleçleri ve rota adlarını dayanıklı JSON olarak koruyun.

Neden Google ADK’yı TwexAPI ile kullanmalısınız?

ADK, Gemini’nin ilkidir. TwexAPI uzak bir MCP araç seti olarak görünür: ‘explore’ rotaları keşfeder, twexapi_request onları yürütür.

Çalışma zamanı Gemini olduğunda ADK’yı kullanın. Model gerektirmeyen zamanlanmış işler için Python SDK veya Prefect kullanın.

Önkoşullar

• Python 3.10 veya üzeri
• Bir TwexAPI API anahtarı
• Bir Google AI API anahtarı Public docs focus on API-key reads.

Genel X okumalarının X Developer kimlik bilgilerine ihtiyacı yoktur. TwexAPI ile kimlik doğrulaması yapın.

Kurulum

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

TWEXAPI_API_KEY=YOUR_API_KEY
GOOGLE_API_KEY=...

TwexAPI MCP’ye bağlanın

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

Kimliği doğrulanmamış MCP istekleri 401 döndürür. İlk istekte x-api-key gönderin.

Tam örnek

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

Model JSON’u Markdown fence içine alırsa, model_validate_json öncesinde kaldırın. Dosyayı ADK oturumunun dışında kalıcı hale getirin.

MCP yanıt sözleşmesini koruyun

Her sayfada aynı sorgu ve filtreleri yeniden kullanın. Her imleci opak kabul edin.

İstenen toplam karşılandığında, has_more false olduğunda, next_cursor tekrarlandığında veya sayfa sınırına ulaşıldığında sayfalandırmayı durdurun.

Devam ettirilebilir agent devrini sürdürün

Bkz. Agent MCP Handoff.

Hata işleme oluşturun

Bkz. Error Handling ve Rate Limits.

Çoklu agent kurulumu

TwexAPI araçlarını yalnızca toplayıcıya verin. Analiz ve yazma agent’larını araçsız tutun; böylece yazma onayları açık kalır.

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

Dinamik başlıklar ve araç filtreleme

Bir ADK uygulaması birden fazla TwexAPI hesabına hizmet verdiğinde dinamik başlıklar kullanın.

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

Planlama agent’larına yalnızca keşfi açın:

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

Paket versiyonları

Sonraki adımlar

• MCP Araçları
• Ajan MCP Aktarımı
• Python SDK’sı
• Gelişmiş Twitter Araması