---
title: "크루AI"
description: "트윗 검색, 프로필, 팔로어 및 입력된 핸드오프를 위해 TwexAPI MCP를 사용하여 CrewAI 다중 에이전트 Twitter 연구 팀을 구축하세요."
---

TwexAPI의 원격 MCP 서버를 통해 CrewAI MCP 통합을 구축하세요. CrewAI 에이전트에게 트윗 검색, 프로필 조회, 팔로어 내보내기 및 검토된 X 작업을 제어할 수 있는 기능을 제공합니다. 모든 트윗 ID, 프로필 ID, 커서 및 경로 이름을 유지합니다.

## TwexAPI MCP와 함께 CrewAI를 사용하는 이유는 무엇입니까?

CrewAI는 복잡한 작업을 위한 에이전트 프레임워크를 제공합니다. 각 상담원에게 하나의 역할을 부여하세요. TwexAPI는 'explore' 및 'twexapi_request'를 통해 엔드포인트 검색 및 Twitter API 작업을 제공합니다.

| 경계 | CrewAI 제어 | 혜택 |
| ---------- | -------------------------- | -------------------------------------------------- |
| 원격 MCP | `MCPServerHTTP` | 트윗, 프로필, 팔로어 및 트렌드 경로에 도달 |
| 핸드오프 | 피단틱 `output_pydantic` | 잘못된 트윗과 커서 거부 |
| 순서 | `Process.sequential` | 전문가 간에 정확한 트윗 전달 |
| 발견 | 정적 도구 필터 | 실행 없이 '탐색' 노출 |
| 검토 | 도구가 필요 없는 작업 | 쓰기 전에 X 작업 검토 |
| 실패 | `has_tool_failures` | 불완전한 연구를 중단하세요 |

이 패턴은 연구, 검증 및 보고에 적합합니다. 모델 결정 없이 결정적 작업을 수행하려면 [Python SDK](/sdks/python) 또는 직접 REST를 사용하세요.

## 전제조건

