CrewAI task design tutorial: context sharing between tasks

If you’ve built anything beyond a toy CrewAI pipeline, you’ve hit the context sharing problem: Task B needs Task A’s output, but the framework doesn’t make this automatic. This crewai task context sharing tutorial walks through the three patterns that actually work in production — direct output passing, shared memory, and callback-driven state — with runnable code at each step.

Prerequisites

You need Python 3.10+ and a working OpenAI-compatible endpoint. The examples use crewai>=0.80.0 and langchain-openai for the LLM wrapper. Install dependencies:

pip install crewai langchain-openai python-dotenv

Create a .env file with your API credentials:

OPENAI_API_KEY=your-key-here
OPENAI_API_BASE=https://api.openai.com/v1  # or your gateway endpoint

If you route through a gateway like n4n.ai, set OPENAI_API_BASE to its endpoint and use your gateway key — the OpenAI-compatible interface means zero code changes.

The Default Behavior: Isolated Tasks

By default, CrewAI tasks run independently. Each task receives only its own description and expected_output. The agent’s prompt includes the task description but not sibling task outputs unless you explicitly wire them.

# isolated_tasks.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_API_BASE"),
    temperature=0.1
)

researcher = Agent(
    role="Research Analyst",
    goal="Find key facts about a topic",
    backstory="You extract concise, verifiable information.",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Write a summary from research",
    backstory="You transform raw facts into readable prose.",
    llm=llm,
    verbose=True
)

research_task = Task(
    description="Research the key differences between REST and GraphQL APIs. List 5 concrete differences.",
    expected_output="A bulleted list of 5 differences with brief explanations",
    agent=researcher
)

write_task = Task(
    description="Write a 150-word comparison article based on the research",
    expected_output="A cohesive 150-word article comparing REST and GraphQL",
    agent=writer
)

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

result = crew.kickoff()
print(result)

Run it:

python isolated_tasks.py

Expected output: The writer produces a generic comparison because it never saw the researcher’s output. The tasks ran sequentially but shared nothing. This is the problem we’re solving.

Pattern 1: Direct Output Passing with context

The context parameter on Task accepts a list of upstream tasks whose outputs get injected into the downstream task’s prompt. This is the simplest pattern and covers 80% of use cases.

# context_passing.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_API_BASE"),
    temperature=0.1
)

researcher = Agent(
    role="Research Analyst",
    goal="Find key facts about a topic",
    backstory="You extract concise, verifiable information.",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Write a summary from research",
    backstory="You transform raw facts into readable prose.",
    llm=llm,
    verbose=True
)

research_task = Task(
    description="Research the key differences between REST and GraphQL APIs. List 5 concrete differences.",
    expected_output="A bulleted list of 5 differences with brief explanations",
    agent=researcher
)

write_task = Task(
    description=(
        "Write a 150-word comparison article based on the research provided. "
        "Use the specific differences found by the researcher. Do not invent new ones."
    ),
    expected_output="A cohesive 150-word article comparing REST and GraphQL using the provided research",
    agent=writer,
    context=[research_task]  # <-- This wires the output
)

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

result = crew.kickoff()
print(result)

Run it:

python context_passing.py

Expected output: The writer now references specific differences from the researcher’s output (e.g., “As the research noted, GraphQL allows clients to specify exact data requirements…”). The context parameter injects the upstream task’s raw output into the downstream agent’s prompt as a context block.

How context Works Internally

When you specify context=[task_a, task_b], CrewAI builds a prompt section like:

Context from previous tasks:
Task: Research the key differences...
Output: - REST uses multiple endpoints; GraphQL uses a single endpoint...
- REST over-fetches data; GraphQL prevents over-fetching...

The downstream agent sees this before its own task description. No parsing logic required — but you lose structure. The output is a string blob.

Pattern 2: Structured Output with Pydantic Models

