CrewAI

CrewAI is a Python framework for role-based multi-agent systems. You define Agent objects with a role, goal, and backstory, hand them Task objects, assemble them into a Crew, and let CrewAI coordinate the conversation. The framework owns the conversation; Flux owns whether it survives a crash, gets retried on failure, runs on a schedule, and pulls API keys out of the secrets store.

The integration boundary is exactly one Flux @task that wraps crew.kickoff(). Everything CrewAI does happens inside that task. Everything Flux does happens around it.

The Phase 4 page Connecting CrewAI walks through the basic pattern. This page covers the deeper concerns: how to size retries and timeouts, where secrets fit, how to scale crews across workers, and what each side is actually responsible for.

Prerequisites

pip install crewai litellm flux-core

CrewAI routes every LLM call through LiteLLM, so install both. The LLM constructor accepts model strings in LiteLLM’s provider/model format: openai/gpt-4o, anthropic/claude-3-5-sonnet-20241022, ollama/llama3.

Wrapping a full crew

A real crew has multiple agents and tasks with explicit context passing between them. The whole thing lives inside a single Flux task:

from crewai import Agent, Crew, LLM, Process, Task
from flux import ExecutionContext, task, workflow


@task.with_options(
    retry_max_attempts=3,
    retry_delay=2,
    timeout=600,
    secret_requests=["openai_api_key"],
)
async def run_research_crew(topic: str, secrets: dict) -> str:
    """Execute a two-agent research crew and return the final report."""
    import os
    os.environ["OPENAI_API_KEY"] = secrets["openai_api_key"]

    llm = LLM(model="openai/gpt-4o-mini")

    researcher = Agent(
        role="Research Analyst",
        goal="Gather facts and supporting evidence for {topic}",
        backstory="A meticulous analyst who cites sources and avoids speculation.",
        llm=llm,
        verbose=False,
    )

    editor = Agent(
        role="Technical Editor",
        goal="Turn raw research into a polished 500-word briefing",
        backstory="An editor who values clarity, structure, and concrete examples.",
        llm=llm,
        verbose=False,
    )

    research_task = Task(
        description="Research {topic}. List key facts, trends, and three primary sources.",
        expected_output="A bulleted list of findings with source attributions.",
        agent=researcher,
    )

    writing_task = Task(
        description="Write a 500-word briefing on {topic} using the research findings.",
        expected_output="A complete briefing with headline, intro, body, and takeaways.",
        agent=editor,
        context=[research_task],
    )

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

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


@workflow
async def research_brief(ctx: ExecutionContext[dict]):
    topic = (ctx.input or {}).get("topic")
    if not topic:
        return {"error": "topic required"}
    brief = await run_research_crew(topic)
    return {"topic": topic, "brief": brief, "execution_id": ctx.execution_id}

Run it:

flux secrets set openai_api_key sk-...
flux workflow register research_brief.py
flux workflow run research_brief '{"topic": "Vector databases in 2026"}'

Sizing retries and timeouts

A sequential crew with two agents and an openai/gpt-4o-mini model takes roughly 30 to 90 seconds end-to-end. A hierarchical crew with a manager agent and three specialists can take three to ten minutes. Two rules:

• Set timeout to cover the whole crew, not one agent call. CrewAI does not surface per-agent timeouts to Flux. If you set timeout=60 on a crew that runs two LLM calls back-to-back, Flux cancels the task mid-conversation and you lose any partial output.
• Retries re-run the entire crew from scratch. CrewAI does not expose a checkpoint between agents, so a Flux retry means the researcher and editor both run again. Plan for that cost. retry_max_attempts=3 against a long crew can mean three full crew runs.

Retry delays use exponential backoff. The wait before attempt N is retry_delay × retry_backoff^(N-1), capped at 600 seconds. With retry_delay=2 and the default retry_backoff=2, retries wait 2s, then 4s, then 8s, and so on. Set retry_backoff=1 for a flat delay. The task timeout also applies to each retried attempt.

