Multi-agent orchestration: LangGraph, CrewAI and Mastra vs. Claude Code

# A minimal LangGraph agent example (Python)
from typing import TypedDict
from langgraph.graph import StateGraph, END

# Define the state
class ResearchState(TypedDict):
    query: str
    sources: list[str]
    draft: str
    final: str
    iteration: int

# Nodes are functions or agents
def researcher(state: ResearchState) -> ResearchState:
    # here the agent searches for information
    return {"sources": ["source1", "source2"], "iteration": state["iteration"] + 1}

def writer(state: ResearchState) -> ResearchState:
    # here the agent writes a draft
    return {"draft": "Draft based on sources..."}

def should_revise(state: ResearchState) -> str:
    # Conditional edge: go to revision or finish
    if state["iteration"] < 3 and len(state["draft"]) < 500:
        return "revise"
    return "finalize"

# Build the graph
graph = StateGraph(ResearchState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)

graph.set_entry_point("researcher")
graph.add_edge("researcher", "writer")
graph.add_conditional_edges("writer", should_revise, {
    "revise": "researcher",  # loop back!
    "finalize": END
})

app = graph.compile()
result = app.invoke({"query": "AI trends", "iteration": 0})

# Example: a CrewAI research team
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# Agent 1: market researcher
market_researcher = Agent(
    role="Market Research Analyst",
    goal="Find and analyze the latest data on the market for AI developer tools",
    backstory="""You are an experienced market analyst with 10 years in the field.
    You specialize in technology markets.
    You always back up your conclusions with concrete data and numbers.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, web_tool],
    llm="anthropic/claude-sonnet-5-5"  # CrewAI supports Claude; the model is set as "provider/model"; current models: the What's current page
)

# Agent 2: technical expert
tech_expert = Agent(
    role="Technical Expert",
    goal="Evaluate the technical capabilities and limitations of AI tools",
    backstory="""You are a senior engineer with deep AI/ML expertise.
    You can explain complex technical concepts in plain language.
    You're skeptical of marketing claims and look at real benchmarks.""",
    verbose=True,
    allow_delegation=False,
    tools=[web_tool],
    llm="anthropic/claude-sonnet-5-5"
)

