CrewAI

pip install crewai

from pathlib import Path
from crewai import Agent, Task, Crew, Process
from crewai.mcp import MCPServerHTTP

researcher = Agent(
    role="X Research Analyst",
    goal="Find and analyze trending conversations on X",
    backstory="Expert social media analyst with deep experience in trend analysis",
    mcps=[
        MCPServerHTTP(
            url="https://xquik.com/mcp",
            headers={"x-api-key": "xq_YOUR_KEY_HERE"},
        ),
    ],
)

task = Task(
    description="Search for the latest tweets about AI agents. Return compact JSON with query, route_used, tweets[{tweet_id,text,author_username,created_at}], has_more, next_cursor, and key influencers.",
    expected_output="Compact JSON with tweet rows, pagination state, and influencer notes",
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
Path("xquik-crewai-handoff.json").write_text(str(result), encoding="utf-8")

from pathlib import Path
from crewai import Agent, Task, Crew, Process
from crewai.mcp import MCPServerHTTP

xquik_mcp = MCPServerHTTP(
    url="https://xquik.com/mcp",
    headers={"x-api-key": "xq_YOUR_KEY_HERE"},
)

researcher = Agent(
    role="X Researcher",
    goal="Gather compact JSON handoff rows from X about a given topic",
    backstory="Data collection specialist who finds relevant tweets and profiles",
    mcps=[xquik_mcp],
)

analyst = Agent(
    role="Engagement Analyst",
    goal="Analyze engagement patterns and identify trends",
    backstory="Data analyst who turns structured social data into actionable insights",
    mcps=[xquik_mcp],
)

writer = Agent(
    role="Report Writer",
    goal="Write a concise executive summary from the analysis",
    backstory="Technical writer who creates clear, data-driven reports",
)

research_task = Task(
    description="Search X for tweets about '{topic}' from the last 24 hours. Collect the top 20 by engagement.",
    expected_output="Compact JSON with tweet_id, author_username, text, created_at, has_more, next_cursor, and route_used",
    agent=researcher,
)

analysis_task = Task(
    description="Analyze the collected tweets. Identify sentiment distribution, peak posting times, and top contributors.",
    expected_output="Structured analysis with charts-ready data",
    agent=analyst,
)

report_task = Task(
    description="Write a 3-paragraph executive summary of the findings.",
    expected_output="Executive summary in markdown format",
    agent=writer,
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "AI agents"})
Path("xquik-crewai-workflow-handoff.json").write_text(
    str(result),
    encoding="utf-8",
)

from crewai.mcp import MCPServerHTTP
from crewai.mcp.filters import create_static_tool_filter

# Discovery only: cannot execute read or write API operations
researcher = Agent(
    role="API Catalog Researcher",
    goal="Inspect Xquik endpoint schemas without executing API calls",
    backstory="Researcher who maps available API operations",
    mcps=[
        MCPServerHTTP(
            url="https://xquik.com/mcp",
            headers={"x-api-key": "xq_YOUR_KEY_HERE"},
            tool_filter=create_static_tool_filter(
                allowed_tool_names=["explore"],
            ),
        ),
    ],
)

import os

xquik_mcp = MCPServerHTTP(
    url="https://xquik.com/mcp",
    headers={"x-api-key": os.environ["XQUIK_API_KEY"]},
)