CrewAI

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.

Ce modèle convient à la recherche, la vérification et le reporting. Utilisez le SDK 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
• 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.

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

Stockez les secrets hors du contrôle de version.

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.

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

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.

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.

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.

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 --dry-run d’abord.

Gérer les erreurs et les échecs d’outils

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

Étapes suivantes

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