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.
class Ticket(BaseModel):
message: str = "" # every field needs a default
order_id: str = ""
class Desk(Flow[Ticket]): # self.state is now a Ticket
...View the code here
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."
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.
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.
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.
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)A17 | Looking into A17.
What the state held
- message came from
inputsand validated as a string. - order_id was written by
read_ticket, andreplybyanswer. - 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.
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})pydantic_core._pydantic_core.ValidationError: 1 validation error for StateWithId
message
Input should be a valid string [type=string_type, input_value=None, input_type=NoneType]
For further information visit https://errors.pydantic.dev/2.12/v/string_type
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "main.py", line 29, in <module>
desk.kickoff(inputs={"message": None})
ValueError: Invalid inputs for structured state: 1 validation error for StateWithId
message
Input should be a valid string [type=string_type, input_value=None, input_type=NoneType]
For further information visit https://errors.pydantic.dev/2.12/v/string_typeA 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 state | Flow[Ticket] | |
|---|---|---|
| Fields | Whatever you write | Declared up front |
| A wrong-type input | Stored, fails later | Refused at kickoff |
| A typo in a key | Silent until it runs | Caught 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.
inputs are applied. A misspelled input key is dropped silently, so the field keeps its default with no warning.Related
- Previous: First flow: steps in plain Python
- Next: Router: choosing a path in a flow
- Reference: Mastering flow state
- Pass
inputs={"mesage": "Where is A17?"}and print the state. - Give
order_idno default and read the error. - Print
desk.state.model_dump()after a run.
Little by little, you're building something great.