Coverage for src/data_api/mcp/server.py: 96%

1""" 

2NTHU Data API MCP Server. 

3 

4Curated MCP tools designed for LLM agents, following agent-first design principles: 

51. Minimal tool count - combine related functionality 

62. LLM-friendly names and descriptions in natural language 

73. Reasonable defaults to hide complexity 

84. Focus on agent stories rather than API structure 

9 

10These tools are designed to answer common questions about NTHU campus life. 

11""" 

12 

13from typing import TYPE_CHECKING 

14 

15from fastmcp import FastMCP 

16from fastmcp.exceptions import ToolError 

17from fastmcp.server.middleware import Middleware, MiddlewareContext 

18 

19from data_api import __version__ 

20from data_api.core.exceptions import DataNotAvailableException 

21 

22if TYPE_CHECKING: 

23 from data_api.runtime import Runtime 

24 

25 

26class DatasetErrors(Middleware): 

27 async def on_call_tool(self, context: MiddlewareContext, call_next): 

28 try: 

29 return await call_next(context) 

30 except DataNotAvailableException as exc: 

31 raise ToolError("Dataset temporarily unavailable. Please try again later.") from exc 

32 except ToolError as exc: 

33 # FastMCP wraps execution failures before middleware sees them. 

34 if isinstance(exc.__cause__, DataNotAvailableException): 

35 raise ToolError("Dataset temporarily unavailable. Please try again later.") from exc 

36 raise 

37 

38 

39# Create curated MCP server 

40mcp = FastMCP( 

41 name="NTHU Campus Assistant", 

42 version=__version__, 

43 mask_error_details=True, 

44 instructions="""You are an assistant for National Tsing Hua University (NTHU) in Taiwan. 

45You can help with: 

46- Finding campus locations and directions 

47- Looking up departments and staff contact information 

48- Checking bus schedules and next arrivals 

49- Searching for courses 

50- Getting campus announcements 

51- Finding restaurants and dining options 

52- Checking library information 

53- Getting newsletter updates 

54- Checking real-time electricity usage 

55 

56Always respond in the user's language (Traditional Chinese or English). 

57""", 

58) 

59mcp.add_middleware(DatasetErrors()) 

60 

61 

62def create_mcp(runtime: "Runtime") -> FastMCP: 

63 """Reuse tool definitions, not the HTTP session manager or runtime state.""" 

64 from data_api.runtime import bind_runtime 

65 

66 class RuntimeBinding(Middleware): 

67 async def on_call_tool(self, context: MiddlewareContext, call_next): 

68 with bind_runtime(runtime): 

69 return await call_next(context) 

70 

71 return FastMCP( 

72 name=mcp.name, 

73 version=mcp.version, 

74 instructions=mcp.instructions, 

75 providers=[mcp.local_provider], 

76 middleware=[RuntimeBinding(), DatasetErrors()], 

77 mask_error_details=True, 

78 ) 

79 

80 

81# Import tools after mcp is created to avoid circular imports 

82# Tools register themselves with the mcp instance when imported 

83from data_api.mcp.tools import ( # noqa: E402, F401 

84 _find_dining, 

85 _get_announcements, 

86 _get_bus_schedule, 

87 _get_energy_usage, 

88 _get_library_info, 

89 _get_newsletters, 

90 _search_courses, 

91 _search_directory, 

92 _search_locations, 

93 find_dining, 

94 get_announcements, 

95 get_bus_schedule, 

96 get_energy_usage, 

97 get_library_info, 

98 get_newsletters, 

99 search_courses, 

100 search_directory, 

101 search_locations, 

102) 

103 

104__all__ = [ 

105 "mcp", 

106 # Implementation functions (for testing) 

107 "_search_directory", 

108 "_search_locations", 

109 "_get_bus_schedule", 

110 "_search_courses", 

111 "_get_announcements", 

112 "_find_dining", 

113 "_get_library_info", 

114 "_get_newsletters", 

115 "_get_energy_usage", 

116 # MCP tool wrappers 

117 "search_directory", 

118 "search_locations", 

119 "get_bus_schedule", 

120 "search_courses", 

121 "get_announcements", 

122 "find_dining", 

123 "get_library_info", 

124 "get_newsletters", 

125 "get_energy_usage", 

126]