* 파이썬 3.10부터 3.13까지
* A [TwexAPI API 키](https://twexapi.io/dashboard)
* CrewAI에서 지원하는 LLM 공급자 키
* 쓰기 작업을 위한 Twitter 쿠키 또는 'auth_token'

## 설치

CrewAI 코어에는 기본 MCP 클라이언트가 포함되어 있습니다.

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

소스 제어 외부에 비밀을 저장합니다.

```bash
export TWEXAPI_API_KEY="YOUR_API_KEY"
export OPENAI_API_KEY="YOUR_OPENAI_KEY"
```

## 타이핑된 트윗 검색 크루를 구성하세요

예상되는 출력으로 시작한 다음 작업을 빌드합니다. CrewAI는 Pydantic 모델에 대한 최종 핸드오프를 검증합니다.

```python
import os
from pathlib import Path
from typing import Literal

from crewai import Agent, Crew, Process, Task
from crewai.mcp import MCPServerHTTP
from pydantic import BaseModel


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
    pages_fetched: int
    stop_reason: Literal["complete", "requested_limit", "cursor_stalled"]


twexapi_mcp = MCPServerHTTP(
    url="https://api.twexapi.io/mcp",
    headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    streamable=True,
    cache_tools_list=True,
)

researcher = Agent(
    role="Twitter API Researcher",
    goal="Return exact tweet records and resumable pagination state",
    backstory=(
        "You inspect Twitter conversations through TwexAPI MCP. "
        "Call explore before twexapi_request. "
        "You preserve source IDs and never invent missing fields."
    ),
    llm="openai/gpt-4o",
    mcps=[twexapi_mcp],
    allow_delegation=False,
    verbose=False,
)

search_task = Task(
    description=(
        "Use TwexAPI MCP to search 50 latest tweets about CrewAI Twitter MCP. "
        "Call explore first. Use POST /twitter/advanced_search/page. "
        "Preserve exact tweet IDs, created timestamps, and cursors. "
        "Stop at the requested limit. Stop if a cursor repeats."
    ),
    expected_output="A validated tweet search handoff with pagination state.",
    agent=researcher,
    output_pydantic=TweetSearchHandoff,
)

crew = Crew(
    agents=[researcher],
    tasks=[search_task],
    process=Process.sequential,
    verbose=False,
)

result = crew.kickoff()
if result.has_tool_failures:
    raise RuntimeError("TwexAPI MCP tool failed. Inspect result.tool_failures.")

handoff = TweetSearchHandoff.model_validate(result.to_dict())
Path("twexapi-crewai-handoff.json").write_text(
    handoff.model_dump_json(indent=2),
    encoding="utf-8",
)
```

항상 `has_tool_failures`를 검사하세요. 불완전한 결과를 쓰기 작업이나 내보내기에 전달하지 마세요.

## 집중된 쿼리로 트윗 검색

| 의지 | 예시 쿼리 |
| ------------------------ | -------------------------------------------------- |
| 프레임워크 게시물 | `"CrewAI" MCP` |
| 계정 타임라인 | `from:crewAIInc since:2026-07-01 until:2026-08-01` |
| 해시태그 검색 | `#crewai #agents lang:en` |
| 재게시물 제외 | `"multi-agent workflow" -filter:retweets` |

모니터링하려면 `sortBy: 최신`을 사용하세요. 참여도 순위 조사에는 '상위'를 사용하세요. 'next_cursor'를 변경하지 않고 전달하세요.

## 역할 기반 연구 인력 구축

연구원에게만 TwexAPI MCP에 대한 액세스 권한을 부여하십시오. 도구가 필요 없는 분석가에게 검증된 작업을 제공합니다.

```python
class TweetAnalysis(BaseModel):
    query: str
    analyzed_tweet_ids: list[str]
    recurring_topics: list[str]
    top_author_usernames: list[str]
    next_cursor: str | None = None

analyst = Agent(
    role="Tweet Conversation Analyst",
    goal="Analyze only the supplied tweet rows",
    backstory="You compare exact tweets without fetching extra records.",
    llm="openai/gpt-4o",
    allow_delegation=False,
)

analysis_task = Task(
    description=(
        "Analyze the supplied tweet rows. "
        "Keep every analyzed tweet_id. Preserve the next_cursor."
    ),
    expected_output="A typed topic analysis tied to source tweet IDs.",
    agent=analyst,
    context=[search_task],
    output_pydantic=TweetAnalysis,
)

research_crew = Crew(
    agents=[researcher, analyst],
    tasks=[search_task, analysis_task],
    process=Process.sequential,
)
```

## 엔드포인트 검색만 노출

엔드포인트 검색을 위해 '탐색'만 노출합니다.

```python
from crewai.mcp import MCPServerHTTP
from crewai.mcp.filters import create_static_tool_filter

discovery_mcp = MCPServerHTTP(
    url="https://api.twexapi.io/mcp",
    headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    tool_filter=create_static_tool_filter(
        allowed_tool_names=["explore"],
    ),
    cache_tools_list=True,
)
```

`twexapi_request`를 추가하면 승인된 실행이 활성화됩니다.

## 트위터 활동을 연구진 외부로 유지

자율연구진에게 쓰기 권한을 절대 부여하지 마세요.

```python
class TweetWritePlan(BaseModel):
    tweet_content: str
    reply_to_tweet_id: str | None = None
    media_urls: list[str]
    requires_human_confirmation: bool = True

planner = Agent(
    role="Twitter Action Planner",
    goal="Prepare one reviewable X action without executing it",
    backstory="You preserve approved text, target IDs, and route names.",
    llm="openai/gpt-4o",
    tools=[],
    allow_delegation=False,
)

plan_task = Task(
    description="Prepare a tweet or reply plan from reviewed source tweets.",
    expected_output="One typed action plan. Do not execute any X request.",
    agent=planner,
    output_pydantic=TweetWritePlan,
    human_input=True,
)
```

승인 후 하나의 REST 또는 SDK 요청을 보냅니다. 먼저 [CLI](/sdks/cli) `--dry-run`을 사용하여 쓰기 페이로드를 미리 봅니다.

## 오류 및 도구 오류 처리

| 상태 | 의미 | 승무원 행동 |
| ------ | --------------------------- | --------------------------------- |
| `400` | 누락되거나 잘못된 매개변수 | 요청을 수정하세요. 변경하지 않고 다시 시도하지 마세요. |
| `401` | 인증 실패 | API 키 확인 |
| `403` | 접근 불가 | 중지 및 계정 조치 요청 |
| `429` | 비율 제한이 적용됩니다. | 잠시 기다린 후 커서를 다시 시작하세요. |
| `5xx` | 서버 장애 | ID를 변경하지 않고 나중에 다시 시도하세요. |

실패 후 `result.tool_failures`를 검사하세요. '429' 이후에는 'next_cursor'와 완료된 트윗 ID를 유지합니다.

## 패키지 버전

| 패키지 | 호환 범위 |
| ---------- | ---------------- |
| `crewai` | `>=1.0` |
| `pydantic` | `>=2.7` |

## 다음 단계

* [MCP 도구](/mcp/tools)
* [에이전트 MCP 핸드오프](/mcp/agent-handoff)
* [랭체인](/guides/langchain)
* [고급 Twitter 검색](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post)