String blobs break when downstream tasks need to extract specific fields. Use output_json or output_pydantic on the upstream task to enforce structure, then access typed fields in the downstream task.

# structured_context.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List
import os

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_API_BASE"),
    temperature=0.1
)

class Difference(BaseModel):
    aspect: str = Field(description="The category of difference (e.g., 'fetching', 'endpoints')")
    rest_behavior: str = Field(description="How REST handles this")
    graphql_behavior: str = Field(description="How GraphQL handles this")

class ResearchOutput(BaseModel):
    differences: List[Difference] = Field(description="List of 5 differences")
    summary: str = Field(description="One-sentence summary")

researcher = Agent(
    role="Research Analyst",
    goal="Find key facts about a topic",
    backstory="You extract concise, verifiable information in structured format.",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Write a summary from structured research",
    backstory="You transform structured data into readable prose.",
    llm=llm,
    verbose=True
)

research_task = Task(
    description="Research the key differences between REST and GraphQL APIs. Return structured data.",
    expected_output="A ResearchOutput object with 5 differences and a summary",
    agent=researcher,
    output_pydantic=ResearchOutput
)

write_task = Task(
    description=(
        "Write a 150-word comparison article using the structured research. "
        "Reference each difference by its 'aspect' field. "
        "The research is available as a ResearchOutput object with 'differences' list and 'summary'."
    ),
    expected_output="A cohesive 150-word article comparing REST and GraphQL",
    agent=writer,
    context=[research_task]
)

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

result = crew.kickoff()
print(result)

Run it:

python structured_context.py

Expected output: The writer references structured fields like “In the ‘fetching’ aspect, REST over-fetches while GraphQL prevents over-fetching.” The Pydantic model guarantees the shape, so downstream prompts can reference differences[0].aspect reliably.

Accessing Structured Context in Code

Sometimes you need the parsed object in your own Python logic, not just the agent prompt. After kickoff(), each task’s output attribute holds the raw string. For Pydantic tasks, use output.pydantic:

result = crew.kickoff()

# Access the structured output programmatically
research_output = research_task.output.pydantic
print(f"Found {len(research_output.differences)} differences")
for diff in research_output.differences:
    print(f"  - {diff.aspect}: REST={diff.rest_behavior} | GraphQL={diff.graphql_behavior}")

This enables hybrid workflows: agents consume context via prompts, your orchestration code consumes it via typed objects.

Pattern 3: Shared Memory for Long-Running State

context passes output forward once. For state that accumulates across many tasks — or that multiple downstream tasks need — use CrewAI’s memory system. Enable it on the crew and give agents memory=True.

# shared_memory.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_API_BASE"),
    temperature=0.1
)

# Shared memory requires a memory backend. CrewAI uses ChromaDB by default.
# pip install chromadb

fact_finder = Agent(
    role="Fact Finder",
    goal="Discover and store interesting facts about topics",
    backstory="You research topics and commit key facts to long-term memory.",
    llm=llm,
    verbose=True,
    memory=True  # <-- Enables memory for this agent
)

synthesizer = Agent(
    role="Synthesizer",
    goal="Connect facts across multiple research sessions",
    backstory="You retrieve stored facts and find patterns.",
    llm=llm,
    verbose=True,
    memory=True
)

research_task = Task(
    description="Research 3 surprising facts about the history of the Python programming language. Store each fact in memory with the tag 'python-history'.",
    expected_output="Confirmation that 3 facts were stored",
    agent=fact_finder
)

recall_task = Task(
    description="Retrieve all facts tagged 'python-history' from memory and write a connecting narrative showing how they relate.",
    expected_output="A narrative connecting the retrieved facts",
    agent=synthesizer
)

crew = Crew(
    agents=[fact_finder, synthesizer],
    tasks=[research_task, recall_task],
    process=Process.sequential,
    verbose=True,
    memory=True  # <-- Enables crew-level memory
)

result = crew.kickoff()
print(result)

Run it:

