Why your agent's tool needs to say pending, not missed, when data lags
Shipped
v0.7.0 added get_training_plan_progress, one MCP tool that returns the whole training plan in a single call: every prescribed workout with its verdict, plus the goal, days to race, and adherence. Before this, asking my fitness agent “how’s my plan going” made it drop into a shell and query the SQLite database by hand, because no tool returned the graded plan day by day. The release also added a short instruction, in the prompt and in the project’s own instructions, telling the agent to prefer the structured tools and never touch the database directly.
The part worth teaching is what the grading inside that tool actually has to get right. The plan’s source of truth is Garmin activity data, and Garmin data doesn’t arrive the instant a workout finishes; it syncs on a lag. A tool that grades against “today” will confidently tell you that you missed a run before your watch has even finished uploading it. The fix is a boundary: grade only up to the last date you actually have confirmed data for, and call everything after that “pending,” not “missed.”
Model the plan and the data you actually have
Before there’s a tool, there’s a data model. A prescribed plan is a list of items keyed by date. Actual activity is a separate list of what happened, also keyed by date. The two get joined to produce a verdict per day.
# plan_progress.py
from dataclasses import dataclass
from datetime import date
Verdict = str # "done" | "partial" | "missed" | "compliant" | "pending"
@dataclass(frozen=True)
class Prescribed:
day: date
label: str
target_units: float # 0 means a rest/off day
@dataclass(frozen=True)
class Actual:
day: date
units: float
target_units is generic on purpose. It could be miles, minutes, or reps; the grading logic below only cares whether it’s zero (a rest day) or positive (something is owed).
Grade against the frontier, not the calendar
The frontier is the last date you have confirmed data through, not date.today(). On the fitness side that’s whichever day the sync last landed for. Anything on or after the frontier is ungraded, full stop, regardless of what the wall clock says.
# plan_progress.py (continued)
DONE_FRACTION = 0.9
PARTIAL_FRACTION = 0.5
def grade_day(p: Prescribed, actual: Actual | None, frontier: date | None) -> Verdict:
if frontier is None or p.day >= frontier:
return "pending" # we don't have confirmed data for this day yet
if p.target_units == 0:
return "compliant" # rest days owe nothing
done_units = actual.units if actual else 0.0
frac = done_units / p.target_units
if frac >= DONE_FRACTION:
return "done"
if frac >= PARTIAL_FRACTION:
return "partial"
return "missed"
The frontier check runs first and short-circuits everything else. A day at or past the frontier is pending even if it’s a rest day, even if the target is zero, because the point isn’t “what would this day grade to if we had data,” it’s “do we have data for this day at all.” That ordering is the whole boundary.
Assemble the plan and wrap it as one tool
The aggregation walks the plan once, grades each day, and rolls up an adherence score over only the days that actually got graded (pending days aren’t credited or penalized).
# plan_progress.py (continued)
def plan_progress(plan: list[Prescribed], actuals: list[Actual], frontier: date | None) -> dict:
by_day = {a.day: a for a in actuals}
days, credit, graded = [], 0.0, 0
score = {"done": 1.0, "compliant": 1.0, "partial": 0.5, "missed": 0.0}
for p in plan:
verdict = grade_day(p, by_day.get(p.day), frontier)
if verdict != "pending":
graded += 1
credit += score[verdict]
days.append({"date": p.day.isoformat(), "prescribed": p.label, "verdict": verdict})
adherence = round(credit / graded, 2) if graded else None
return {"days": days, "adherence": adherence}
Now expose it as a single tool instead of leaving the agent to reconstruct this by querying the database and doing the join itself. I used the Claude Agent SDK’s @tool decorator, which takes a name, a description, and an input schema, and create_sdk_mcp_server, which registers the tool and exposes it to the agent as mcp__<server>__<tool> (Claude Agent SDK Python reference).
# tool_server.py
import json
from datetime import date
from claude_agent_sdk import tool, create_sdk_mcp_server
from plan_progress import Prescribed, Actual, plan_progress
# Stand-ins for real data access; swap these for your DB reads.
def load_plan() -> list[Prescribed]: ...
def load_actuals() -> list[Actual]: ...
def load_frontier() -> date | None: ... # last date your sync has confirmed
@tool(
"get_plan_progress",
"Full day-by-day plan progress: every prescribed item with its graded "
"verdict, plus adherence. Never grades a day the data hasn't caught up to.",
{},
)
async def get_plan_progress(_args: dict) -> dict:
result = plan_progress(load_plan(), load_actuals(), load_frontier())
return {"content": [{"type": "text", "text": json.dumps(result)}]}
server = create_sdk_mcp_server(name="plan", version="1.0.0", tools=[get_plan_progress])
This is the same shape Anthropic’s own guidance describes: build a tool that returns what the caller actually wants in one call instead of exposing the pieces and letting the model assemble them, the same way a schedule_event tool beats separate list_events and create_event calls (Anthropic: Writing tools for AI agents). It also matches the broader advice to keep the agent’s own logic simple and put the design effort into the tool instead, since a well-designed tool does more to keep a system reliable than a longer prompt does (Anthropic: Building effective agents). The agent gets one call and one clean answer instead of a database connection.
Run it and check the boundary
Fill in the three loader stubs with real data and the aggregation runs standalone, no agent required:
# run_demo.py
from datetime import date
from plan_progress import Prescribed, Actual, plan_progress
plan = [
Prescribed(date(2026, 6, 19), "5mi easy", 5.0),
Prescribed(date(2026, 6, 20), "8mi easy", 8.0),
Prescribed(date(2026, 6, 21), "rest", 0.0),
Prescribed(date(2026, 6, 22), "5mi tempo", 5.0),
Prescribed(date(2026, 6, 23), "12mi long run", 12.0),
]
actuals = [Actual(date(2026, 6, 20), 8.1)]
frontier = date(2026, 6, 22) # sync only has confirmed data through the 21st
result = plan_progress(plan, actuals, frontier)
for d in result["days"]:
print(d["date"], d["prescribed"], "->", d["verdict"])
print("adherence:", result["adherence"])
Running it prints:
2026-06-19 5mi easy -> missed
2026-06-20 8mi easy -> done
2026-06-21 rest -> compliant
2026-06-22 5mi tempo -> pending
2026-06-23 12mi long run -> pending
adherence: 0.67
The 22nd is the frontier date itself, and it reads pending, not missed, even though it’s already prescribed and the day has technically started. Adherence lands on 0.67 because only three days are graded (one done, one missed, one compliant); the two days at or past the frontier don’t enter the score at all.
Pin the boundary with a direct test so a future refactor can’t quietly loosen it:
# test_plan_progress.py
from datetime import date
from plan_progress import Prescribed, Actual, grade_day
FRONTIER = date(2026, 6, 22)
def test_day_at_frontier_is_pending_not_missed():
p = Prescribed(FRONTIER, "5mi tempo", 5.0)
assert grade_day(p, None, FRONTIER) == "pending"
$ pytest -q test_plan_progress.py
.....
5 passed in 0.01s
Loosen the check from p.day >= frontier to p.day > frontier and this specific test fails: assert 'missed' == 'pending'. That’s the exact off-by-one that would make the frontier date itself gradeable a day early, before the sync for it has actually landed.
Gotchas
Grading against date.today() instead of a data frontier. The moment your data source lags, calendar-based grading tells the user they failed something before you actually know what happened. Track the last date you have confirmed data through and grade strictly before it; everything on or after is pending, never done or missed. This is the general form of the pattern stream-processing systems call a watermark, a boundary that says “I believe all data up to this point has arrived” (Databricks: watermarks and late data).
Wiring a new tool into every calling context by default. This project has another loop that runs on a schedule and calls a fixed, cheap set of read-only tools. The new plan-progress tool does more work per call than that loop needs, so it was deliberately left off that loop’s tool list instead of being added automatically. A new capability shouldn’t join every existing call site just because it exists; keep an explicit allow-list for the loops that care about staying cheap, and add to it on purpose.
Composing a new tool on top of an internal function with assumed fields. The plan-progress tool wraps an existing internal assembly function that expects a race date to be present. A crafted or edge-case plan without that field would crash the tool at call time instead of degrading gracefully. Read optional fields with .get(), not a bare subscript, and write a test that exercises the missing-field path directly, not just the happy path.
Sources
- Anthropic: Writing tools for AI agents — build high-level tools that return what the caller wants instead of primitives the model has to assemble.
- Anthropic: Building effective agents — keep the system simple and invest the effort in tool design, not agent logic.
- Claude Agent SDK Python reference — the
@tooldecorator,create_sdk_mcp_server, and themcp__server__toolnaming convention. - Databricks: watermarks and late data — the watermark pattern for deciding when data is complete enough to grade.