Connecting CrewAI

CrewAI organizes LLM agents into role-based teams. Each agent has a role, goal, and backstory, and CrewAI coordinates them through sequential or hierarchical processes. Flux wraps that orchestration as a durable task: the crew runs under Flux’s retry logic, timeout enforcement, and execution tracing. If the LLM call fails mid-way, Flux retries the crew. If the worker crashes, Flux resumes from the last checkpoint.

The integration is one @task that calls crew.kickoff() and returns its output. The rest of the workflow treats that output like any other task result.

Prerequisites

Install CrewAI and LiteLLM alongside Flux. LiteLLM is required for local Ollama models because CrewAI routes its LLM calls through it:

pip install crewai litellm

For hosted providers (OpenAI, Anthropic), LiteLLM is still pulled in transitively but no additional setup is needed beyond setting the relevant API key environment variable (OPENAI_API_KEY, ANTHROPIC_API_KEY, etc.). The LLM constructor accepts a model string in LiteLLM format: openai/gpt-4o, anthropic/claude-3-5-sonnet-20241022, ollama/llama3, and so on.

Wrapping a crew as a task

Define agents and tasks inside a @task function, assemble the Crew, call kickoff(), and return the result.

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


@task.with_options(retry_max_attempts=3, retry_delay=2, retry_backoff=2, timeout=300)
async def run_blog_crew(topic: str, model: str, ollama_url: str) -> str:
    """Execute the CrewAI blog post pipeline and return the raw output."""
    llm = LLM(model=f"ollama/{model}", base_url=ollama_url)

    researcher = Agent(
        role="Research Analyst",
        goal="Research the topic and identify key points, trends, and insights",
        backstory=(
            "An experienced research analyst who organizes findings into clear "
            "summaries that others can build upon."
        ),
        llm=llm,
        verbose=False,
    )

    writer = Agent(
        role="Content Writer",
        goal="Write an engaging blog post from the research findings",
        backstory=(
            "A content writer who makes technical topics accessible and structures "
            "posts with strong introductions and memorable conclusions."
        ),
        llm=llm,
        verbose=False,
    )

    research_task = Task(
        description="Research the topic: {topic}\n\nProvide key findings, trends, and examples.",
        expected_output="A structured research summary with key findings and supporting evidence.",
        agent=researcher,
    )

    writing_task = Task(
        description="Write a blog post about: {topic}\n\nUse the research findings to write an engaging post.",
        expected_output="A complete blog post with title, sections, and conclusion.",
        agent=writer,
        context=[research_task],
    )

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

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

@task.with_options(retry_max_attempts=3, ...) means the full crew re-runs if the LLM call raises an exception — for example, a connection timeout to a local Ollama server. The timeout=300 gives the sequential pipeline five minutes before Flux cancels it.

Using the task in a workflow

A second task can process the raw crew output before the workflow returns it:

from typing import Any


@task
async def format_blog_output(topic: str, raw_output: str) -> dict[str, Any]:
    """Parse the crew output into a structured result."""
    lines = raw_output.strip().splitlines()
    title = lines[0].strip().lstrip("#").strip() if lines else topic
    word_count = len(raw_output.split())
    return {
        "title": title,
        "content": raw_output.strip(),
        "word_count": word_count,
    }


@workflow
async def blog_post_writer_crewai(ctx: ExecutionContext[dict[str, Any]]):
    """
    Blog post writer backed by a CrewAI sequential pipeline.

    Input:
        topic      (required) — blog topic
        model      (optional) — Ollama model name, default "llama3"
        ollama_url (optional) — Ollama server URL, default "http://localhost:11434"
    """
    input_data = ctx.input or {}
    topic = input_data.get("topic")
    if not topic:
        return {"error": "Missing required parameter 'topic'", "execution_id": ctx.execution_id}

    model = input_data.get("model", "llama3")
    ollama_url = input_data.get("ollama_url", "http://localhost:11434")

    raw_output = await run_blog_crew(topic, model, ollama_url)
    blog_post = await format_blog_output(topic, raw_output)

    return {
        "topic": topic,
        "blog_post": blog_post["content"],
        "title": blog_post["title"],
        "word_count": blog_post["word_count"],
        "execution_id": ctx.execution_id,
    }

Run it via the CLI:

flux workflow run blog_post_writer_crewai '{"topic": "The Future of AI Agents"}'

Multi-agent code review example

The same pattern scales to larger crews. This example has four specialist agents — security, performance, style, and testing — each reviewing a code snippet, with a separate Flux task aggregating their output into a structured report.

Notice the two-task split: run_crewai_review holds the crew and calls kickoff(), while build_report is a plain Flux task that processes the results. Keeping the aggregation outside the crew task means it runs at normal Flux task speed, and a failure in parsing doesn’t trigger a full crew retry.

import json
from datetime import datetime
from typing import Any

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


