CrewAI

Crea crews de investigación multiagente de Twitter con CrewAI y TwexAPI MCP para búsqueda de tweets, perfiles, seguidores y entregas tipadas.

Crea una integración MCP de CrewAI a través del servidor MCP remoto de TwexAPI. Dale a los agentes de CrewAI búsquedas controladas de tweets, consultas de perfiles, exportaciones de seguidores y acciones en X revisadas. Conserva cada ID de tweet, ID de perfil, cursor y nombre de ruta.

¿Por qué usar CrewAI con TwexAPI MCP?

CrewAI ofrece un framework de agentes para tareas complejas. Asigna un rol a cada agente. TwexAPI proporciona descubrimiento de endpoints y operaciones de Twitter API a través de explore y twexapi_request.

Este patrón encaja con investigación, verificación e informes. Usa el Python SDK o REST directo para trabajos deterministas sin decisiones del modelo.

Requisitos previos

• Python 3.10 a 3.13
• Una API key de TwexAPI
• Una API key de proveedor LLM compatible con CrewAI
• Una cookie de Twitter o auth_token para acciones de escritura

Instalación

CrewAI core incluye un cliente MCP nativo.

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

Guarda los secretos fuera del control de versiones.

export TWEXAPI_API_KEY="YOUR_API_KEY"
export OPENAI_API_KEY="YOUR_OPENAI_KEY"

Construir un crew tipado de búsqueda de tweets

Empieza con la salida esperada y luego construye la tarea. CrewAI valida la entrega final contra su modelo 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",
)

Siempre inspecciona has_tool_failures. Nunca pases resultados incompletos a acciones de escritura o exportaciones.

Buscar tweets con consultas enfocadas

Usa sortBy: Latest para monitoreo. Usa Top para investigación ordenada por engagement. Pasa next_cursor sin cambios.

Construir un crew de investigación basado en roles

Dale solo al investigador acceso a TwexAPI MCP. Alimenta su tarea validada a un analista sin herramientas.

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

Exponer solo descubrimiento de endpoints

Expone solo explore para descubrimiento de 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,
)

Añadir twexapi_request habilita la ejecución autorizada.

Mantener acciones de Twitter fuera del crew de investigación

Nunca des permisos de escritura a crews de investigación autónomos.

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

Después de la aprobación, envía una solicitud REST o SDK. Previsualiza cargas de escritura con el CLI --dry-run primero.

Manejar errores y fallos de herramientas

Inspecciona result.tool_failures después de fallos. Tras 429, conserva next_cursor y los IDs de tweet completados.

Versiones de paquetes

Próximos pasos

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