python shared_memory.py

Expected output: The synthesizer retrieves facts the finder stored, even though no explicit context link exists. Memory persists across crew runs if you reuse the same ChromaDB directory.

Memory Configuration Details

By default, CrewAI creates a ./memory directory with ChromaDB. For production, configure a persistent path:

from crewai.memory import LongTermMemory, ShortTermMemory, EntityMemory
from crewai.memory.storage import ChromaDBStorage

crew = Crew(
    agents=[fact_finder, synthesizer],
    tasks=[research_task, recall_task],
    process=Process.sequential,
    verbose=True,
    memory=True,
    long_term_memory=LongTermMemory(
        storage=ChromaDBStorage(path="./crew_memory/long_term")
    ),
    short_term_memory=ShortTermMemory(
        storage=ChromaDBStorage(path="./crew_memory/short_term")
    ),
    entity_memory=EntityMemory(
        storage=ChromaDBStorage(path="./crew_memory/entities")
    )
)

Three memory types exist:

• Long-term: Persistent facts across sessions (ChromaDB)
• Short-term: Current session context, auto-pruned
• Entity: Tracks entities (people, concepts) and their attributes

Agents with memory=True can read/write all three. The memory parameter on Crew enables the system; the agent flag opts individual agents in.

Pattern 4: Callback-Driven State for Complex Orchestration

When you need deterministic state transitions — validation, branching, external API calls between tasks — use task callbacks. The callback parameter on Task receives the task output and lets you mutate shared state before the next task runs.

# callback_state.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from typing import Dict, Any
import os
import json

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_API_BASE"),
    temperature=0.1
)

# Shared state container
class PipelineState:
    def __init__(self):
        self.data: Dict[str, Any] = {}
    
    def set(self, key: str, value: Any):
        self.data[key] = value
    
    def get(self, key: str, default=None):
        return self.data.get(key, default)

state = PipelineState()

def validate_research(output: str) -> str:
    """Callback: validate research output has 5 items, else retry."""
    lines = [l.strip() for l in output.split('\n') if l.strip().startswith(('-', '•', '*'))]
    if len(lines) < 5:
        raise ValueError(f"Expected 5 differences, got {len(lines)}. Output: {output}")
    state.set("validated_differences", lines)
    return output  # Return original output for context passing

def enrich_with_external_data(output: str) -> str:
    """Callback: simulate external enrichment."""
    diffs = state.get("validated_differences", [])
    enriched = []
    for d in diffs:
        # In reality, call an API here
        enriched.append(f"{d} [ENRICHED: verified via API]")
    state.set("enriched_differences", enriched)
    return output

researcher = Agent(
    role="Research Analyst",
    goal="Find key differences between REST and GraphQL",
    backstory="You produce exactly 5 bulleted differences.",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Write article from enriched differences",
    backstory="You use the enriched differences provided in context.",
    llm=llm,
    verbose=True
)

research_task = Task(
    description="Research 5 key differences between REST and GraphQL APIs. Output as a bulleted list only.",
    expected_output="Exactly 5 bullet points, each describing one difference",
    agent=researcher,
    callback=validate_research
)

enrich_task = Task(
    description="No-op task to trigger enrichment callback. The callback runs after research_task.",
    expected_output="Pass-through",
    agent=researcher,  # Reuse agent; callback does the work
    context=[research_task],
    callback=enrich_with_external_data
)

write_task = Task(
    description=(
        "Write a 150-word article using the enriched differences from context. "
        "Each difference has '[ENRICHED: verified via API]' suffix."
    ),
    expected_output="150-word article",
    agent=writer,
    context=[enrich_task]
)

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

result = crew.kickoff()
print(result)
print("\n--- Final State ---")
print(json.dumps(state.data, indent=2))

Run it:

python callback_state.py

