Coverage for sources/librovore/server.py: 49%
90 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-17 23:43 +0000
1# vim: set filetype=python fileencoding=utf-8:
2# -*- coding: utf-8 -*-
4#============================================================================#
5# #
6# Licensed under the Apache License, Version 2.0 (the "License"); #
7# you may not use this file except in compliance with the License. #
8# You may obtain a copy of the License at #
9# #
10# http://www.apache.org/licenses/LICENSE-2.0 #
11# #
12# Unless required by applicable law or agreed to in writing, software #
13# distributed under the License is distributed on an "AS IS" BASIS, #
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
15# See the License for the specific language governing permissions and #
16# limitations under the License. #
17# #
18#============================================================================#
21''' MCP server implementation. '''
24from mcp.server.fastmcp import FastMCP as _FastMCP
26# FastMCP uses Pydantic to generate JSON schemas from function signatures.
27from pydantic import Field as _Field
29from . import __
30from . import exceptions as _exceptions
31from . import functions as _functions
32from . import interfaces as _interfaces
33from . import state as _state
36@__.dcls.dataclass( kw_only = True, slots = True )
37class SearchBehaviorsMutable:
38 ''' Mutable version of SearchBehaviors for FastMCP/Pydantic compatibility.
40 Note: Fields are manually duplicated from SearchBehaviors to avoid
41 immutable dataclass internals leaking into JSON schema generation.
42 '''
44 match_mode: _interfaces.MatchMode = _interfaces.MatchMode.Fuzzy
45 fuzzy_threshold: int = 50
48FiltersMutable: __.typx.TypeAlias = dict[ str, __.typx.Any ]
49GroupByArgument: __.typx.TypeAlias = __.typx.Annotated[
50 __.typx.Optional[ str ],
51 _Field( description = __.access_doctab( 'group by argument' ) ),
52]
53IncludeSnippets: __.typx.TypeAlias = __.typx.Annotated[
54 bool,
55 _Field( description = __.access_doctab( 'include snippets argument' ) ),
56]
57TermArgument: __.typx.TypeAlias = __.typx.Annotated[
58 str, _Field( description = __.access_doctab( 'term argument' ) ) ]
59ResultsMax: __.typx.TypeAlias = __.typx.Annotated[
60 int, _Field( description = __.access_doctab( 'results max argument' ) ) ]
61LocationArgument: __.typx.TypeAlias = __.typx.Annotated[
62 str, _Field( description = __.access_doctab( 'location argument' ) ) ]
65_filters_default = FiltersMutable( )
66_search_behaviors_default = SearchBehaviorsMutable( )
68_scribe = __.acquire_scribe( __name__ )
72async def serve(
73 auxdata: _state.Globals, /, *,
74 port: int = 0,
75 transport: str = 'stdio',
76 extra_functions: bool = False,
77) -> None:
78 ''' Runs MCP server. '''
79 _scribe.debug( "Initializing FastMCP server." )
80 mcp = _FastMCP( 'Sphinx MCP Server', port = port )
81 _register_server_functions(
82 auxdata, mcp, extra_functions = extra_functions )
83 match transport:
84 case 'sse': await mcp.run_sse_async( mount_path = None )
85 case 'stdio': await mcp.run_stdio_async( )
86 case _: raise ValueError
89def _exception_to_error_response( exc: Exception ) -> dict[ str, str ]: # noqa: PLR0911
90 ''' Maps exceptions to structured error responses for MCP tools. '''
91 match exc:
92 case _exceptions.ProcessorInavailability( ):
93 return {
94 'error_type': 'ProcessorInavailability',
95 'message':
96 'No processor found to handle this documentation source',
97 'details': f"Source: {exc.source}",
98 'suggestion': 'Verify the URL is a Sphinx documentation site'
99 }
100 case _exceptions.InventoryInaccessibility( ):
101 cause = exc.__cause__ or 'Unknown'
102 return {
103 'error_type': 'InventoryInaccessibility',
104 'message': 'Cannot access documentation inventory',
105 'details': f"Source: {exc.source}, Cause: {cause}",
106 'suggestion': 'Check URL accessibility and network connection'
107 }
108 case _exceptions.DocumentationContentAbsence( ):
109 return {
110 'error_type': 'DocumentationContentAbsence',
111 'message': 'Documentation page structure not recognized',
112 'details': f"URL: {exc.url}",
113 'suggestion': 'This may be an unsupported Sphinx theme'
114 }
115 case _exceptions.DocumentationObjectAbsence( ):
116 return {
117 'error_type': 'ObjectNotFoundError',
118 'message': 'Requested object not found in documentation page',
119 'details': f"Object: {exc.object_id}, URL: {exc.url}",
120 'suggestion': 'Verify the object name and try a broader search'
121 }
122 case _exceptions.InventoryInvalidity( ):
123 cause = exc.__cause__ or 'Unknown'
124 return {
125 'error_type': 'InventoryInvalidity',
126 'message': 'Documentation inventory has invalid format',
127 'details': f"Source: {exc.source}, Cause: {cause}",
128 'suggestion': 'The documentation site may be corrupted'
129 }
130 case _exceptions.DocumentationInaccessibility( ):
131 cause = exc.__cause__ or 'Unknown'
132 return {
133 'error_type': 'DocumentationInaccessibility',
134 'message': 'Documentation file or resource is inaccessible',
135 'details': f"URL: {exc.url}, Cause: {cause}",
136 'suggestion': 'Check URL accessibility and network connection'
137 }
138 case _:
139 return {
140 'error_type': 'UnknownError',
141 'message': 'An unexpected error occurred',
142 'details': f"Exception: {type( exc ).__name__}: {exc}",
143 'suggestion': 'Please report this issue if it persists'
144 }
147def _produce_detect_function( auxdata: _state.Globals ):
148 async def detect(
149 location: LocationArgument,
150 genus: __.typx.Annotated[
151 _interfaces.ProcessorGenera,
152 _Field( description = "Processor genus (inventory or structure)" ),
153 ],
154 processor_name: __.typx.Annotated[
155 __.typx.Optional[ str ],
156 _Field( description = "Optional processor name." ),
157 ] = None,
158 ) -> dict[ str, __.typx.Any ]:
159 nomargs: __.NominativeArguments = { }
160 if processor_name is not None:
161 nomargs[ 'processor_name' ] = processor_name
162 return await _functions.detect( auxdata, location, genus, **nomargs )
164 return detect
167def _produce_query_content_function( auxdata: _state.Globals ):
168 async def query_content( # noqa: PLR0913
169 location: LocationArgument,
170 term: TermArgument,
171 search_behaviors: __.typx.Annotated[
172 SearchBehaviorsMutable,
173 _Field( description = "Search behavior configuration" ),
174 ] = _search_behaviors_default,
175 filters: __.typx.Annotated[
176 FiltersMutable,
177 _Field( description = "Processor-specific filters" ),
178 ] = _filters_default,
179 include_snippets: IncludeSnippets = True,
180 results_max: ResultsMax = 10,
181 ) -> dict[ str, __.typx.Any ]:
182 immutable_search_behaviors = (
183 _to_immutable_search_behaviors( search_behaviors ) )
184 immutable_filters = _to_immutable_filters( filters )
185 return await _functions.query_content(
186 auxdata, location, term,
187 search_behaviors = immutable_search_behaviors,
188 filters = immutable_filters,
189 include_snippets = include_snippets,
190 results_max = results_max )
192 return query_content
195def _produce_query_inventory_function( auxdata: _state.Globals ):
196 async def query_inventory( # noqa: PLR0913
197 location: LocationArgument,
198 term: TermArgument,
199 search_behaviors: __.typx.Annotated[
200 SearchBehaviorsMutable,
201 _Field( description = "Search behavior configuration" ),
202 ] = _search_behaviors_default,
203 filters: __.typx.Annotated[
204 FiltersMutable,
205 _Field( description = "Processor-specific filters" ),
206 ] = _filters_default,
207 details: __.typx.Annotated[
208 _interfaces.InventoryQueryDetails,
209 _Field( description = "Detail level for inventory results" ),
210 ] = _interfaces.InventoryQueryDetails.Documentation,
211 results_max: ResultsMax = 5,
212 ) -> dict[ str, __.typx.Any ]:
213 immutable_search_behaviors = (
214 _to_immutable_search_behaviors( search_behaviors ) )
215 immutable_filters = _to_immutable_filters( filters )
216 return await _functions.query_inventory(
217 auxdata, location, term,
218 search_behaviors = immutable_search_behaviors,
219 filters = immutable_filters,
220 details = details,
221 results_max = results_max )
223 return query_inventory
226def _produce_summarize_inventory_function( auxdata: _state.Globals ):
227 async def summarize_inventory(
228 location: LocationArgument,
229 search_behaviors: __.typx.Annotated[
230 SearchBehaviorsMutable,
231 _Field( description = "Search behavior configuration." ),
232 ] = _search_behaviors_default,
233 filters: __.typx.Annotated[
234 FiltersMutable,
235 _Field( description = "Processor-specific filters." ),
236 ] = _filters_default,
237 group_by: GroupByArgument = None,
238 term: TermArgument = '',
239 ) -> dict[ str, __.typx.Any ]:
240 immutable_search_behaviors = (
241 _to_immutable_search_behaviors( search_behaviors ) )
242 immutable_filters = _to_immutable_filters( filters )
243 return await _functions.summarize_inventory(
244 auxdata, location, term,
245 search_behaviors = immutable_search_behaviors,
246 filters = immutable_filters,
247 group_by = group_by )
249 return summarize_inventory
252def _produce_survey_processors_function( auxdata: _state.Globals ):
253 async def survey_processors(
254 genus: __.typx.Annotated[
255 _interfaces.ProcessorGenera,
256 _Field( description = "Processor genus (inventory or structure)" ),
257 ],
258 name: __.typx.Annotated[
259 __.typx.Optional[ str ],
260 _Field( description = "Optional processor name to filter." )
261 ] = None,
262 ) -> dict[ str, __.typx.Any ]:
263 return await _functions.survey_processors( auxdata, genus, name )
265 return survey_processors
268def _register_server_functions(
269 auxdata: _state.Globals, mcp: _FastMCP, /, *, extra_functions: bool = False
270) -> None:
271 ''' Registers MCP server tools with closures for auxdata access. '''
272 _scribe.debug( "Registering tools." )
273 mcp.tool( )( _produce_query_inventory_function( auxdata ) )
274 mcp.tool( )( _produce_query_content_function( auxdata ) )
275 mcp.tool( )( _produce_summarize_inventory_function( auxdata ) )
276 if extra_functions: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 mcp.tool( )( _produce_detect_function( auxdata ) )
278 mcp.tool( )( _produce_survey_processors_function( auxdata ) )
279 _scribe.debug( "All tools registered successfully." )
282def _to_immutable_filters(
283 mutable_filters: FiltersMutable
284) -> __.immut.Dictionary[ str, __.typx.Any ]:
285 ''' Converts mutable filters dict to immutable dictionary. '''
286 return __.immut.Dictionary[ str, __.typx.Any ]( mutable_filters )
289def _to_immutable_search_behaviors(
290 mutable_behaviors: SearchBehaviorsMutable
291) -> _interfaces.SearchBehaviors:
292 ''' Converts mutable search behaviors to immutable. '''
293 field_values = {
294 field.name: getattr( mutable_behaviors, field.name )
295 for field in __.dcls.fields( mutable_behaviors )
296 if not field.name.startswith( '_' ) }
297 return _interfaces.SearchBehaviors( **field_values )