Lesson 09a: Agents Router — Deep Dive#

"937 lines, 15 endpoints, 7 helper functions — the biggest file in the backend."

backend/app/routers/agents.py is the largest source file in the entire project (937 lines) — bigger than every model, schema, and service file. This lesson goes through it top to bottom.

Module-Level Helper Functions#

Seven plain functions sit above the route handlers. None of them are FastAPI dependencies (no Depends() involved) — they're just shared logic called directly from inside the endpoints.

normalize_due_date(deadline)#

def normalize_due_date(deadline: str | None) -> str | None:
    if not deadline:
        return None
    simple_deadlines = ["today", "tomorrow", "next week", "this week"]
    if deadline.lower() in simple_deadlines:
        return deadline
    try:
        parsed_date = datetime.fromisoformat(deadline)
        if parsed_date < datetime.now():
            return None
        return deadline
    except ValueError:
        return deadline

Takes whatever string the LLM extracted as a deadline from an email (analyze_email's deadline field) and decides whether to keep it. Three outcomes:

• A recognized relative phrase ("tomorrow") → kept as-is.
• A parseable ISO date in the past → discarded (returns None) — this stops the task list from filling up with tasks due "yesterday" just because the LLM echoed a date mentioned in the email body.
• Anything else, including a parseable future ISO date or unparseable free text → kept as-is, unmodified.

Note what it does not do: it never reformats the date. A future ISO date and a free-text phrase like "by end of quarter" are both stored on Task.due_date (a plain String(100) column) exactly as the LLM produced them — there's no canonical date type here.

agent_has_tool(agent, tool_name)#

def agent_has_tool(agent: models.Agent, tool_name: str) -> bool:
    if not agent.enabled_tools:
        return False
    tools = [tool.strip() for tool in agent.enabled_tools.split(",")]
    return tool_name in tools

Agent.enabled_tools is a single Text column storing a comma-separated string like "email_analyzer,task_creator" — there is no AgentTool join table. This function is the entire authorization model for tool access in this backend (see Lesson 13: AI Agents and Lesson 19: Prompt Engineering & Tool Calling). It's duplicated verbatim in routers/documents.py — there is no shared utils.py it lives in.

log_tool_call(...)#

Builds and db.add()s a models.ToolCall row — but does not call db.commit(). Every caller is responsible for committing afterward, usually bundled with other writes in the same transaction. This function is called after essentially every LLM operation in this file: email analysis, task creation, and every step of the multi-agent workflow.

create_workflow_run(...) and log_workflow_step(...)#

create_workflow_run is the odd one out — it does call db.commit() and db.refresh() itself, because the caller needs the generated workflow_run.id immediately to attach subsequent WorkflowStep rows to it. log_workflow_step mirrors log_tool_call's pattern (add, don't commit) for the same reason: the caller batches several steps into one commit.

safe_float(value) and safe_int(value)#

def safe_float(value):
    try:
        if value is None:
            return None
        return float(value)
    except (ValueError, TypeError):
        return None

Defensive coercion helpers used only inside the dashboard endpoint, because ToolCall.estimated_cost and WorkflowRun.quality_score are stored as String columns (see Lesson 06: Models for why), not numeric types — so aggregating them requires converting strings back to numbers, tolerating None or malformed values along the way.

The Endpoints#

Agent CRUD: POST /, GET /, GET /{id}#

Straightforward create/list/get-by-id against models.Agent. GET / orders by Agent.id.desc() — newest first, no pagination, no filtering. GET /{id} is the first place you see the "fetch or 404" pattern that repeats through the rest of the file (see Lesson 10 for why this is a missed dependency-injection opportunity).

POST /{agent_id}/chat#

The single most important endpoint in the file. Walked through fully in Lesson 03: Request Lifecycle, so only the parts that lesson doesn't cover:

• Timing is manual: start_time = time.time() before the LLM call, latency_ms = int((time.time() - start_time) * 1000) after — no middleware or decorator does this automatically.
• The try/except around generate_agent_response is the template every other LLM call in this file follows: on failure, produce a user-facing fallback string ("Sorry, I could not generate a response right now."), never let the exception propagate to the client as a raw 500 for this particular endpoint. Contrast this with email/analyze and workflows/email below, which do let exceptions become 500s.
• Both the user message and the assistant reply are persisted regardless of success or failure — even a failed LLM call leaves a Message row with the fallback text, so the conversation transcript stays coherent.

GET /{agent_id}/runs#

Read-only history of AgentRun rows for one agent, newest first. No pagination — for an agent with thousands of chats, this returns everything.

POST /{agent_id}/email/analyze#

Gated by agent_has_tool(agent, "email_analyzer") — a 403 if the tool isn't enabled on this agent. Two nested tool calls happen inside one endpoint:

1. analyze_email() is called and logged as a "email_analyzer" tool call.
2. If agent_has_tool(agent, "task_creator") is also true, every string in result["action_items"] becomes a Task row, and that whole batch is logged as a second, separate "task_creator" tool call.

Both tool calls, plus any tasks created, are committed in a single transaction at the end — if task creation somehow failed partway through, the whole try block's except clause logs one failure record and raises a 500, but by then the email-analysis tool call has already been added to the session (not yet committed) and would roll back too. This is worth tracing carefully: the earlier log_tool_call for "email_analyzer" success is only durable once the final db.commit() at the end of the try block runs — an exception raised between logging it and that commit discards it, and a new, failure-flavored ToolCall (tool_name="email_analyzer_or_task_creator") is logged instead. The system never has both a success and a failure record for the same request.

GET /{agent_id}/tasks and PUT /tasks/{task_id}/status#

Note the URL for the update endpoint: /tasks/{task_id}/status, not /agents/{agent_id}/tasks/{task_id}/status. Because the router prefix is /agents, the actual mounted path is /agents/tasks/{task_id}/status — a task is looked up only by its own ID, with no verification that it belongs to any particular agent. status_data.status is validated against a hardcoded allow-list (["open", "in_progress", "completed"]) inline in the handler, not via a Pydantic Literal type on schemas.TaskStatusUpdate (contrast with EmailAnalysisResponse.priority, which does use Literal — see Lesson 07: Schemas).

GET /{agent_id}/tool-calls#

Plain read of the ToolCall observability log for one agent — the human-facing view of everything log_tool_call() has written.

POST /{agent_id}/workflows/email — the biggest endpoint#

Roughly 280 lines on its own. Orchestrates the four-agent pipeline from Lesson 08c and Lesson 12: analysis → tasks → reply → review. What this router-level code adds on top of the service functions:

• Creates a WorkflowRun row first, before anything else runs, with status="running" — so even a workflow that crashes on step one leaves an auditable row.
• Calls estimate_openai_cost() from monitoring_service after every agent step (analysis, reply, review) and attaches model_name, estimated_tokens, estimated_cost to the corresponding ToolCall.
• Writes two parallel records per step: a WorkflowStep (for the step-by-step timeline UI) and a ToolCall (for the cost/observability dashboard) — same event, two different tables, two different consumers.
• On any exception, it logs a failure ToolCall, marks workflow_run.status = "failed", logs a "workflow_error" WorkflowStep, commits, then re-raises as an HTTPException(500, ...).
• Reads os.getenv("OPENAI_MODEL", "gpt-4o-mini") directly, three separate times, purely to pass a display string into estimate_openai_cost() — the router doesn't call OpenAI itself here, but it does read an environment variable directly rather than through a service (see Lesson 27: Configuration).

GET /{agent_id}/workflow-runs and GET /workflow-runs/{workflow_run_id}/steps#

History endpoints mirroring runs/tool-calls above, but for workflows. Note the second one is not nested under /agents/{agent_id}/... — a workflow run's steps are looked up purely by workflow_run_id, with no agent-ownership check, the same pattern as the task-status-update endpoint above.