Expected output: The validation callback ensures 5 differences exist before proceeding. The enrichment callback mutates state.data with enriched versions. The writer receives the enriched output via context=[enrich_task]. The final state dump shows both validated_differences and enriched_differences.

When to Use Callbacks vs. Context

Combining Patterns: A Production Template

Real pipelines mix these. Here’s a template that validates structured output, stores key facts in memory, and passes enriched context to writers.

# production_template.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List, Optional
import os

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_API_BASE"),
    temperature=0.1
)

class Fact(BaseModel):
    claim: str
    source: str
    confidence: float = Field(ge=0.0, le=1.0)

class ResearchResult(BaseModel):
    topic: str
    facts: List[Fact]
    gaps: List[str] = Field(default_factory=list)

researcher = Agent(
    role="Senior Researcher",
    goal="Produce verified, structured research",
    backstory="You output ResearchResult with high-confidence facts only.",
    llm=llm,
    verbose=True,
    memory=True
)

fact_checker = Agent(
    role="Fact Checker",
    goal="Validate claims and flag low-confidence items",
    backstory="You review research and identify gaps.",
    llm=llm,
    verbose=True,
    memory=True
)

writer = Agent(
    role="Technical Writer",
    goal="Write accurate articles from verified research",
    backstory="You only write claims backed by high-confidence facts.",
    llm=llm,
    verbose=True,
    memory=True
)

def validate_research(output: str) -> str:
    # In practice, parse JSON and validate against ResearchResult
    # For demo, we trust the Pydantic output
    return output

research_task = Task(
    description="Research 'async vs sync Python for I/O workloads'. Return ResearchResult with 5 facts, each with claim, source, confidence.",
    expected_output="ResearchResult JSON",
    agent=researcher,
    output_pydantic=ResearchResult,
    callback=validate_research
)

check_task = Task(
    description=(
        "Review the research. Identify any facts with confidence < 0.7 as gaps. "
        "Output updated ResearchResult with gaps populated."
    ),
    expected_output="ResearchResult JSON with gaps field filled",
    agent=fact_checker,
    context=[research_task],
    output_pydantic=ResearchResult
)

write_task = Task(
    description=(
        "Write a 200-word technical article using ONLY facts with confidence >= 0.7. "
        "Cite sources inline. Note any gaps identified by the fact checker."
    ),
    expected_output="200-word article with inline citations",
    agent=writer,
    context=[check_task]
)

crew = Crew(
    agents=[researcher, fact_checker, writer],
    tasks=[research_task, check_task, write_task],
    process=Process.sequential,
    verbose=True,
    memory=True
)

result = crew.kickoff()
print(result)

# Programmatic access to final structured output
final_research = check_task.output.pydantic
print(f"\nTopic: {final_research.topic}")
print(f"High-confidence facts: {len([f for f in final_research.facts if f.confidence >= 0.7])}")
print(f"Gaps identified: {final_research.gaps}")

Run it:

python production_template.py

Expected output: A three-stage pipeline where research produces typed output, fact-checking enriches it with gaps, and writing consumes only high-confidence claims. Memory retains facts across runs if you re-execute the crew.

Common Pitfalls

1. Forgetting context on the downstream task The upstream task produces output regardless, but the downstream agent never sees it without context=[upstream_task].

2. Assuming context passes structured objects It passes the raw string output. Use output_pydantic on the upstream task and reference field names explicitly in the downstream description.

3. Memory growing unbounded Long-term memory persists forever. Implement a cleanup job or use short-term memory for session-scoped state.

4. Callbacks mutating output unexpectedly A callback’s return value becomes the task’s output for context purposes. Return the original output if you only want side effects.

5. Agent reuse with conflicting memory If two agents share memory=True but have different roles, they read/write the same ChromaDB collections. Use distinct crews or namespaced memory paths for isolation.

Summary

Start with context. Add output_pydantic when string parsing hurts. Enable memory when state outlives a single crew run. Reach for callback when you need procedural control between tasks. Most production crews use all four.