CrewAI

Build CrewAI multi-agent Twitter research crews with TwexAPI MCP for tweet search, profiles, followers, and typed handoffs.

Build a CrewAI MCP integration through TwexAPI’s remote MCP server. Give CrewAI agents controlled tweet searches, profile lookups, follower exports, and reviewed X actions. Preserve every tweet ID, profile ID, cursor, and route name.

Why use CrewAI with TwexAPI MCP?

CrewAI offers an agent framework for complex tasks. Give each agent one role. TwexAPI supplies endpoint discovery and Twitter API operations through explore and twexapi_request.

This pattern fits research, verification, and reporting. Use the Python SDK or direct REST for deterministic jobs without model decisions.

Prerequisites

• Python 3.10 through 3.13
• A TwexAPI API key
• An LLM provider key supported by CrewAI

Install

CrewAI core includes a native MCP client.

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

Store secrets outside source control.

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

Build a typed tweet search crew

Start with the expected output, then build the task. CrewAI validates the final handoff against its Pydantic model.

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

Always inspect has_tool_failures. Never pass incomplete results into write actions or exports.

Search tweets with focused queries

Use sortBy: Latest for monitoring. Use Top for engagement-ranked research. Pass next_cursor unchanged.

Build a role-based research crew

Give only the researcher access to TwexAPI MCP. Feed its validated task into a tool-free analyst.

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

Expose endpoint discovery only

Expose only explore for endpoint discovery.

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

Adding twexapi_request enables authorized execution.

Keep Twitter actions outside the research crew

Never give write permissions to autonomous research crews.

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

After approval, send one REST or SDK request. Preview write payloads with the CLI --dry-run first.

Handle errors and tool failures

Inspect result.tool_failures after failures. After 429, preserve next_cursor and completed tweet IDs.

Package versions

Next steps

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