Coverage for sources/librovore/server.py: 61%
70 statements
« prev ^ index » next coverage.py v7.10.5, created at 2025-08-29 01:14 +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 functions as _functions
31from . import interfaces as _interfaces
32from . import results as _results
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 _produce_detect_function( auxdata: _state.Globals ):
90 async def detect(
91 location: LocationArgument,
92 genus: __.typx.Annotated[
93 _interfaces.ProcessorGenera,
94 _Field( description = "Processor genus (inventory or structure)" ),
95 ],
96 processor_name: __.typx.Annotated[
97 __.typx.Optional[ str ],
98 _Field( description = "Optional processor name." ),
99 ] = None,
100 ) -> dict[ str, __.typx.Any ]:
101 nomargs: __.NominativeArguments = { }
102 if processor_name is not None:
103 nomargs[ 'processor_name' ] = processor_name
104 result = await _functions.detect( auxdata, location, genus, **nomargs )
105 if isinstance( result, _results.ErrorResponse ):
106 return _results.serialize_for_json( result )
107 return dict( result.render_as_json( ) )
109 return detect
112def _produce_query_content_function( auxdata: _state.Globals ):
113 async def query_content( # noqa: PLR0913
114 location: LocationArgument,
115 term: TermArgument,
116 search_behaviors: __.typx.Annotated[
117 SearchBehaviorsMutable,
118 _Field( description = "Search behavior configuration" ),
119 ] = _search_behaviors_default,
120 filters: __.typx.Annotated[
121 FiltersMutable,
122 _Field( description = "Processor-specific filters" ),
123 ] = _filters_default,
124 include_snippets: IncludeSnippets = True,
125 results_max: ResultsMax = 10,
126 ) -> dict[ str, __.typx.Any ]:
127 immutable_search_behaviors = (
128 _to_immutable_search_behaviors( search_behaviors ) )
129 immutable_filters = _to_immutable_filters( filters )
130 result = await _functions.query_content(
131 auxdata, location, term,
132 search_behaviors = immutable_search_behaviors,
133 filters = immutable_filters,
134 include_snippets = include_snippets,
135 results_max = results_max )
136 return _results.serialize_for_json( result )
138 return query_content
141def _produce_query_inventory_function( auxdata: _state.Globals ):
142 async def query_inventory( # noqa: PLR0913
143 location: LocationArgument,
144 term: TermArgument,
145 search_behaviors: __.typx.Annotated[
146 SearchBehaviorsMutable,
147 _Field( description = "Search behavior configuration" ),
148 ] = _search_behaviors_default,
149 filters: __.typx.Annotated[
150 FiltersMutable,
151 _Field( description = "Processor-specific filters" ),
152 ] = _filters_default,
153 details: __.typx.Annotated[
154 _interfaces.InventoryQueryDetails,
155 _Field( description = "Detail level for inventory results" ),
156 ] = _interfaces.InventoryQueryDetails.Documentation,
157 results_max: ResultsMax = 5,
158 ) -> dict[ str, __.typx.Any ]:
159 immutable_search_behaviors = (
160 _to_immutable_search_behaviors( search_behaviors ) )
161 immutable_filters = _to_immutable_filters( filters )
162 result = await _functions.query_inventory(
163 auxdata, location, term,
164 search_behaviors = immutable_search_behaviors,
165 filters = immutable_filters,
166 details = details,
167 results_max = results_max )
168 return _results.serialize_for_json( result )
170 return query_inventory
175def _produce_survey_processors_function( auxdata: _state.Globals ):
176 async def survey_processors(
177 genus: __.typx.Annotated[
178 _interfaces.ProcessorGenera,
179 _Field( description = "Processor genus (inventory or structure)" ),
180 ],
181 name: __.typx.Annotated[
182 __.typx.Optional[ str ],
183 _Field( description = "Optional processor name to filter." )
184 ] = None,
185 ) -> dict[ str, __.typx.Any ]:
186 result = await _functions.survey_processors( auxdata, genus, name )
187 return dict( result.render_as_json( ) )
189 return survey_processors
192def _register_server_functions(
193 auxdata: _state.Globals, mcp: _FastMCP, /, *, extra_functions: bool = False
194) -> None:
195 ''' Registers MCP server tools with closures for auxdata access. '''
196 _scribe.debug( "Registering tools." )
197 mcp.tool( )( _produce_query_inventory_function( auxdata ) )
198 mcp.tool( )( _produce_query_content_function( auxdata ) )
199 if extra_functions: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 mcp.tool( )( _produce_detect_function( auxdata ) )
201 mcp.tool( )( _produce_survey_processors_function( auxdata ) )
202 _scribe.debug( "All tools registered successfully." )
205def _to_immutable_filters(
206 mutable_filters: FiltersMutable
207) -> __.immut.Dictionary[ str, __.typx.Any ]:
208 ''' Converts mutable filters dict to immutable dictionary. '''
209 return __.immut.Dictionary[ str, __.typx.Any ]( mutable_filters )
212def _to_immutable_search_behaviors(
213 mutable_behaviors: SearchBehaviorsMutable
214) -> _interfaces.SearchBehaviors:
215 ''' Converts mutable search behaviors to immutable. '''
216 field_values = {
217 field.name: getattr( mutable_behaviors, field.name )
218 for field in __.dcls.fields( mutable_behaviors )
219 if not field.name.startswith( '_' ) }
220 return _interfaces.SearchBehaviors( **field_values )