---
title: "Pydantic AI"
description: "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 경로를 제공합니다.

| 경계 | Pydantic AI 제어 | Twitter 에이전트 이점 |
| -------------------- | ---------------------- | -------------------------------------------- |
| MCP 연결 | `MCPServerStreamableHTTP` | 프로세스 내에서 자격 증명 유지 |
| 최종 응답 | Pydantic `output_type` | 잘못된 트윗 행 및 커서 거부 |
| 쓰기 작업 | 사람 검토 단계 | 게시, 답글, 팔로우 전 일시 중지 |
| 연결 수명 주기 | `async with agent` | 관련 호출 간 하나의 MCP 세션 재사용 |

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

## 전제조건

* Python 3.10 이상
* [TwexAPI API 키](https://twexapi.io/dashboard)
* 도구 호출을 지원하는 Pydantic AI 호환 모델
* 쓰기 작업을 위한 Twitter 쿠키 또는 `auth_token`

## 설치

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

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

```bash
export TWEXAPI_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"
```

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

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

```python
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`를 반환하세요.

## 타입 핸드오프 구축

<CardGroup cols={2}>
  <Card title="트윗 검색" icon="search">
    쿼리, 경로, 트윗 ID, 작성자, URL, `has_more`, `next_cursor`, 중지 이유를 저장합니다.
  </Card>

  <Card title="프로필 조회" icon="user-round">
    `user_id`, `username`, `name`, `description`, 팔로워 수를 저장합니다.
  </Card>

  <Card title="팔로워 페이지" icon="users">
    소스 사용자명, 팔로워 행, 커서 체크포인트를 저장합니다.
  </Card>

  <Card title="쓰기 작업" icon="send">
    경로, 미리보기 텍스트, 쿠키 요구 사항, 승인 기록을 저장합니다.
  </Card>
</CardGroup>

API 키는 에이전트 출력 밖에 유지하세요. [Agent MCP Handoff](/mcp/agent-handoff)를 참조하세요.

## MCP 연결 재사용

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

```python
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 작업 전에 사람의 결정이 필요합니다.

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

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

## 오류 처리

| 상태 | Pydantic AI 결정 |
| ------ | ---------------------------------------------- |
| `400` | 재시도 전에 요청을 수정합니다 |
| `401` | 중지하고 자격 증명을 교체합니다 |
| `403` | 액세스 또는 크레딧 문제를 보고합니다 |
| `429` | 백오프하고 커서를 보존합니다 |
| `5xx` | 제한된 백오프로 안전한 읽기를 재시도합니다 |

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

## Pydantic AI MCP 또는 REST 선택

| 요구 사항 | 선택 | 이유 |
| ------------------------------------------------ | --------------- | --------------------------------------------------- |
| 모델이 트윗 또는 프로필 작업을 선택 | Pydantic AI MCP | 에이전트가 `explore`로 경로를 발견 |
| 애플리케이션 코드가 하나의 알려진 경로 호출 | [Python SDK](/sdks/python) | 요청이 결정적 |
| 사람이 X 작업을 검토해야 함 | Pydantic AI MCP + 수동 REST | 실행 전 일시 중지 |
| 모델 없이 예약 내보내기 실행 | [Prefect](/guides/prefect) 또는 REST | 모델 결정 불필요 |

## 패키지 버전

| 패키지 | 지원 범위 |
| ------------------ | --------------- |
| Python | `>=3.10` |
| `pydantic-ai` | `>=0.8` |
| `pydantic` | `>=2.7` |

## 다음 단계

* [MCP Tools](/mcp/tools)
* [Agent MCP Handoff](/mcp/agent-handoff)
* [LangChain](/guides/langchain)
* [Python SDK](/sdks/python)