---
title: "Google ADK"
description: "TwexAPI MCP를 통해 트윗 검색, 프로필, 트렌드 및 검토된 X 쓰기 작업을 위한 Gemini ADK 에이전트를 구축합니다."
---

TwexAPI MCP 서버를 통해 Google ADK Twitter API 에이전트를 구축합니다. 트윗을 검색하고, 프로필을 조회하며, 트렌드를 읽고, 쓰기 작업을 검토합니다. 트윗 ID, 커서 및 경로 이름을 영속 JSON으로 보존합니다.

## TwexAPI와 함께 Google ADK를 사용하는 이유

ADK는 Gemini 우선 프레임워크입니다. TwexAPI는 원격 MCP 도구 세트로 나타납니다. `explore`가 경로를 발견하고, `twexapi_request`가 이를 실행합니다.

| 에이전트 작업 | TwexAPI 경로 | 다음 단계를 위해 보존 |
| --- | --- | --- |
| 트윗 검색 | `POST /twitter/advanced_search/page` | 쿼리, 트윗 ID, 작성자, `created_at`, 커서 |
| 프로필 조회 | `GET /twitter/{screen_name}/about` | 사용자 ID, 사용자명, 약력, 팔로워 수 |
| 트렌드 읽기 | `GET /twitter/global-trending/tweets` | 국가, 주제, 트윗 행 |
| 게시 또는 답글 | `POST /twitter/tweets/create` | 트윗 ID, 경로, 사람 승인, 쿠키 확인 |

런타임이 Gemini인 경우 ADK를 사용하세요. 모델이 필요 없는 예약 작업에는 [Python SDK](/sdks/python) 또는 [Prefect](/guides/prefect)를 사용하세요.

## 전제조건

- Python 3.10 이상
- [TwexAPI API 키](https://twexapi.io/dashboard)
- Google AI API 키
Public docs focus on API-key reads.

공개 X 읽기에는 X Developer 자격 증명이 필요하지 않습니다. TwexAPI로 인증하세요.

## 설치

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

```txt .env
TWEXAPI_API_KEY=YOUR_API_KEY
GOOGLE_API_KEY=...
```

## TwexAPI MCP 연결

```python
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`를 보내세요.

## 전체 예제

```python
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`가 반복되거나, 페이지 상한에 도달하면 페이지네이션을 중지합니다.

## 재개 가능한 에이전트 핸드오프 유지

<CardGroup cols={2}>
  <Card title="트윗 페이지" icon="message-square">
    `tweet_id`, `text`, `author_username`, `created_at`, `has_more`, `next_cursor`, 원본 쿼리를 저장합니다.
  </Card>
  <Card title="프로필 행" icon="user-round">
    `user_id`, `username`, `name`, `description`, 팔로워 수를 저장합니다.
  </Card>
  <Card title="트렌드 행" icon="radio">
    국가, 주제, 콘텐츠 태그, 트윗 ID를 저장합니다.
  </Card>
  <Card title="쓰기 작업" icon="send">
    경로, 미리보기 텍스트, 승인을 저장합니다. 쿠키는 비밀 저장소에 보관하세요. 참조.
  </Card>
</CardGroup>

[Agent MCP Handoff](/mcp/agent-handoff)를 참조하세요.

## 오류 처리 구축

| 상태 | 의미 | 에이전트 결정 |
| --- | --- | --- |
| `400` | 잘못된 경로 또는 매개변수 | 재시도 전에 요청을 수정합니다 |
| `401` | API 키 누락 또는 유효하지 않음 | 중지하고 자격 증명을 교체합니다 |
| `403` | 액세스 거부 또는 크레딧 | 쓰기 일시 중지; [Get Balance](/api-reference/balance-endpoints/get-balance-api-balance-get) 확인 |
| `429` | 속도 제한 도달 | 백오프 후 동일 커서로 재개합니다 |
| `5xx` | 일시적 서버 오류 | 안전한 읽기에 제한된 백오프를 적용합니다 |

[Error Handling](/guides/error-handling) 및 [Rate Limits](/guides/rate-limits)를 참조하세요.

## 멀티 에이전트 설정

TwexAPI 도구는 수집기에만 부여하세요. 분석 및 쓰기 에이전트는 도구 없이 유지하여 쓰기 승인을 명시적으로 유지합니다.

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

## 동적 헤더 및 도구 필터링

하나의 ADK 앱이 여러 TwexAPI 계정을 제공할 때 동적 헤더를 사용합니다.

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

계획 에이전트에는 발견만 노출합니다.

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

## 패키지 버전

| 패키지 | 지원 범위 |
| --- | --- |
| Python | `>=3.10` |
| `google-adk` | `>=1.0` |
| `pydantic` | `>=2.7` |

## 다음 단계

- [MCP Tools](/mcp/tools)
- [Agent MCP Handoff](/mcp/agent-handoff)
-
- [Python SDK](/sdks/python)
- [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post)