# Agent 3: financial analyst
financial_analyst = Agent(
    role="Financial Analyst",
    goal="Evaluate the ROI and business potential of AI tools",
    backstory="""You are a financial analyst specializing in tech startups.
    You can build unit economics and assess market potential.
    You always look at the pricing model and margins.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool],
    llm="anthropic/claude-haiku-4-5"  # A simpler task, so we use a cheaper model
)

# Agent 4: writer-synthesizer
report_writer = Agent(
    role="Strategic Report Writer",
    goal="Synthesize the research into a clear, actionable report",
    backstory="""You are an experienced technical writer and strategy consultant.
    You can turn complex analytical data into clear recommendations.
    You write for tech founders and CTOs.""",
    verbose=True,
    allow_delegation=False,
    tools=[],
    llm="anthropic/claude-sonnet-5-5"
)

# Tasks
market_task = Task(
    description="""Research the market for AI developer tools in 2026.
    Find: the top 5 players, market size, growth rate, key trends.
    Use current data, no older than 6 months.""",
    expected_output="A structured market analysis: size, players, trends, numbers",
    agent=market_researcher,
    async_execution=True  # In parallel with the other tasks!
)

tech_task = Task(
    description="""Do a technical analysis of the top 3 AI coding assistants (Claude Code, Cursor, GitHub Copilot).
    Compare: supported languages, context window, integrations, latency, accuracy.""",
    expected_output="A comparison table with technical specs and benchmarks",
    agent=tech_expert,
    async_execution=True  # In parallel!
)

financial_task = Task(
    description="""Analyze pricing and ROI for AI coding assistants.
    Calculate: cost per developer/month, time saved, payback period.
    Compare pricing models: subscription vs. usage-based.""",
    expected_output="A financial model with an ROI calculation and a pricing comparison",
    agent=financial_analyst,
    async_execution=True  # In parallel!
)

synthesis_task = Task(
    description="""Synthesize all three studies into a final strategy report.
    Structure: Executive Summary (1 page), Market Overview, Tech Comparison, Financial Analysis, Recommendations.
    Audience: a tech founder deciding whether to adopt AI tools.""",
    expected_output="A complete 5-7 page strategy report, ready to present",
    agent=report_writer,
    context=[market_task, tech_task, financial_task]  # Depends on the previous ones
)

# Assemble the team
crew = Crew(
    agents=[market_researcher, tech_expert, financial_analyst, report_writer],
    tasks=[market_task, tech_task, financial_task, synthesis_task],
    process=Process.sequential,  # or Process.hierarchical with a manager agent
    verbose=True,
    memory=True,  # Agents remember previous interactions
    max_rpm=10    # Rate limiting for API calls
)

# Run it
result = crew.kickoff()
print(result.raw)

pip install crewai crewai-tools

// Example: a Mastra agent with a workflow (TypeScript)
import { Mastra } from "@mastra/core";
import { Agent } from "@mastra/core/agent";
import { createTool } from "@mastra/core/tools";
import { createStep, createWorkflow } from "@mastra/core/workflows";
import { z } from "zod";
import Anthropic from "@anthropic-ai/sdk";

const anthropic = new Anthropic();

// Define the tools
const searchTool = createTool({
  id: "web-search",
  description: "Searches the internet for current information",
  inputSchema: z.object({
    query: z.string().describe("Search query"),
  }),
  outputSchema: z.object({ results: z.array(z.string()) }),
  execute: async ({ query }) => {
    // A real search via Serper, Tavily, etc. goes here
    return { results: [`Results for: ${query}`] };
  },
});

// Create a research agent. The model is set as a "provider/model" string
const researchAgent = new Agent({
  id: "research-agent",
  name: "Research Agent",
  instructions: `You are a professional researcher.
  You find current information on the request, structure it,
  and highlight the key facts with sources.`,
  model: "anthropic/claude-sonnet-5-5",   // current models: the What's current page
  tools: { searchTool },
});

// Workflow steps: each step describes its input and output data
const researchStep = createStep({
  id: "research",
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: z.object({ research: z.string() }),
  execute: async ({ inputData, mastra }) => {
    const agent = mastra.getAgent("researchAgent");
    const result = await agent.generate(
      `Research the topic: ${inputData.topic}. Find current data and trends.`
    );
    return { research: result.text };
  },
});

const summarizeStep = createStep({
  id: "summarize",
  inputSchema: z.object({ research: z.string() }),
  outputSchema: z.object({ summary: z.string() }),
  execute: async ({ inputData }) => {
    const response = await anthropic.messages.create({
      model: "claude-haiku-4-5",
      max_tokens: 500,
      messages: [
        {
          role: "user",
          content: `Condense this research into 3 key takeaways:\n\n${inputData.research}`,
        },
      ],
    });
    const summary = response.content.map((b) => (b.type === "text" ? b.text : "")).join("");
    return { summary };
  },
});

const researchWorkflow = createWorkflow({
  id: "research-workflow",
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: z.object({ summary: z.string() }),
})
  .then(researchStep)
  .then(summarizeStep)
  .commit();

// Initialize Mastra
const mastra = new Mastra({
  agents: { researchAgent },
  workflows: { researchWorkflow },
});

// Run it (check the documentation for your Mastra version for how to run it)
const run = await mastra.getWorkflow("researchWorkflow").createRun();
const result = await run.start({ inputData: { topic: "AI agents" } });
console.log(result);

# crew_service.py — wrapping CrewAI as a FastAPI service
from fastapi import FastAPI
from pydantic import BaseModel
from crewai import Agent, Task, Crew, Process

app = FastAPI()

class ResearchRequest(BaseModel):
    topic: str
    output_format: str = "report"

@app.post("/research")
async def run_research_crew(request: ResearchRequest):
    """
    This endpoint is called by the Claude Code orchestrator.
    Inside, a parallel CrewAI team runs.
    """
    researcher = Agent(
        role="Senior Researcher",
        goal=f"Find everything important about: {request.topic}",
        backstory="An experienced analyst with internet access",
        llm="anthropic/claude-sonnet-5-5",
        verbose=False  # Turn verbose off in production
    )

    writer = Agent(
        role="Technical Writer",
        goal="Turn the research into a clear, structured document",
        backstory="A technical documentation specialist",
        llm="anthropic/claude-haiku-4-5",
        verbose=False
    )

    research_task = Task(
        description=f"Research: {request.topic}. Find facts, numbers, trends.",
        expected_output="Structured data with sources",
        agent=researcher
    )

    write_task = Task(
        description=f"Write a {request.output_format} based on the research",
        expected_output=f"A finished {request.output_format} for the client",
        agent=writer,
        context=[research_task]
    )

    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, write_task],
        process=Process.sequential,
        verbose=False
    )

    result = crew.kickoff()
    return {"result": result.raw, "topic": request.topic}

# In .claude/agents/research-orchestrator.md
# Claude Code calls CrewAI as an external API
curl -X POST http://localhost:8000/research \
  -H "Content-Type: application/json" \
  -d '{"topic": "AI orchestration frameworks", "output_format": "report"}'

Is the task simple and linear, with no branches?
  → Native Claude Code. Don't overcomplicate it.

Need "try → fix → repeat" loops?
  → LangGraph

Need several parallel experts on one topic?
  → CrewAI

Whole stack in TypeScript, deploying to Vercel?
  → Mastra

Need enterprise reliability, lots of compliance?
  → Microsoft Agent Framework (AutoGen's successor)

Want a mixed approach?
  → Claude Code orchestrator + CrewAI/LangGraph as an external service

# Create the project
mkdir multiagent-crew && cd multiagent-crew
python -m venv venv && source venv/bin/activate  # Windows: venv\Scripts\activate

# Install the dependencies
pip install crewai crewai-tools anthropic python-dotenv fastapi uvicorn

# The .env file
echo "ANTHROPIC_API_KEY=sk-ant-your-key-here" > .env
echo "SERPER_API_KEY=your-serper-key" >> .env  # for internet search

# simple_crew.py — start with the minimum
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew

load_dotenv()

# Set up the model for CrewAI
os.environ["ANTHROPIC_API_KEY"] = os.getenv("ANTHROPIC_API_KEY")

# Two agents
researcher = Agent(
    role="Researcher",
    goal="Find the key information on the topic",
    backstory="An analyst with experience in the tech industry",
    llm="anthropic/claude-haiku-4-5",  # A cheap model for research
    verbose=True
)

writer = Agent(
    role="Writer",
    goal="Write clear, useful content based on the research",
    backstory="A technical writer specializing in AI topics",
    llm="anthropic/claude-sonnet-5-5",  # A smarter model for the final text
    verbose=True
)

# Tasks
research_task = Task(
    description="Find 5 key trends in multi-agent AI systems in 2026",
    expected_output="A list of 5 trends with a short description of each (50-100 words per trend)",
    agent=researcher
)

write_task = Task(
    description="""Based on the research, write a short-form article (500 words)
    for software developers. Tone: professional, no filler, specific.""",
    expected_output="A finished 500-word article in markdown",
    agent=writer,
    context=[research_task]
)

# Crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    verbose=True
)

# Run
print("Starting the team...")
result = crew.kickoff()
print("\n" + "="*50)
print("RESULT:")
print("="*50)
print(result.raw)

# parallel_crew.py — a real parallel team
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

load_dotenv()

search = SerperDevTool()

# Four specialized experts
market_analyst = Agent(
    role="Market Analyst",
    goal="Analyze the market situation and the competitors",
    backstory="A market analyst focused on SaaS and AI tools",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

tech_reviewer = Agent(
    role="Technical Reviewer",
    goal="Evaluate the technical side and the integrations",
    backstory="A senior developer who knows all the popular AI frameworks",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

pricing_expert = Agent(
    role="Pricing Expert",
    goal="Figure out the pricing and ROI",
    backstory="A financial analyst specializing in SaaS metrics",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

strategy_writer = Agent(
    role="Strategy Consultant",
    goal="Synthesize all the data into actionable recommendations",
    backstory="A McKinsey-level consultant who writes clear executive briefs",
    llm="anthropic/claude-sonnet-5-5",  # A strong model for the final synthesis
    verbose=True
)

# Parallel tasks (async_execution=True)
market_task = Task(
    description="Research the market for multi-agent AI frameworks: size, players, 2026 trends",
    expected_output="Market analysis: 3-5 key players, trends, market size",
    agent=market_analyst,
    async_execution=True  # IN PARALLEL
)

tech_task = Task(
    description="Compare technically: LangGraph, CrewAI, Mastra — capabilities, limitations, ecosystem",
    expected_output="A technical comparison with specific use case examples",
    agent=tech_reviewer,
    async_execution=True  # IN PARALLEL
)

pricing_task = Task(
    description="Break down the pricing models and ROI: what it costs to use, what the savings are",
    expected_output="A financial analysis with specific numbers and an ROI calculation",
    agent=pricing_expert,
    async_execution=True  # IN PARALLEL
)

# Synthesis waits for all three parallel tasks
synthesis_task = Task(
    description="""Synthesize the three analyses into an executive brief for a CTO.
    Structure: Situation → Conclusions → Recommendations (top 3, actionable).
    1 page max. No filler.""",
    expected_output="An executive brief in markdown, 1 page, ready to present",
    agent=strategy_writer,
    context=[market_task, tech_task, pricing_task]  # Depends on all three
)

# Assemble the team
crew = Crew(
    agents=[market_analyst, tech_reviewer, pricing_expert, strategy_writer],
    tasks=[market_task, tech_task, pricing_task, synthesis_task],
    process=Process.sequential,
    verbose=True,
    memory=False  # Turned off in this example for speed
)

# Run
print("Starting the parallel team...")
print("market_analyst, tech_reviewer, pricing_expert work at the same time")
print("strategy_writer starts when all three are done")
print("-" * 50)

result = crew.kickoff()

# Save the result
with open("report.md", "w", encoding="utf-8") as f:
    f.write(result.raw)

print("\nReport saved to report.md")

# Run the simple team
python simple_crew.py

# Run the parallel team
python parallel_crew.py

# If you need to debug, set verbose=True on the agents and the Crew in the code

# Check the token usage
# At the end, CrewAI shows the total tokens usage

# Error: model not found
# Make sure ANTHROPIC_API_KEY is set in .env
# CrewAI looks for the key through os.environ

# Error: rate limit
# Add the max_rpm parameter to Crew:
crew = Crew(..., max_rpm=5)  # at most 5 requests per minute

# The result didn't make it into the context
# Make sure context=[task1, task2] is set on the dependent task

# api_service.py — CrewAI as a microservice
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
from crewai import Agent, Task, Crew

app = FastAPI(title="Research Crew API")

class ResearchRequest(BaseModel):
    topic: str
    style: str = "technical"  # technical, executive, casual

@app.post("/research")
async def research(req: ResearchRequest):
    """
    Claude Code calls this endpoint through bash/curl.
    Inside, a parallel CrewAI team runs.
    """
    researcher = Agent(
        role="Researcher",
        goal=f"Research the topic thoroughly: {req.topic}",
        backstory="An experienced analyst",
        llm="anthropic/claude-haiku-4-5",
        verbose=False
    )

    writer = Agent(
        role="Writer",
        goal=f"Write in this style: {req.style}",
        backstory="A technical writer",
        llm="anthropic/claude-sonnet-5-5",
        verbose=False
    )

    research_task = Task(
        description=f"Research: {req.topic}",
        expected_output="A detailed analysis with facts",
        agent=researcher
    )

    write_task = Task(
        description=f"Write in the '{req.style}' style based on the research",
        expected_output="A finished document",
        agent=writer,
        context=[research_task]
    )

    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, write_task]
    )

    result = crew.kickoff()
    return {
        "topic": req.topic,
        "result": result.raw,
        "tokens_used": result.token_usage.total_tokens if result.token_usage else 0
    }

@app.get("/health")
def health():
    return {"status": "ok"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)

# Start the service
python api_service.py

# Now Claude Code can call it:
curl -X POST http://localhost:8001/research \
  -H "Content-Type: application/json" \
  -d '{"topic": "Multiagent orchestration trends", "style": "executive"}'