Typed results

Production patternsChapter 12 of 144 min read

Extract structured, runtime-validated Python objects from agent tasks using Pydantic schemas instead of unstructured markdown or fragile JSON strings.

The problem#

Prompting language models for JSON without schema enforcement introduces parsing failures. Test this behavior in extract_json.py:

extract_json.py

1# extract_json.py
2
3import json
4
5from crewai import Agent, Crew, Task
6from crewai_tools import FileReadTool
7from dotenv import load_dotenv
8
9load_dotenv()
10
11reader = Agent(
12    role="Contract analyst",
13    goal="Pull every commercially important term out of a contract, with its section number",
14    backstory="You read contracts line by line for a procurement team.",
15    tools=[FileReadTool()],
16)
17
18extract = Task(
19    description="Read the contract at {contract_path} and pull out its key terms.",
20    expected_output=(
21        "A JSON object with these keys: provider, customer, effective_date, "
22        "initial_term_months, auto_renews, renewal_term_months, notice_days, annual_fee_usd."
23    ),
24    agent=reader,
25)
26
27result = Crew(agents=[reader], tasks=[extract]).kickoff(
28    inputs={"contract_path": "contracts/acme-saas-agreement.md"}
29)
30print(result.raw)
31
32terms = json.loads(result.raw)
33print(terms["effective_date"])

Run the script to observe the error:

Terminal

1(env)$ python extract_json.py

The execution fails because the model wraps the JSON string in markdown code fences:

Output

```json{  "provider": "Acme Analytics, Inc.",  "customer": "Northwind Traders LLC",  "effective_date": "March 1, 2026",  "initial_term_months": 12,  "auto_renews": true,  "renewal_term_months": 12,  "notice_days": 60,  "annual_fee_usd": 18000}```Traceback (most recent call last):  ...json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

Core failure modes of unvalidated JSON generation:

Markdown code fences: Surrounding triple backticks break standard json.loads parsing.

Inconsistent type formatting: Dates oscillate between "March 1, 2026" and "2026-03-01".

Missing schema guarantees: Keys may be omitted or returned as strings rather than numbers.

Delete the prototype script:

Terminal

1(env)$ rm extract_json.py

Pydantic models#

Define typed data schemas with Pydantic in contract_crew/models.py:

contract_crew/models.py

1# contract_crew/models.py
2
3from datetime import date
4
5from pydantic import BaseModel, Field
6
7
8class Clause(BaseModel):
9    section: str = Field(description="The section number, like 2.2")
10    topic: str = Field(description="A short label for the clause, like Auto-renewal")
11    text: str = Field(description="The clause, quoted word for word from the contract")
12
13
14class ContractTerms(BaseModel):
15    provider: str = Field(description="Legal name of the company providing the service")
16    customer: str = Field(description="Legal name of the customer")
17    effective_date: date = Field(description="The date the agreement starts")
18    initial_term_months: int = Field(description="Length of the initial term, in months")
19    auto_renews: bool = Field(description="True if the agreement renews automatically")
20    renewal_term_months: int | None = Field(description="Length of each renewal term in months, or null if it does not renew")
21    notice_days: int | None = Field(description="Days of notice needed before a term ends to stop it renewing, or null")
22    annual_fee_usd: float = Field(description="The annual fee, in US dollars")
23    key_clauses: list[Clause] = Field(
24        description="Every clause that affects cost, deadlines, data, or liability, or that lets one party change the deal"
25    )

Each field description guides the language model during extraction, while Pydantic guarantees type casting and date parsing at runtime.

Attach ContractTerms to extract_terms in contract_crew/crew.py:

contract_crew/crew.py

1# contract_crew/crew.py
2
3from crewai import Agent, Crew, Process, Task
4from crewai.project import CrewBase, agent, crew, task
5from crewai_tools import FileReadTool
6
7from contract_crew.models import ContractTerms
8
9
10@CrewBase
11class ContractReviewCrew:
12    """Reads a contract, flags the risks, and checks the memo against the source."""
13
14    agents_config = "config/agents.yaml"
15    tasks_config = "config/tasks.yaml"
16
17    def __init__(self):
18        self.read_file = FileReadTool()
19
20    @agent
21    def reader(self) -> Agent:
22        return Agent(config=self.agents_config["reader"], tools=[self.read_file], verbose=True)
23
24    @agent
25    def risk_analyst(self) -> Agent:
26        return Agent(config=self.agents_config["risk_analyst"], verbose=True)
27
28    @agent
29    def fact_checker(self) -> Agent:
30        return Agent(config=self.agents_config["fact_checker"], tools=[self.read_file], verbose=True)
31
32    @task
33    def extract_terms(self) -> Task:
34        return Task(config=self.tasks_config["extract_terms"], output_pydantic=ContractTerms)
35
36    @task
37    def assess_risks(self) -> Task:
38        return Task(config=self.tasks_config["assess_risks"])
39
40    @task
41    def fact_check(self) -> Task:
42        return Task(config=self.tasks_config["fact_check"])
43
44    @crew
45    def crew(self) -> Crew:
46        return Crew(
47            agents=self.agents,
48            tasks=self.tasks,
49            process=Process.sequential,
50            verbose=True,
51        )

Update contract_crew/config/tasks.yaml to focus purely on extraction criteria, since the Pydantic schema defines output structure:

YAML

1extract_terms:
2  description: >
3    Read the contract at {contract_path}. List its key terms: parties, dates,
4    term and renewal, fees and payment, service levels, data handling,
5    termination, liability, and any clause that lets one party change the
6    deal on its own.
7  expected_output: >
8    The contract's key terms, with every important clause quoted word for
9    word along with its section number.
10  agent: reader

Access the parsed model instance in main.py:

main.py

1# main.py
2
3from dotenv import load_dotenv
4
5from contract_crew.crew import ContractReviewCrew
6
7load_dotenv()
8
9
10def main():
11    result = ContractReviewCrew().crew().kickoff(
12        inputs={
13            "contract_path": "contracts/acme-saas-agreement.md",
14            "customer": "Northwind Traders",
15        }
16    )
17
18    terms = result.tasks_output[0].pydantic
19    print(f"Effective date: {terms.effective_date}")
20    print(f"Renews automatically: {terms.auto_renews}")
21    print(f"Notice needed: {terms.notice_days} days")
22    print(f"Clauses found: {len(terms.key_clauses)}")
23    print()
24    print(result.raw)
25
26
27if __name__ == "__main__":
28    main()

Run the pipeline:

Terminal

1(env)$ python main.py

Review the typed output:

Output

Effective date: 2026-03-01Renews automatically: TrueNotice needed: 60 daysClauses found: 20

The output provides native Python types (datetime.date, bool, int), giving downstream agents and application code structured, validated inputs.