LLM rate limits

Hosted providers throttle. When OpenAI or Anthropic returns 429, CrewAI raises an exception, which Flux catches and retries. A modest policy works for most workloads:

@task.with_options(retry_max_attempts=3, retry_delay=5, timeout=600)
async def run_crew(...): ...

A 5-second delay between attempts buys time for the provider’s rate-limit window to roll over. For larger crews (five-plus agents) or shared API keys, bump retry_delay to 15 or 30.

Secrets

API keys belong in flux secrets, not in workflow source or environment variables baked into worker images. Declare what the task needs with secret_requests, and Flux injects them as a secrets kwarg:

@task.with_options(
    secret_requests=["openai_api_key", "anthropic_api_key"],
)
async def run_multi_model_crew(topic: str, secrets: dict) -> str:
    import os
    os.environ["OPENAI_API_KEY"] = secrets["openai_api_key"]
    os.environ["ANTHROPIC_API_KEY"] = secrets["anthropic_api_key"]
    # ... build crew with LLMs from both providers

Set the secret values once with flux secrets set <name> <value>. The encrypted blob lives in the Flux database; tasks pull it on demand and the value never appears in the event log.

What each side handles

The split has one important consequence: CrewAI does not know it is running inside Flux. If the crew prints to stdout, it lands in the worker’s log; if it raises, Flux sees the exception and decides retry vs failure. CrewAI’s own retry handling (max_iter on an agent) operates inside one Flux task call.

Scaling crews across workers

Each crew.kickoff() is one Flux task call on one worker. To run many crews in parallel — say, one per customer in a batch — fan out at the workflow level using parallel:

from flux.tasks import parallel


@workflow
async def daily_briefings(ctx: ExecutionContext[dict]):
    topics = (ctx.input or {}).get("topics", [])
    briefs = await parallel(*[run_research_crew(t) for t in topics])
    return {"briefs": briefs}

A worker runs multiple workflow executions concurrently — the runtime uses asyncio.create_task to dispatch them — but each LLM call inside a crew is still synchronous from CrewAI’s perspective. So one worker with five concurrent workflows really does run five crews in parallel, each holding its own LiteLLM HTTP connections.

Handling crew output

crew.kickoff() returns a CrewOutput. The shape matters when you want per-agent results:

• crew_output.raw — final agent’s output as a string. Use it for single-result crews.
• crew_output.tasks_output — list of per-task outputs aligned with the tasks=[...] list you passed to Crew. Use it when each agent’s output matters independently.

crew_output = crew.kickoff(inputs={"topic": topic})

for task_output in crew_output.tasks_output:
    print(str(task_output))  # one entry per Task

When agents are instructed to return JSON, the output is a string. Call json.loads(str(task_output)) and catch json.JSONDecodeError — LLMs occasionally wrap JSON in markdown fences or truncate long arrays.

When CrewAI is the right tool

Pick CrewAI when the work decomposes naturally into role-based collaboration: a research agent, a writer agent, an editor agent. The role/goal/backstory abstraction lets you encode “this agent thinks differently than that one” in plain language.

Pick Flux’s built-in agent() task when you want a single agent with tools, structured output, and tight integration with the workflow’s event log. Pick LangGraph (covered on the LangGraph page) when you want explicit graph state with conditional edges.

You can also mix. A workflow can call a CrewAI crew for the multi-agent research phase, then hand the result to a Flux agent() for structured-output extraction:

@workflow
async def research_and_extract(ctx: ExecutionContext[dict]):
    topic = ctx.input["topic"]
    brief = await run_research_crew(topic)
    summary = await summarize_agent(brief)  # a flux.agent() task
    return summary

CrewAI does the multi-agent dance; Flux’s agent() (which returns a callable @task) does the structured output. Both run inside the same Flux workflow.

Derived against CrewAI 0.x and Flux 0.56.0, 2026-07.