---
title: "CrewAI"
description: "Construisez des équipes CrewAI de recherche Twitter avec TwexAPI MCP pour la recherche de tweets, les profils, les abonnés et les handoffs typés."
---

Construisez une intégration CrewAI MCP via le serveur MCP distant de TwexAPI. Donnez aux agents CrewAI des recherches de tweets contrôlées, des consultations de profils, des exports d'abonnés et des actions X revues. Conservez chaque ID de tweet, ID de profil, curseur et nom de route.

## Pourquoi utiliser CrewAI avec TwexAPI MCP ?

CrewAI propose un framework d'agents pour les tâches complexes. Donnez à chaque agent un seul rôle. TwexAPI fournit la découverte d'endpoints et les opérations de l'API Twitter via `explore` et `twexapi_request`.

| Limite | Contrôle CrewAI | Avantage |
| ---------- | -------------------------- | -------------------------------------------------- |
| MCP distant | `MCPServerHTTP` | Atteindre les routes tweets, profils, abonnés et tendances |
| Handoff | `output_pydantic` Pydantic | Rejeter les tweets et curseurs mal formés |
| Séquence | `Process.sequential` | Passer les tweets exacts entre spécialistes |
| Découverte | Filtre d'outils statique | Exposer `explore` sans exécution |
| Revue | Tâche sans outils | Revoir les actions X avant les écritures |
| Échecs | `has_tool_failures` | Arrêter la recherche incomplète |

Ce modèle convient à la recherche, la vérification et le reporting. Utilisez le [SDK Python](/sdks/python) ou REST direct pour les tâches déterministes sans décisions du modèle.

## Prérequis

* Python 3.10 à 3.13
* Une [clé API TwexAPI](https://twexapi.io/dashboard)
* Une clé de fournisseur LLM prise en charge par CrewAI
* Un cookie Twitter ou `auth_token` pour les actions d'écriture

## Installation

Le cœur de CrewAI inclut un client MCP natif.

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

Stockez les secrets hors du contrôle de version.

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

## Construire une équipe de recherche de tweets typée

Commencez par la sortie attendue, puis construisez la tâche. CrewAI valide le handoff final contre son modèle 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",
)
```

Inspectez toujours `has_tool_failures`. Ne transmettez jamais de résultats incomplets aux actions d'écriture ou aux exports.

## Rechercher des tweets avec des requêtes ciblées

| Intention | Exemple de requête |
| ------------------------ | -------------------------------------------------- |
| Publications framework | `"CrewAI" MCP` |
| Timeline de compte | `from:crewAIInc since:2026-07-01 until:2026-08-01` |
| Recherche hashtag | `#crewai #agents lang:en` |
| Exclure les reposts | `"multi-agent workflow" -filter:retweets` |

Utilisez `sortBy: Latest` pour le monitoring. Utilisez `Top` pour la recherche classée par engagement. Passez `next_cursor` sans modification.

## Construire une équipe de recherche par rôles

Donnez uniquement au chercheur l'accès à TwexAPI MCP. Alimentez sa tâche validée dans un analyste sans outils.

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

## Exposer uniquement la découverte d'endpoints

Exposez uniquement `explore` pour la découverte d'endpoints.

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

Ajouter `twexapi_request` active l'exécution autorisée.

## Garder les actions Twitter hors de l'équipe de recherche

Ne donnez jamais de permissions d'écriture aux équipes de recherche autonomes.

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

Après approbation, envoyez une seule requête REST ou SDK. Prévisualisez les payloads d'écriture avec le [CLI](/sdks/cli) `--dry-run` d'abord.

## Gérer les erreurs et les échecs d'outils

| Statut | Signification | Action Crew |
| ------ | --------------------------- | --------------------------------- |
| `400` | Paramètres manquants ou invalides | Corrigez la requête ; ne jamais réessayer sans modification |
| `401` | Échec d'authentification | Vérifiez la clé API |
| `403` | Accès refusé | Arrêtez et demandez une action sur le compte |
| `429` | Limite de débit atteinte | Attendez, puis reprenez le curseur |
| `5xx` | Défaillance serveur | Réessayez plus tard sans modifier les ID |

Inspectez `result.tool_failures` après les échecs. Après `429`, conservez `next_cursor` et les ID de tweets complétés.

## Versions des packages

| Package | Plage compatible |
| ---------- | ---------------- |
| `crewai` | `>=1.0` |
| `pydantic` | `>=2.7` |

## Étapes suivantes

* [Outils MCP](/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)