CrewAI task design tutorial: chaining task outputs

Chaining task outputs is what turns a pile of independent agents into a pipeline that actually gets work done. This crewai chaining task outputs tutorial walks through building a three-stage research-to-email crew where each task consumes the previous task’s result via CrewAI’s context parameter. You’ll see runnable code, expected intermediate shapes, and the failure modes that bite engineers in production.

Prerequisites

• Python 3.10 or newer.
• crewai installed (>=0.30.0): pip install crewai.
• An OpenAI-compatible API key. If you want a single endpoint that fronts 240+ models with automatic fallback when a provider is degraded, point CrewAI’s ChatOpenAI at n4n.ai’s gateway.
• Basic familiarity with Agent and Task construction.

pip install crewai pydantic
export OPENAI_API_KEY="sk-..."  # or use the gateway key

The core mechanism: Task context

In CrewAI, a Task can declare a context list containing other Task objects. At execution time, the crew resolves each referenced task, serializes its output, and injects it into the dependent task’s description. This is the primitive for chaining.

from crewai import Agent, Task

researcher = Agent(role="Researcher", goal="Find facts", backstory="Expert searcher")
writer = Agent(role="Writer", goal="Summarize", backstory="Concise author")

task_a = Task(
    description="List three recent advances in vector databases.",
    expected_output="Bullet list of three advances with one sentence each.",
    agent=researcher,
)

task_b = Task(
    description="Rewrite the research into a 2-paragraph summary for executives.",
    expected_output="Two paragraphs, no bullet points.",
    agent=writer,
    context=[task_a],  # <-- chaining happens here
)

task_b receives task_a.output as contextual grounding. No shared global state, no manual string formatting.

Step 1: Define the agents

We’ll model a realistic pipeline: a researcher gathers raw findings, an analyst condenses them into a structured brief, and a comms agent drafts a customer email. Each agent gets a narrow role and explicit goal to reduce drift.

from crewai import Agent

researcher = Agent(
    role="Market Researcher",
    goal="Collect factual, sourced findings on a given topic",
    backstory="Former analyst who cites everything",
    verbose=True,
)

analyst = Agent(
    role="Briefing Analyst",
    goal="Convert raw research into a tight structured brief",
    backstory="Writes for time-poor executives",
    verbose=True,
)

comms = Agent(
    role="Customer Comms",
    goal="Draft a clear, non-salesy email from a brief",
    backstory="Senior PR writer",
    verbose=True,
)

Step 2: Create tasks with chaining

Define three tasks. The second takes the first as context; the third takes the second. We use expected_output to force shape, which matters when the next task parses the text.

from crewai import Task

research_task = Task(
    description="Research the topic: 'edge deployment of LLMs'. "
                "Return 4 concrete facts with a source URL each.",
    expected_output="4 numbered items, each: fact + URL.",
    agent=researcher,
)

brief_task = Task(
    description="From the research, produce a brief with sections: "
                "Summary, Risks, Opportunities. Keep under 150 words.",
    expected_output="Markdown with three headings as specified.",
    agent=analyst,
    context=[research_task],
)

email_task = Task(
    description="Using the brief, draft an email to a technical customer. "
                "Subject line, then body. No fluff.",
    expected_output="Subject: ...\n\nBody text.",
    agent=comms,
    context=[brief_task],
)

Note the chain is linear, but context accepts multiple tasks if you need fan-in (e.g., context=[research_task, brief_task]).

Step 3: Assemble and run the crew

CrewAI defaults to sequential Process.sequential, which respects context dependencies automatically. You don’t need to topologically sort.

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, analyst, comms],
    tasks=[research_task, brief_task, email_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)

Expected output at checkpoints

After research_task completes, its output looks like:

1. llama.cpp now runs 70B models on Apple M2 Ultra at 8 tok/s (source: https://example.com/a)
2. Tinygrad shipped a CUDA backend for edge GPUs (source: https://example.com/b)
3. On-device quantization reduces memory 4x with 2% accuracy drop (source: https://example.com/c)
4. Qualcomm NPU inference API stabilized in Android 15 (source: https://example.com/d)

brief_task receives that text block prepended to its own description. Its output:

## Summary
Edge LLM deployment is now viable on consumer hardware via llama.cpp and NPU APIs.

## Risks
Quantization tradeoffs and fragmented tooling across vendors.

## Opportunities
Offline assistants, lower latency, data sovereignty.

email_task then produces:

Subject: Edge LLMs are ready for piloting

Hi Team,

Our research shows edge deployment has crossed a practical threshold...

If you see the email ignoring the brief, the chain broke—usually because context was omitted or the agent overrode instructions.

Advanced: structured outputs and guarding chains

Free-text chaining works, but for longer pipelines you want schemas. Use output_pydantic to force a task to return validated objects. The next task still gets a stringified version, but you can also access .pydantic on the task output.

from pydantic import BaseModel

class Brief(BaseModel):
    summary: str
    risks: list[str]
    opportunities: list[str]

brief_task = Task(
    description="From the research, extract structured brief.",
    expected_output="Valid Brief object.",
    agent=analyst,
    context=[research_task],
    output_pydantic=Brief,
)

After run, brief_task.output.pydantic is a Brief instance. The downstream email_task still uses context=[brief_task] and receives the model’s __str__ representation, which is stable and parseable.

Configuring the LLM through a gateway

CrewAI accepts any LangChain chat model. To avoid vendor lock and get fallback, instantiate ChatOpenAI with a custom base URL:

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    base_url="https://api.n4n.ai/v1",
    api_key="your-gateway-key",
)

researcher = Agent(..., llm=llm)

The gateway forwards provider cache-control hints and meters per-token usage, so the same crew code scales across model swaps without touching task logic.

Common pitfalls

Missing context. Forgetting context=[prev_task] silently yields an agent guessing from the description alone. Always assert task_b.context == [task_a] in tests.

Context overflow. Concatenating three task outputs can exceed context windows. Trim expected_output sizes or use output_pydantic to compress.

Non-deterministic shapes. If expected_output is vague, the next task gets messy input. Write strict output contracts.

Running async incorrectly. async_execution=True on a task in a sequential crew still blocks the chain; it only helps with fan-out inside a manager process. Don’t assume parallelism from context.

Agent role leakage. A writer agent tasked with “summarize” may editorialize if its backstory says “creative”. Keep goals narrow; chaining amplifies drift.

Wrapping up

The crewai chaining task outputs tutorial above is the minimal viable pattern: define tasks, wire them with context, and let the crew serialize dependencies. For production, add pydantic schemas at each hop and cap output lengths. The moment you treat task outputs as typed data rather than prompts, multi-agent pipelines become debuggable.