Coverage for src/data_api/mcp/tools/courses.py: 100%

1"""Course search MCP tool.""" 

2 

3import re 

4from typing import Annotated, Optional 

5 

6from fastmcp.exceptions import ToolError 

7from pydantic import Field, validate_call 

8 

9from data_api.domain.courses import models as courses_models 

10from data_api.domain.courses.services import UnknownSemester 

11from data_api.mcp.server import mcp 

12from data_api.mcp.tools.result_types import CoursesResult 

13from data_api.models.courses import SemesterCode 

14from data_api.runtime import get_runtime 

15 

16CourseSearchLimit = Annotated[int, Field(ge=1, le=100)] 

17 

18 

19@validate_call 

20async def _search_courses( 

21 keyword: Optional[str] = None, 

22 teacher: Optional[str] = None, 

23 course_id: Optional[str] = None, 

24 limit: CourseSearchLimit = 20, 

25 semester: SemesterCode | None = None, 

26) -> CoursesResult: 

27 """ 

28 Search for courses at NTHU. 

29 

30 Args: 

31 keyword: Literal, case-sensitive substring in Chinese or English titles. 

32 teacher: Literal, case-sensitive substring in teacher names. 

33 course_id: Literal, case-sensitive substring in course IDs. 

34 limit: Maximum number of results to return (1-100, default 20). 

35 semester: Published five-digit semester code; omitted means the upstream latest feed. 

36 

37 All supplied filters are combined with AND. 

38 

39 Returns: 

40 Dictionary with matching courses. 

41 """ 

42 service = get_runtime().courses 

43 try: 

44 snapshot = await service.get_snapshot(semester) 

45 except UnknownSemester as exc: 

46 raise ToolError("Semester is not published.") from exc 

47 condition = courses_models.Conditions(list_build_target=[]) 

48 

49 if keyword: 

50 condition &= courses_models.Conditions( 

51 "chinese_title", re.escape(keyword), regex_match=True 

52 ) | courses_models.Conditions("english_title", re.escape(keyword), regex_match=True) 

53 

54 if teacher: 

55 teacher_condition = courses_models.Conditions( 

56 "lecturers.label", re.escape(teacher), regex_match=True 

57 ) 

58 for field in ("lecturers.name_zh", "lecturers.name_en"): 

59 teacher_condition |= courses_models.Conditions( 

60 field, re.escape(teacher), regex_match=True 

61 ) 

62 condition &= teacher_condition 

63 if course_id: 

64 condition &= courses_models.Conditions("id", re.escape(course_id), regex_match=True) 

65 

66 courses = service.query(snapshot.data, condition)[:limit] 

67 

68 # Format response 

69 return { 

70 "count": len(courses), 

71 "courses": [course.model_dump(mode="json") for course in courses], 

72 } 

73 

74 

75@mcp.tool( 

76 name="search_courses", 

77 title="搜尋課程列表", 

78 description="Search for courses at NTHU by title, teacher, or course ID. " 

79 "Inputs are literal, case-sensitive substrings, not regular expressions. " 

80 "All supplied filters must match. Limit must be 1-100 (default 20). " 

81 "Semester is an optional published five-digit code; omitted means latest. " 

82 "Returns normalized course fields, including lecturers, meetings and nullable metadata.", 

83) 

84async def search_courses( 

85 keyword: Optional[str] = None, 

86 teacher: Optional[str] = None, 

87 course_id: Optional[str] = None, 

88 limit: CourseSearchLimit = 20, 

89 semester: SemesterCode | None = None, 

90) -> dict: 

91 """Search for courses at NTHU.""" 

92 return await _search_courses(keyword, teacher, course_id, limit, semester)