---
title: "CrewAI"
description: "通过 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 操作。

| 边界 | CrewAI 控制 | 收益 |
| ---------- | -------------------------- | -------------------------------------------------- |
| 远程 MCP | `MCPServerHTTP` | 访问推文、资料、粉丝与趋势路由 |
| Handoff | Pydantic `output_pydantic` | 拒绝畸形推文与游标 |
| 顺序 | `Process.sequential` | 在专家之间传递精确推文 |
| 发现 | 静态工具过滤 | 暴露 `explore` 而不执行 |
| 审核 | 无工具任务 | 写操作前审核 X 操作 |
| 失败 | `has_tool_failures` | 阻止不完整研究 |

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

## 前置条件

* Python 3.10 至 3.13
* [TwexAPI API 密钥](https://twexapi.io/dashboard)
* CrewAI 支持的 LLM 提供商密钥
* 写操作所需的 Twitter cookie 或 `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"
```

## 构建类型化推文搜索 crew

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

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

## 构建基于角色的研究 crew

仅让研究员访问 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,
)
```

## 仅暴露端点发现

端点发现仅暴露 `explore`。

```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` 即可启用授权执行。

## 将 Twitter 操作与研究 crew 分离

切勿向自主研究 crew 授予写权限。

```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` 预览。

## 处理错误与工具失败

| 状态 | 含义 | Crew 操作 |
| ------ | --------------------------- | --------------------------------- |
| `400` | 参数缺失或无效 | 修正请求；切勿原样重试 |
| `401` | 认证失败 | 检查 API 密钥 |
| `403` | 访问被拒 | 停止并请求账号处理 |
| `429` | 速率限制 | 等待后从游标继续 |
| `5xx` | 服务故障 | 稍后重试且不改变 ID |

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

## 包版本

| 包 | 兼容范围 |
| ---------- | ---------------- |
| `crewai` | `>=1.0` |
| `pydantic` | `>=2.7` |

## 下一步

* [MCP Tools](/mcp/tools)
* [Agent MCP Handoff](/mcp/agent-handoff)
* [LangChain](/guides/langchain)
* [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post)