GET /{agent_id}/dashboard#

Pure aggregation — no writes. Pulls every Task, ToolCall, and WorkflowRun for the agent into memory, then computes counts and averages in Python (len([x for x in tasks if x.status == "open"]), etc.) rather than with SQL COUNT/AVG/GROUP BY. For an agent with a large history this means every dashboard load fetches full row data just to count it — see Lesson 24: Performance for the concrete cost of this pattern and how you'd rewrite it with aggregate queries.

GET /{agent_id}/messages#

The only endpoint in the file with real pagination (limit: int = 50, offset: int = 0 as query parameters) — every other list endpoint in this router returns its full result set unpaginated.

The /stats Endpoint — A Real Bug#

@router.get("/{agent_id}/stats", response_model=schemas.AgentStatsResponse)
def get_agent_stats(
    agent_id: int,
    force: bool = Query(False, description="Force delete even if agent has data"),
    db: Session = Depends(get_db),
):
    agent = db.query(models.Agent).filter(models.Agent.agent_id == agent_id).first()
    ...
    db.delete(agent)
    db.commit()
    return {"message": "Agent deleted successfully", "agent_id": agent_id, ...}

Three independent problems stack up here:

1. Wrong attribute. models.Agent has no agent_id column — it has id. models.Agent.agent_id raises an AttributeError at query-build time. This endpoint cannot currently succeed on any request.
2. A GET that deletes. Even setting the attribute bug aside, an HTTP GET request performing db.delete(agent) violates the basic REST convention that GET requests must be safe (no side effects) — a crawler, browser prefetcher, or monitoring probe hitting this URL would delete data.
3. Response shape mismatch. The endpoint declares response_model=schemas.AgentStatsResponse (fields: agent_id, agent_name, total_messages, total_runs, total_tasks, total_tool_calls, has_data, safe_to_delete) but the return statement produces a completely different dict (message, agent_id, deleted_messages, deleted_runs, deleted_tasks, deleted_tool_calls) — FastAPI's response_model validation would reject this at serialization time even if the query bug were fixed.

This endpoint's name (/stats) and behavior (delete) don't match either — it reads like an in-progress rename from a delete endpoint to a stats endpoint that was never finished. It's grouped, in spirit, with the standalone practice scripts backend/app/delete.py, exercise_delete.py, practice_solution.py, and solution_delete.py that CLAUDE.md already flags as scratch/practice code — this one just never got cleaned out of the real router.

Common Mistakes#

• Trusting /agents/{id}/stats at all — it cannot run successfully as written; see above.
• Assuming PUT /tasks/{id}/status verifies the task belongs to the agent in the URL — it doesn't check agent ownership at all (there isn't even an agent ID in that URL).
• Assuming the dashboard endpoint is cheap — it loads full row objects for every task, tool call, and workflow run just to count and average them.

Exercises#

1. Fix the /stats endpoint: correct the attribute (models.Agent.id), decide whether it should actually delete data or just report statistics (the name says one thing, the body says another), and make the return value match schemas.AgentStatsResponse.
2. Rewrite get_agent_dashboard to compute total_tasks, open_tasks, in_progress_tasks, and completed_tasks with a single grouped SQL query instead of loading every Task row into Python.
3. Add an ownership check to PUT /tasks/{task_id}/status so a task can only be updated through its own agent's URL.

Key Takeaways#

🎯 agents.py is 937 lines because it hosts CRUD, chat, email analysis, task management, tool-call logging, the entire multi-agent workflow, and a dashboard — six responsibilities in one file. 🎯 The "fetch agent or 404" pattern repeats in nearly every endpoint and is a textbook missed dependency (Lesson 10). 🎯 /agents/{id}/stats is broken in three independent ways — a genuine finding from reading the source, not a hypothetical. 🎯 The dashboard endpoint aggregates in Python, not SQL — fine at small scale, a real cost at large scale.

Next Steps#

👉 Lesson 09b: Documents Router Deep Dive →

Module 4 · API & Business Logic