CrewAICrewAI 1.15 · Python 3.10 to 3.13
Dashboard
0%
1
Curious builder0 XP earned · 300 to level 2
0 daysFinish a lesson to begin
Badge collection0 of 6 unlocked
35 small wins to finish your pathNext lesson →

Flow state with a Pydantic model

Flow[Ticket] gives a flow a typed state: a Pydantic model with named fields and defaults that every method reads and writes.

Last updated: 28 Sep, 2026 · CrewAI 1.15

The last flow's state was a dictionary, and a typo in a key would only show up when that line ran. A model class fixes the fields up front and checks the inputs.

Typing the state

Write a Pydantic model for the state, then parameterise the flow with it. self.state becomes an instance of that model.

python
class Ticket(BaseModel):
    message: str = ""    # every field needs a default
    order_id: str = ""

class Desk(Flow[Ticket]):    # self.state is now a Ticket
    ...
Project files used on this pageThis lesson builds on a project from earlier lessons. The code below imports these files. Click a file to see its code, or follow the link to the lesson that wrote it. To run the code yourself, keep them in the same folder.
View the code here
tools.py
from crewai.tools import tool

ORDERS = {"A17": "shipped on 3 March", "C40": "waiting for stock"}


@tool
def lookup_order(order_id: str) -> str:
    """Look up an order's shipping status by its id, such as A17."""
    status = ORDERS.get(order_id)
    return f"{order_id} {status}." if status else f"{order_id} is not an order we have."
shop_llm.py
import json
import os
import re

from crewai import BaseLLM

os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"
os.environ["CREWAI_TRACING_ENABLED"] = "false"
os.environ["CREWAI_DISABLE_VERSION_CHECK"] = "true"


class ShopLLM(BaseLLM):
    script: list = []

    def supports_function_calling(self):
        return True

    def call(self, messages, tools=None, **kwargs):
        if isinstance(messages, str):
            messages = [{"role": "user", "content": messages}]
        if self.script:
            return self.script.pop(0)
        names = [t["function"]["name"] for t in tools or []]
        return self.decide(messages, names)

    def decide(self, messages, tools):
        last = messages[-1]
        if last["role"] == "tool":
            return last["content"]
        text = last["content"]
        orders = re.findall(r"\b[A-Z]\d+\b", text)
        wanted = "refund_order" if "refund" in text.lower() else "lookup_order"
        matches = [name for name in tools if name.endswith(wanted)]
        if orders and matches:
            args = json.dumps({"order_id": orders[0]})
            return [{"id": f"call_{orders[0]}", "type": "function",
                     "function": {"name": matches[0], "arguments": args}}]
        if "working with:" in text:
            context = text.split("working with:")[1].strip().split("\n\n")[0]
            return f"Dear customer, {context}"
        if orders:
            return f"I have no way to look up {orders[0]} yet."
        return "Hello. Which order is this about?"

The model

Ticket holds the three things the desk tracks. Every field needs a default.

python
import re

import shop_llm
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel


class Ticket(BaseModel):
    message: str = ""
    order_id: str = ""
    reply: str = ""

The steps

Flow[Ticket] makes self.state a Ticket. The methods pass nothing to each other; they share the state.

python
class Desk(Flow[Ticket]):
    @start()
    def read_ticket(self):
        found = re.findall(r"\b[A-Z]\d+\b", self.state.message)
        self.state.order_id = found[0] if found else ""

    @listen(read_ticket)
    def answer(self):
        self.state.reply = f"Looking into {self.state.order_id}."

A typed state after a run

inputs fills message; each step writes its field; your code reads them after.

Example
import os
os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"

import re

import shop_llm
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel


class Ticket(BaseModel):
    message: str = ""
    order_id: str = ""
    reply: str = ""


class Desk(Flow[Ticket]):
    @start()
    def read_ticket(self):
        found = re.findall(r"\b[A-Z]\d+\b", self.state.message)
        self.state.order_id = found[0] if found else ""

    @listen(read_ticket)
    def answer(self):
        self.state.reply = f"Looking into {self.state.order_id}."

desk = Desk(suppress_flow_events=True)
desk.kickoff(inputs={"message": "Where is my order A17?"})
print(desk.state.order_id, "|", desk.state.reply)

What the state held

  • message came from inputs and validated as a string.
  • order_id was written by read_ticket, and reply by answer.
  • After the run the state still holds what each step wrote, for your code to read.

An input of the wrong type

Pass None where a string is declared and the kickoff is refused before any method runs.

Example
import os
os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"

import re

import shop_llm
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel


class Ticket(BaseModel):
    message: str = ""
    order_id: str = ""
    reply: str = ""


class Desk(Flow[Ticket]):
    @start()
    def read_ticket(self):
        found = re.findall(r"\b[A-Z]\d+\b", self.state.message)
        self.state.order_id = found[0] if found else ""

    @listen(read_ticket)
    def answer(self):
        self.state.reply = f"Looking into {self.state.order_id}."

desk = Desk(suppress_flow_events=True)
desk.kickoff(inputs={"message": None})

A dictionary state would have stored None and failed later inside re.findall. A key with no matching field is dropped without a word, so a misspelled input name still needs care.

Dictionary state vs typed state

Dictionary stateFlow[Ticket]
FieldsWhatever you writeDeclared up front
A wrong-type inputStored, fails laterRefused at kickoff
A typo in a keySilent until it runsCaught by the field list

When to type the state

  • Any flow whose state has more than one field or is written by several steps.
  • When you want inputs checked before the flow starts running.
Watch out. Every field needs a default, because the state exists before inputs are applied. A misspelled input key is dropped silently, so the field keeps its default with no warning.
Try it yourself
  • Pass inputs={"mesage": "Where is A17?"} and print the state.
  • Give order_id no default and read the error.
  • Print desk.state.model_dump() after a run.

Little by little, you're building something great.