크루AI

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

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

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

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

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

전제조건

• 파이썬 3.10부터 3.13까지
• A TwexAPI API 키
• CrewAI에서 지원하는 LLM 공급자 키
• 쓰기 작업을 위한 Twitter 쿠키 또는 ‘auth_token’

설치

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

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

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

export TWEXAPI_API_KEY="YOUR_API_KEY"
export OPENAI_API_KEY="YOUR_OPENAI_KEY"

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

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

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를 검사하세요. 불완전한 결과를 쓰기 작업이나 내보내기에 전달하지 마세요.

집중된 쿼리로 트윗 검색

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

역할 기반 연구 인력 구축

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

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

엔드포인트 검색만 노출

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

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를 추가하면 승인된 실행이 활성화됩니다.

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

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

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 --dry-run을 사용하여 쓰기 페이로드를 미리 봅니다.

오류 및 도구 오류 처리

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

패키지 버전

다음 단계

• MCP 도구
• 에이전트 MCP 핸드오프
• 랭체인
• 고급 Twitter 검색