@task.with_options(retry_max_attempts=2, timeout=300)
async def run_crewai_review(
    code: str,
    model: str,
    ollama_url: str,
    file_path: str | None = None,
) -> list[dict[str, Any]]:
    """Run a four-agent code review crew and return per-agent findings."""
    llm = LLM(model=f"ollama/{model}", base_url=ollama_url)

    security_agent = Agent(
        role="Security Code Reviewer",
        goal="Find security vulnerabilities: SQL injection, XSS, hardcoded secrets, path traversal.",
        backstory="Expert security auditor with penetration testing experience.",
        llm=llm,
        verbose=False,
    )

    security_task = Task(
        description=(
            f"Review this code for security vulnerabilities:\n```\n{code}\n```\n\n"
            "Respond as a JSON array where each item has: severity, issue, line, recommendation."
        ),
        expected_output="A JSON array of security findings.",
        agent=security_agent,
    )

    crew = Crew(
        agents=[security_agent],
        tasks=[security_task],
        process=Process.sequential,
        verbose=False,
    )

    crew_output = crew.kickoff()

    # crew_output.tasks_output holds per-task results when available
    tasks_output = getattr(crew_output, "tasks_output", None)
    results = []
    if tasks_output:
        for agent_name, task_output in zip(["security"], tasks_output):
            try:
                findings = json.loads(str(task_output))
                if not isinstance(findings, list):
                    findings = [findings]
                results.append({"agent": agent_name, "status": "success", "findings": findings})
            except json.JSONDecodeError:
                results.append({"agent": agent_name, "status": "parse_error", "findings": []})
    return results


@task
async def build_report(
    reviews: list[dict[str, Any]],
    execution_id: str,
    execution_time: float,
) -> dict[str, Any]:
    """Aggregate reviews from all agents into a summary report."""
    counts: dict[str, int] = {"critical": 0, "high": 0, "medium": 0, "low": 0}
    recommendations: list[str] = []

    for review in reviews:
        for finding in review.get("findings", []):
            severity = finding.get("severity", "low")
            counts[severity] = counts.get(severity, 0) + 1
            if severity in ("critical", "high"):
                issue = finding.get("issue", "Unknown issue")
                line = finding.get("line")
                line_info = f" (line {line})" if line else ""
                recommendations.append(f"{severity.upper()}: {issue}{line_info}")

    return {
        "summary": {
            "total_issues": sum(counts.values()),
            **counts,
            "execution_time": execution_time,
        },
        "recommendations": recommendations[:10],
        "metadata": {"execution_id": execution_id},
    }


@workflow
async def multi_agent_code_review_crewai(ctx: ExecutionContext[dict[str, Any]]):
    """
    Multi-agent code review using CrewAI specialist agents, wrapped in a Flux workflow.

    Input:
        code       (required) — source code string
        model      (optional) — Ollama model, default "llama3.2"
        ollama_url (optional) — Ollama server URL, default "http://localhost:11434"
        file_path  (optional) — file path for context
    """
    input_data = ctx.input or {}
    code = input_data.get("code")
    if not code:
        return {"error": "No code provided", "execution_id": ctx.execution_id}

    model = input_data.get("model", "llama3.2")
    ollama_url = input_data.get("ollama_url", "http://localhost:11434")
    file_path = input_data.get("file_path")

    start_time = datetime.now()

    reviews = await run_crewai_review(code, model, ollama_url, file_path)

    execution_time = (datetime.now() - start_time).total_seconds()
    return await build_report(reviews, ctx.execution_id, execution_time)

Run it:

flux workflow run multi_agent_code_review_crewai '{
    "code": "def login(user, pw):\n    q = f\"SELECT * FROM users WHERE name={user}\"\n    return db.execute(q)",
    "file_path": "auth.py"
}'

How the layers divide

CrewAI and Flux handle different things. CrewAI owns agent definitions, task descriptions, and the coordination process. Flux owns durability, retries, scheduling, and secrets. Neither library knows about the other’s internals. The boundary between them is the @task function that wraps crew.kickoff().

Handling crew output

crew.kickoff() returns a CrewOutput object. raw holds the final agent’s output as a plain string. For sequential crews, tasks_output is a list of per-task results, aligned in order with the task definitions you passed to Crew. Use it to extract each agent’s output separately rather than parsing the single concatenated string:

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

# Final agent's output as plain text
final_text = crew_output.raw

# Per-task outputs (sequential crews)
for task_result in crew_output.tasks_output:
    print(str(task_result))

When agents are instructed to return JSON, the output is still a string. Call json.loads(str(task_output)) and handle json.JSONDecodeError. LLMs occasionally wrap JSON in markdown fences (```json ... ```) or truncate it on very long responses — strip the fences before parsing if you see that pattern.

Error handling

Wrap crew.kickoff() in a try/except and re-raise as RuntimeError with a clear message. Flux’s retry logic triggers on any exception from the task body, so a descriptive message tells you at a glance whether it was a connectivity problem or a model issue:

@task.with_options(retry_max_attempts=3, retry_delay=2, retry_backoff=2, timeout=300)
async def run_crew(topic: str, model: str, ollama_url: str) -> str:
    try:
        # ... build and kick off crew
        result = crew.kickoff(inputs={"topic": topic})
        return result.raw
    except Exception as e:
        raise RuntimeError(
            f"CrewAI pipeline failed: {e}. "
            "Verify that Ollama is running (ollama serve) and the model is pulled."
        ) from e

On the final retry, Flux marks the task and its parent workflow execution as failed. Check the execution output with:

flux execution show <execution-id>

Running the full examples

The two runnable examples live in examples/ai/crewai/ in the Flux source tree: blog_post_writer.py and multi_agent_code_review.py. Start Ollama first, then run them via the CLI:

# Pull a model
ollama pull llama3

# Start Flux
flux start server &
flux start worker worker-1 &

# Blog post writer
flux workflow run blog_post_writer_crewai '{"topic": "The Future of AI Agents"}'

# Multi-agent code review
flux workflow run multi_agent_code_review_crewai '{
    "code": "def foo(): pass",
    "model": "llama3.2"
}'