CrewAI

CrewAI マルチエージェントリサーチクルーと TwexAPI MCP — ツイート検索、プロフィール、フォロワー、型付きハンドオフ。

TwexAPI リモート MCP サーバー経由で CrewAI MCP 統合を構築します。CrewAI エージェントに制御されたツイート検索、プロフィール参照、フォロワーエクスポート、レビュー付き X アクションを提供。tweet ID、プロフィール ID、カーソル、ルート名をすべて保持してください。

CrewAI と TwexAPI MCP を組み合わせる理由

CrewAI は複雑タスク向けエージェントフレームワークです。各エージェントに 1 ロールを割り当て。TwexAPI は explore と twexapi_request でエンドポイント探索と Twitter API 操作を提供します。

リサーチ、検証、レポートに適したパターンです。モデル判断不要の決定的ジョブは Python SDK または直接 REST を使用。

前提条件

• Python 3.10 through 3.13
• A TwexAPI API key
• An LLM provider key supported by CrewAI
• A Twitter cookie or auth_token for write actions

インストール

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 を確認。不完全結果を書き込みやエクスポートに渡さない。

ツイート検索 with focused queries

Use sortBy: Latest for monitoring. Use Top for engagement-ranked research. Pass next_cursor unchanged.

Build a role-based research crew

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

Expose endpoint discovery only

explore のみ公開してエンドポイント探索。

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

Adding twexapi_request enables authorized execution.

Keep Twitter actions outside the research crew

自律リサーチクルーに書き込み権限を与えない。

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 を 1 回送信。先に CLI --dry-run で書き込みペイロードをプレビュー。

エラーとツール失敗の処理

Inspect result.tool_failures after failures. After 429, preserve next_cursor and completed tweet IDs.

パッケージバージョン

次のステップ

• MCP Tools
• Agent MCP Handoff
• LangChain
• Advanced Twitter Search