CrewAI

通过 TwexAPI MCP 构建 CrewAI 多AI Agent Twitter 研究 crew，实现推文搜索、资料查询、粉丝导出与类型化 handoff。

通过 TwexAPI 远程 MCP 服务器构建 CrewAI MCP 集成。为 CrewAI Agent提供受控推文搜索、资料查询、粉丝导出与经审核的 X 操作。保留每条 tweet ID、资料 ID、游标与路由名。

为何将 CrewAI 与 TwexAPI MCP 搭配使用？

CrewAI 提供面向复杂任务的AI Agent框架。每个AI Agent单一职责。TwexAPI 通过 explore 与 twexapi_request 提供端点发现与 Twitter API 操作。

该模式适合研究、验证与报告。无需模型决策的确定性作业请用 Python SDK 或直接 REST。

前置条件

• Python 3.10 至 3.13
• TwexAPI API 密钥
• CrewAI 支持的 LLM 提供商密钥
• 写操作所需的 Twitter cookie 或 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"

构建类型化推文搜索 crew

先定义期望输出，再构建任务。CrewAI 根据 Pydantic 模型校验最终 handoff。

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: Latest；按互动排序的研究用 Top。原样传递 next_cursor。

构建基于角色的研究 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,
)

仅暴露端点发现

端点发现仅暴露 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,
)

加入 twexapi_request 即可启用授权执行。

将 Twitter 操作与研究 crew 分离

切勿向自主研究 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 请求。写载荷请先用 CLI --dry-run 预览。

处理错误与工具失败

失败后检查 result.tool_failures。429 后保留 next_cursor 与已完成的 tweet ID。

包版本

下一步

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