Skip to content
/

Your own tool

Production patternsChapter 13 of 147 min read

Offload calendar arithmetic and contract deadline calculations from probabilistic language models to deterministic, unit-tested Python functions wrapped as custom tools.


Drawing diagram…
InputAgentToolResultThe finished crew. Data flows between agents; dashed lines indicate tool calls.

Write the tests first#

Define unit tests in tests/test_tools.py covering month additions, notice offsets, leap years, and rollover conditions:

tests/test_tools.py
# tests/test_tools.py

from datetime import date

import pytest

from contract_crew.tools import add_months, next_notice_deadline


@pytest.mark.parametrize(
    "start, months, expected",
    [
        (date(2026, 3, 1), 12, date(2027, 3, 1)),
        (date(2026, 1, 31), 1, date(2026, 2, 28)),
        (date(2027, 11, 30), 3, date(2028, 2, 29)),
    ],
)
def test_add_months(start, months, expected):
    assert add_months(start, months) == expected


def test_deadline_is_notice_days_before_the_term_ends():
    term_end, deadline = next_notice_deadline(
        effective_date=date(2026, 3, 1),
        initial_term_months=12,
        renewal_term_months=12,
        notice_days=60,
        today=date(2026, 10, 19),
    )
    assert term_end == date(2027, 2, 28)
    assert deadline == date(2026, 12, 30)


def test_deadline_on_the_last_day_still_counts():
    _, deadline = next_notice_deadline(
        effective_date=date(2026, 3, 1),
        initial_term_months=12,
        renewal_term_months=12,
        notice_days=60,
        today=date(2026, 12, 30),
    )
    assert deadline == date(2026, 12, 30)


def test_missed_deadline_rolls_over_to_the_next_term():
    term_end, deadline = next_notice_deadline(
        effective_date=date(2026, 3, 1),
        initial_term_months=12,
        renewal_term_months=12,
        notice_days=60,
        today=date(2026, 12, 31),
    )
    assert term_end == date(2028, 2, 29)
    assert deadline == date(2027, 12, 31)

Key test cases:

  1. Month-end snapping: January 31 plus one month resolves to February 28 (or 29 in leap years).
  1. Notice period offset: A 12-month term starting March 1, 2026 ends February 28, 2027; a 60-day notice cutoff falls on December 30, 2026.
  1. Term rollover: A missed deadline rolls over to the subsequent 12-month term.

Execute pytest to confirm test failure before implementing the logic:

Terminal
(env)$ python -m pytest

Write the code#

Implement the calendar arithmetic functions in contract_crew/tools.py:

contract_crew/tools.py
# contract_crew/tools.py

import calendar
from datetime import date, timedelta


def add_months(day, months):
    month_index = day.month - 1 + months
    year = day.year + month_index // 12
    month = month_index % 12 + 1
    last_day = calendar.monthrange(year, month)[1]
    return date(year, month, min(day.day, last_day))


def next_notice_deadline(effective_date, initial_term_months, renewal_term_months, notice_days, today):
    term_end = add_months(effective_date, initial_term_months) - timedelta(days=1)
    while term_end - timedelta(days=notice_days) < today:
        term_start = term_end + timedelta(days=1)
        term_end = add_months(term_start, renewal_term_months) - timedelta(days=1)
    return term_end, term_end - timedelta(days=notice_days)

Passing today explicitly as a parameter enables deterministic unit testing across past, present, and leap-year dates.

Run pytest to verify the implementation:

Terminal
(env)$ python -m pytest

tests/test_crew.py .                                                     [ 14%]
tests/test_tools.py ......                                               [100%]

======================== 7 passed, 11 warnings in 3.00s ========================

Turn it into a tool#

Wrap the logic in a CrewAI BaseTool subclass with an explicit Pydantic argument schema in contract_crew/tools.py:

contract_crew/tools.py
# contract_crew/tools.py

import calendar
from datetime import date, timedelta

from crewai.tools import BaseTool
from pydantic import BaseModel, Field


def add_months(day, months):
    month_index = day.month - 1 + months
    year = day.year + month_index // 12
    month = month_index % 12 + 1
    last_day = calendar.monthrange(year, month)[1]
    return date(year, month, min(day.day, last_day))


def next_notice_deadline(effective_date, initial_term_months, renewal_term_months, notice_days, today):
    term_end = add_months(effective_date, initial_term_months) - timedelta(days=1)
    while term_end - timedelta(days=notice_days) < today:
        term_start = term_end + timedelta(days=1)
        term_end = add_months(term_start, renewal_term_months) - timedelta(days=1)
    return term_end, term_end - timedelta(days=notice_days)


class NoticeDeadlineInput(BaseModel):
    effective_date: date = Field(description="The date the agreement started, as YYYY-MM-DD")
    initial_term_months: int = Field(description="Length of the initial term, in months")
    renewal_term_months: int = Field(description="Length of each renewal term, in months")
    notice_days: int = Field(description="Days of notice needed before a term ends to stop it renewing")


