Pydantic AI

TwexAPI MCP를 통해 타입 트윗 검색, 프로필, 팔로어 및 검토된 X 작업을 위한 Pydantic AI Twitter API 에이전트를 구축합니다.

TwexAPI MCP 서버를 통해 Pydantic AI Twitter API 에이전트를 구축합니다. 트윗을 검색하고, 프로필을 조회하며, 모든 영속 핸드오프를 Pydantic 모델로 검증합니다. 제안된 모든 게시, 답글, 좋아요, 팔로우, DM을 검토합니다.

TwexAPI와 함께 Pydantic AI를 사용하는 이유

Pydantic AI는 모델 도구 호출과 타입 Python 출력을 결합합니다. TwexAPI는 MCP 도구 explore 및 twexapi_request를 통해 Twitter API 경로를 제공합니다.

트윗 검색 또는 프로필 enrichment에는 하나의 타입 에이전트를 사용하세요. 각 에이전트는 집중적으로 유지하세요.

전제조건

• Python 3.10 이상
• TwexAPI API 키
• 도구 호출을 지원하는 Pydantic AI 호환 모델
• 쓰기 작업을 위한 Twitter 쿠키 또는 auth_token

설치

python -m pip install "pydantic-ai[mcp]" python-dotenv

소스 제어 외부에 비밀을 저장하세요.

export TWEXAPI_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"

타입 트윗 검색 에이전트 구축

에이전트 생성 전에 최종 핸드오프를 정의하세요. Pydantic AI는 이 스키마에 대해 모델 출력을 검증합니다.

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

from dotenv import load_dotenv
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP


class TweetRow(BaseModel):
    tweet_id: str
    text: str
    author_username: str | None = None
    created_at: str | 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
    stop_reason: Literal["complete", "requested_limit", "cursor_stalled"]


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

    server = MCPServerStreamableHTTP(
        "https://api.twexapi.io/mcp",
        headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    )

    agent = Agent(
        "anthropic:claude-sonnet-4-20250514",
        toolsets=[server],
        output_type=TweetSearchHandoff,
        instructions=(
            "Use TwexAPI MCP 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."
        ),
    )

    result = await agent.run(
        "Search 25 recent tweets about Pydantic AI MCP. "
        "Return the query, route, tweet rows, cursor state, "
        "and an explicit stop reason."
    )

    Path("twexapi-pydantic-ai-handoff.json").write_text(
        result.output.model_dump_json(indent=2),
        encoding="utf-8",
    )


asyncio.run(main())

에이전트는 explore와 twexapi_request를 발견합니다. 에이전트가 경로 또는 매개변수 형태를 모를 때 먼저 explore를 호출하세요.

저장 전 필드 검증

출력 모델에서 다음 소스 필드를 변경하지 마세요.

• 트윗 행: tweet_id, text, author_username, created_at, url
• 프로필 행: user_id, username, name, description, 팔로워 수
• 페이지 상태: has_more, next_cursor, 또는 언어별 커서 이름
• 쓰기 영수증: 경로 이름, 반환된 트윗 ID, 확인 상태

큰 ID를 부동소수점으로 캐스팅하지 마세요. 출력 모델에서만 소스 ID를 tweet_id 또는 user_id에 매핑하세요.

has_more가 true인 동안 빈 페이지를 계속 진행하세요. 커서가 없거나 서버가 커서를 반복하면 중지합니다. 수집된 행 수와 함께 cursor_stalled를 반환하세요.

타입 핸드오프 구축

API 키는 에이전트 출력 밖에 유지하세요. Agent MCP Handoff를 참조하세요.

MCP 연결 재사용

여러 실행이 하나의 연결을 공유해야 할 때 관련 호출을 감싸세요.

async def collect_two_search_pages(agent: Agent) -> None:
    async with agent:
        first_page = await agent.run(
            "Search 25 tweets about Pydantic AI MCP. Preserve the next cursor."
        )
        cursor = first_page.output.next_cursor
        if not first_page.output.has_more or cursor is None:
            return

        second_page = await agent.run(
            f"Continue tweet search for {first_page.output.query!r}. "
            f"Use explore, then twexapi_request with cursor {cursor!r}."
        )
        _ = second_page

X 작업 전 승인 요구

읽기 전용 에이전트는 트윗을 자동으로 검색할 수 있습니다. 쓰기 가능 에이전트는 모든 X 작업 전에 사람의 결정이 필요합니다.

class WritePlan(BaseModel):
    action: str
    endpoint: str
    preview_text: str
    requires_human_confirmation: bool = True

read_only: false 경로를 호출하기 전에 에이전트를 중지하세요. CLI --dry-run으로 페이로드를 미리보기한 다음 승인 후 REST 또는 Python SDK를 통해 실행하세요.

오류 처리

타임아웃 후 반환된 상태를 확인하지 않고 보류 중인 쓰기를 재생성하지 마세요.

Pydantic AI MCP 또는 REST 선택

패키지 버전

다음 단계

• MCP Tools
• Agent MCP Handoff
• LangChain
• Python SDK