Component Implementation: Tools

All five tools live in server/components/tools.py. The file follows a single pattern throughout: a top-level register_tools function receives the mcp instance and uses it as a decorator factory to define and register each tool in one step. Let's walk through the file from top to bottom.

Imports and Registration Shell

Here is the code that starts the file, with imports and the register_tools function:

# components/tools.py
import os
from typing import Annotated
from typing import Literal

import httpx
from pydantic import Field
from sqlalchemy import desc
from sqlalchemy import func
from sqlalchemy.orm import Session

from fastmcp import FastMCP
from fastmcp.dependencies import CurrentContext
from fastmcp.dependencies import Depends
from fastmcp.exceptions import ToolError
from fastmcp.server.context import Context

from database import Movie
from database import ViewSummary
from database import get_db_session


def register_tools(mcp: FastMCP):
    ...

The imports are split into three layers.

The standard-library and third-party block brings in httpx for HTTP calls to the OMDB API.
SQLAlchemy helpers for building queries
Pydantic's Field for attaching validation metadata to parameters
The FastMCP block imports the Depends and CurrentContext dependency injectors, ToolError for raising user-visible errors, and Context for type-annotating the context parameter.
The local block imports the ORM models and the get_db_session factory from database.py.

All five tools are defined as nested functions inside register_tools. This pattern lets main.py call register_tools(mcp) once and have all tools registered without any circular import issues — mcp is just passed in rather than imported at module level. This is personal choice to avoid having to import the mcp instance directly in the tools.py file, which can lead to circular imports.

Tool 1: `search_movies` — Title Search with Elicitation

This is the tool an LLM calls when it needs to find a movie by name but doesn't yet have its database ID.

    @mcp.tool(timeout=60.0)
    async def search_movies(
        title: str,
        db: Session = Depends(get_db_session),
        ctx: Context = CurrentContext(),
    ) -> dict:
        """Find a movie by title when you don't already have its ID.

        ONLY use this when:
        - User asks to find a specific movie by name
        - You want to get the movie ID for a title you don't have
        - You want to show the user multiple matches and ask them to choose

        DO NOT use this if you have movie IDs from get_top_movies - use those IDs directly.

        Args:
          title: Partial or full movie title to search for.
        """
        # Step 1: Try exact match first (case-insensitive)
        results = db.query(Movie).filter(Movie.title.ilike(title)).limit(5).all()

        # Step 2: If no exact match, try partial match
        if not results:
            results = (
                db.query(Movie).filter(Movie.title.ilike(f"%{title}%")).limit(5).all()
            )

        if not results:
            raise ToolError(f"No movies found matching '{title}'.")

        # Exactly one match - confirm with user via elicitation
        if len(results) == 1:
            movie = results[0]
            year = movie.release_date.year if movie.release_date else "Unknown year"

            # Ask user to confirm this is the right movie
            # response_type=bool means the user answers yes/no
            confirm = await ctx.elicit(
                message=f"I found '{movie.title}' ({year}). Is this the movie you were looking for?\n"
                "Type 'yes' to confirm, 'no' to try a different search, or press Enter to cancel.",
                response_type=bool,
            )

            # confirm.action is "accept", "decline", or "cancel"
            # confirm.data is True (yes) or False (no) when action is "accept"
            # If user declined (Enter), cancelled, or said "no" → stop
            if confirm.action != "accept" or not confirm.data:
                raise ToolError("User cancelled the search. Do not retry.")
        # Multiple matches - ask user to choose
        else:
            options = []
            for i, movie in enumerate(results):
                year = movie.release_date.year if movie.release_date else "Unknown year"
                options.append(f"{i + 1}. {movie.title} ({year}) [ID: {movie.id}]")
            options_text = "\n".join(options)

            elicit_result = await ctx.elicit(
                message=f"While searching for a movie matching '{title}', "
                "I found multiple matches. "
                "Please select the correct one by entering the number:\n"
                f"{options_text}"
                "Press Enter without typing a number to cancel.",
                response_type=int,
            )

            if elicit_result.action == "accept":
                choice = elicit_result.data
                if 1 <= choice <= len(results):
                    movie = results[choice - 1]
                else:
                    raise ToolError(f"Invalid choice. Enter 1-{len(results)}.")
            else:
                raise ToolError("Selection cancelled.")

        result = {
            "id": movie.id,
            "title": movie.title,
            "release_date": movie.release_date.isoformat()
            if movie.release_date
            else None,
            "runtime": movie.runtime,
        }
        return result

The search runs in two passes:

The first pass tries an exact case-insensitive match using SQL's ILIKE.
If nothing comes back, the second pass wraps the title in % wildcards for a partial-match scan.

This ordering means a precise title like "Squid Game" won't accidentally return unrelated sequels before the main entry.

Elicitation is used in both branches. When there is exactly one result, the tool still asks the user to confirm it is the right movie before continuing — this avoids silently returning a wrong match when the search term is ambiguous. The confirmation uses response_type=bool, so the client presents a yes/no question. If the user declines or cancels, a ToolError is raised and the LLM sees a clear "cancelled" message.

When there are several matches, the user is presented with a numbered list and asked to pick one. The response_type=int variant collects the choice as a number. The server continues from the if elicit_result.action == "accept" branch, validates the number is in range, and selects the corresponding movie. If the user cancels or provides an out-of-range number, a ToolError is raised, which FastMCP converts into a clean error response the LLM can read.

Tool 2: `get_top_movies` — Ranked Results with Progress Reporting

    @mcp.tool(timeout=20.0)
    async def get_top_movies(
        metric: Literal["hours_viewed", "views"] = "hours_viewed",
        n: Annotated[
            int, Field(description="Number of top movies to return", ge=1)
        ] = 10,
        db: Session = Depends(get_db_session),
        ctx: Context = CurrentContext(),
    ) -> list:
        """Get top N movies ranked by total views or hours watched."""
        await ctx.report_progress(0, n, message=f"Fetching top {n} by {metric}...")

        metric_col = (
            func.sum(ViewSummary.hours_viewed)
            if metric == "hours_viewed"
            else func.sum(ViewSummary.views)
        )

        all_results = (
            db.query(
                Movie.id, Movie.title, Movie.release_date, metric_col.label("total")
            )
            .join(ViewSummary, ViewSummary.movie_id == Movie.id)
            .group_by(Movie.id, Movie.title, Movie.release_date)
            .order_by(