class NoticeDeadlineTool(BaseTool):
    name: str = "notice_deadline_calculator"
    description: str = (
        "Works out when the current contract term ends and the last day to send "
        "a notice of non-renewal. Use it for any renewal or notice date instead "
        "of calculating dates yourself."
    )
    args_schema: type[BaseModel] = NoticeDeadlineInput

    def _run(self, effective_date, initial_term_months, renewal_term_months, notice_days):
        today = date.today()
        term_end, deadline = next_notice_deadline(
            effective_date=date.fromisoformat(str(effective_date)),
            initial_term_months=int(initial_term_months),
            renewal_term_months=int(renewal_term_months),
            notice_days=int(notice_days),
            today=today,
        )
        days_left = (deadline - today).days
        return (
            f"The next term that can still be stopped ends on {term_end.isoformat()}. "
            f"Notice of non-renewal must be sent by {deadline.isoformat()}, "
            f"which is {days_left} days from today ({today.isoformat()})."
        )

Test the tool directly in an interactive Python session:

Terminal
(env)$ python -c '
from contract_crew.tools import NoticeDeadlineTool
print(NoticeDeadlineTool().run(effective_date="2026-03-01", initial_term_months=12, renewal_term_months=12, notice_days=60))
'

Give the tool to the agents#

Provision NoticeDeadlineTool to both risk_analyst and fact_checker in contract_crew/crew.py:

contract_crew/crew.py
# contract_crew/crew.py

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import FileReadTool

from contract_crew.models import ContractTerms
from contract_crew.tools import NoticeDeadlineTool


@CrewBase
class ContractReviewCrew:
    """Reads a contract, flags the risks, and checks the memo against the source."""

    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    def __init__(self):
        self.read_file = FileReadTool()
        self.notice_deadline = NoticeDeadlineTool()

    @agent
    def reader(self) -> Agent:
        return Agent(config=self.agents_config["reader"], tools=[self.read_file], verbose=True)

    @agent
    def risk_analyst(self) -> Agent:
        return Agent(config=self.agents_config["risk_analyst"], tools=[self.notice_deadline], verbose=True)

    @agent
    def fact_checker(self) -> Agent:
        return Agent(
            config=self.agents_config["fact_checker"],
            tools=[self.read_file, self.notice_deadline],
            verbose=True,
        )

    @task
    def extract_terms(self) -> Task:
        return Task(config=self.tasks_config["extract_terms"], output_pydantic=ContractTerms)

    @task
    def assess_risks(self) -> Task:
        return Task(config=self.tasks_config["assess_risks"])

    @task
    def fact_check(self) -> Task:
        return Task(config=self.tasks_config["fact_check"])

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

Update contract_crew/config/tasks.yaml to instruct the agents to invoke the tool for all renewal calculations:

YAML
assess_risks:
  description: >
    Using the key terms, write a risk memo for {customer}. Pick the five terms
    most likely to cost them money or catch them out, and say what they
    should do about each one. For any renewal or notice date, use the notice
    deadline calculator instead of working out dates yourself. The memo is
    for the finance team, so don't mention your tools in it.
  expected_output: >
    A Markdown memo of no more than 350 words: a two-sentence summary, then a
    numbered list of five risks. Each risk has a section number, why it
    matters, and a recommended action. Give the exact date for every
    deadline.
  agent: risk_analyst

fact_check:
  description: >
    Check every claim in the risk memo against the contract at {contract_path}.
    Fix any number, date, or section reference that does not match the
    contract, and remove anything the contract does not say. Check renewal
    and notice dates with the notice deadline calculator, and don't mention
    your tools in the memo.
  expected_output: >
    The corrected memo in Markdown, followed by a "Corrections" section that
    lists each change you made, or says "None".
  agent: fact_checker
  output_file: output/review.md

Run the crew:

Terminal
(env)$ python main.py

The agent execution log records tool calls with exact arguments:

Output
Tool notice_deadline_calculator executed with result: The next term that can still bestopped ends on 2027-02-28. Notice of non-renewal must be sent by 2026-12-30, which is 72days from today (2026-10-19).

Review the verified memo and automated corrections:

Output
### 1. Auto-Renewal Without Clear Exit StrategyRisk: The agreement auto-renews for 12-month terms unless Northwind provides 60 days'written notice before each renewal. The last day to send notice for the first renewal isDecember 30, 2026. ...
### Corrections1. Auto-Renewal Deadline: Updated the last day to send notice for the first renewal from   August 31, 2026, to December 30, 2026 (calculated using the correct renewal term end   date).

The fact-checker verifies dates against the calculator, correcting erroneous preliminary estimates in the finalized risk memo.

Review the final project architecture:

Project structure
contract-crew├── .env├── .gitignore├── contract_crew│   ├── __init__.py│   ├── config│   │   ├── agents.yaml│   │   └── tasks.yaml│   ├── crew.py│   ├── models.py│   └── tools.py├── contracts│   └── acme-saas-agreement.md├── explain.py├── main.py├── output│   └── review.md├── summarize.py└── tests    ├── __init__.py    ├── test_crew.py    └── test_tools.py