Coverage for src/openapi_navigator/server.py: 81%
137 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-15 01:16 +0000
1"""OpenAPI Navigator - Tools for navigating OpenAPI specifications."""
3import logging
4import time
5import json as json_module
6from typing import Dict, List, Optional, Any
7import requests
8from fastmcp import FastMCP
9from openapi_navigator.spec_manager import SpecManager
11logger = logging.getLogger(__name__)
13# Create a global spec manager instance that all tools share
14_spec_manager = SpecManager()
16# Create the main MCP server instance for CLI tools to find
17mcp = FastMCP("openapi-navigator")
20# Register all tools
21@mcp.tool
22def load_spec(file_path: str, spec_id: Optional[str] = None) -> str:
23 """
24 Load an OpenAPI specification from a local file.
26 Args:
27 file_path: Absolute path to the OpenAPI spec file (YAML or JSON)
28 spec_id: Optional custom identifier for the spec. If not provided, will use 'file:{file_path}'
30 Returns:
31 The spec ID that was assigned to the loaded specification
33 Note:
34 File path must be absolute for security reasons.
35 """
36 try:
37 return _spec_manager.load_spec_from_file(file_path, spec_id)
38 except Exception as e:
39 logger.error(f"Failed to load spec from file: {e}")
40 raise
43@mcp.tool
44def load_spec_from_url(url: str, spec_id: Optional[str] = None) -> str:
45 """
46 Load an OpenAPI specification from a URL.
48 Args:
49 url: URL to the OpenAPI spec (YAML or JSON)
50 spec_id: Optional custom identifier for the spec. If not provided, will use 'url:{url}'
52 Returns:
53 The spec ID that was assigned to the loaded specification
54 """
55 try:
56 return _spec_manager.load_spec_from_url(url, spec_id)
57 except Exception as e:
58 logger.error(f"Failed to load spec from URL: {e}")
59 raise
62@mcp.tool
63def unload_spec(spec_id: str) -> str:
64 """
65 Unload an OpenAPI specification from memory.
67 Args:
68 spec_id: ID of the loaded spec to unload
70 Returns:
71 Confirmation message
72 """
73 success = _spec_manager.unload_spec(spec_id)
74 if success:
75 return f"Successfully unloaded spec: {spec_id}"
76 else:
77 return f"Spec not found or already unloaded: {spec_id}"
80@mcp.tool
81def list_loaded_specs() -> List[str]:
82 """
83 List all currently loaded OpenAPI specifications.
85 Returns:
86 List of spec IDs that are currently loaded
87 """
88 return _spec_manager.list_loaded_specs()
91@mcp.tool
92def get_endpoint(
93 spec_id: str, path: str, method: str, summary_only: bool = False
94) -> Optional[Dict[str, Any]]:
95 """
96 Get the operation definition for a specific endpoint.
98 Args:
99 spec_id: ID of the loaded spec to query
100 path: API path (e.g., '/users/{id}')
101 method: HTTP method (e.g., 'GET', 'POST')
102 summary_only: If True, return only essential fields to reduce token usage (default: False)
104 Returns:
105 The operation object from the OpenAPI spec (full or summary), or None if not found
106 """
107 spec = _spec_manager.get_spec(spec_id)
108 if not spec:
109 raise ValueError(f"No spec found with ID: {spec_id}")
111 endpoint = spec.get_endpoint(path, method.upper())
112 if not endpoint:
113 return None
115 if summary_only:
116 # Return only essential information to reduce token usage
117 return {
118 "summary": endpoint.get("summary", ""),
119 "description": endpoint.get("description", ""),
120 "operationId": endpoint.get("operationId", ""),
121 "tags": endpoint.get("tags", []),
122 "parameters": [
123 {
124 "name": p.get("name", ""),
125 "in": p.get("in", ""),
126 "required": p.get("required", False),
127 "type": p.get("schema", {}).get("type", p.get("type", "")),
128 "description": p.get("description", ""),
129 }
130 for p in endpoint.get("parameters", [])
131 ],
132 "responses": {
133 code: {
134 "description": resp.get("description", ""),
135 "content_types": (
136 list(resp.get("content", {}).keys())
137 if "content" in resp
138 else []
139 ),
140 }
141 for code, resp in endpoint.get("responses", {}).items()
142 },
143 "requestBody": (
144 {
145 "required": endpoint.get("requestBody", {}).get("required", False),
146 "content_types": list(
147 endpoint.get("requestBody", {}).get("content", {}).keys()
148 ),
149 }
150 if "requestBody" in endpoint
151 else None
152 ),
153 }
155 return endpoint
158@mcp.tool
159def search_endpoints(
160 spec_id: str, query: str, limit: int = 50, offset: int = 0
161) -> Dict[str, Any]:
162 """
163 Search endpoints using fuzzy matching across paths, summaries, and operation IDs with pagination.
165 To get a full list of all endpoints, use an empty string "" or a very short query like "a" as the search term.
166 The search will return all endpoints with a relevance score of 100 when the query is very short.
168 Args:
169 spec_id: ID of the loaded spec to query
170 query: Search query string. Use "" or "a" to get all endpoints.
171 limit: Maximum number of results to return (default 50, max 200)
172 offset: Number of results to skip for pagination (default 0)
174 Returns:
175 Dictionary containing:
176 - endpoints: List of matching endpoints with relevance scores
177 - total: Total number of matches (before pagination)
178 - limit: Applied limit
179 - offset: Applied offset
180 - has_more: Whether there are more results available
181 """
182 spec = _spec_manager.get_spec(spec_id)
183 if not spec:
184 raise ValueError(f"No spec found with ID: {spec_id}")
186 return spec.search_endpoints(query, limit, offset)
189@mcp.tool
190def get_schema(spec_id: str, schema_name: str) -> Optional[Dict[str, Any]]:
191 """
192 Get a specific schema definition from a loaded OpenAPI specification.
194 Args:
195 spec_id: ID of the loaded spec to query
196 schema_name: Name of the schema to retrieve
198 Returns:
199 The raw schema object from the OpenAPI spec, or None if not found
200 """
201 spec = _spec_manager.get_spec(spec_id)
202 if not spec:
203 raise ValueError(f"No spec found with ID: {spec_id}")
205 return spec.get_schema(schema_name)
208@mcp.tool
209def search_schemas(
210 spec_id: str, query: str, limit: int = 50, offset: int = 0
211) -> Dict[str, Any]:
212 """
213 Search schema names using fuzzy matching with pagination.
215 To get a full list of all schemas, use an empty string "" or a very short query like "a" as the search term.
216 The search will return all schemas with a relevance score of 100 when the query is very short.
218 Args:
219 spec_id: ID of the loaded spec to query
220 query: Search query string. Use "" or "a" to get all schemas.
221 limit: Maximum number of results to return (default 50, max 200)
222 offset: Number of results to skip for pagination (default 0)
224 Returns:
225 Dictionary containing:
226 - schemas: List of matching schema names with relevance scores
227 - total: Total number of matches (before pagination)
228 - limit: Applied limit
229 - offset: Applied offset
230 - has_more: Whether there are more results available
231 """
232 spec = _spec_manager.get_spec(spec_id)
233 if not spec:
234 raise ValueError(f"No spec found with ID: {spec_id}")
236 return spec.search_schemas(query, limit, offset)
239@mcp.tool
240def get_spec_metadata(spec_id: str) -> Dict[str, Any]:
241 """
242 Get comprehensive metadata about a loaded OpenAPI specification.
244 This includes information about the spec version, title, description, base path,
245 servers, contact info, license, and counts of endpoints and schemas.
247 Args:
248 spec_id: ID of the loaded spec to query
250 Returns:
251 Dictionary containing spec metadata including base path and help text
252 """
253 spec = _spec_manager.get_spec(spec_id)
254 if not spec:
255 raise ValueError(f"No spec found with ID: {spec_id}")
257 return spec.get_spec_metadata()
260def _make_api_request_impl(
261 url: str,
262 method: str = "GET",
263 headers: Optional[Dict[str, str]] = None,
264 params: Optional[Dict[str, str]] = None,
265 data: Optional[str] = None,
266 timeout: int = 30,
267 spec_id: Optional[str] = None,
268) -> Dict[str, Any]:
269 """
270 Make a generic REST API request with full control over method, headers, parameters, and body.
272 This tool enables direct interaction with REST APIs, complementing the OpenAPI exploration tools
273 by allowing you to actually call the endpoints you've discovered.
275 Args:
276 url: The full URL to make the request to
277 method: HTTP method (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS). Defaults to GET.
278 headers: Optional dictionary of HTTP headers to include in the request
279 params: Optional dictionary of URL parameters to append to the URL
280 data: Optional request body data as a string (JSON, XML, form data, etc.)
281 timeout: Request timeout in seconds. Defaults to 30.
282 spec_id: Optional spec ID to use for default headers. If provided, headers from
283 set_spec_headers will be merged (request headers take precedence).
285 Returns:
286 Dictionary containing:
287 - status_code: HTTP status code
288 - headers: Response headers as a dictionary
289 - body: Raw response body as string
290 - json: Parsed JSON response (if response is valid JSON, otherwise None)
291 - url: Final URL after any redirects
292 - elapsed_ms: Request duration in milliseconds
293 - method: The HTTP method that was used
295 Raises:
296 ValueError: If the URL is invalid or method is not supported, or spec_id doesn't exist
297 ConnectionError: If the request fails due to network issues
298 TimeoutError: If the request times out
299 """
300 # Validate method
301 valid_methods = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
302 method = method.upper()
303 if method not in valid_methods:
304 raise ValueError(
305 f"Invalid HTTP method: {method}. Must be one of {valid_methods}"
306 )
308 # Validate URL
309 if not url or not isinstance(url, str):
310 raise ValueError("URL must be a non-empty string")
312 # Merge headers from spec if spec_id provided
313 merged_headers = {}
314 if spec_id:
315 spec = _spec_manager.get_spec(spec_id)
316 if not spec:
317 raise ValueError(f"No spec found with ID: {spec_id}")
318 # Start with spec's default headers
319 merged_headers = spec.get_headers()
321 # Override with explicit headers (request headers take precedence)
322 if headers:
323 merged_headers.update(headers)
325 try:
326 # Prepare the request
327 request_kwargs = {
328 "method": method,
329 "url": url,
330 "timeout": timeout,
331 }
333 if merged_headers: # Changed from `if headers:`
334 request_kwargs["headers"] = merged_headers
336 if params:
337 request_kwargs["params"] = params
339 if data:
340 request_kwargs["data"] = data
342 # Make the request and time it
343 start_time = time.time()
344 logger.info(f"Making {method} request to {url}")
346 response = requests.request(**request_kwargs)
348 end_time = time.time()
349 elapsed_ms = int((end_time - start_time) * 1000)
351 # Parse JSON response if possible
352 response_json = None
353 try:
354 if response.text.strip(): # Only try to parse if there's content
355 response_json = response.json()
356 except (json_module.JSONDecodeError, ValueError):
357 # Response is not JSON, that's fine
358 pass
360 # Build response dictionary
361 result = {
362 "status_code": response.status_code,
363 "headers": dict(response.headers),
364 "body": response.text,
365 "json": response_json,
366 "url": response.url,
367 "elapsed_ms": elapsed_ms,
368 "method": method,
369 }
371 logger.info(
372 f"Request completed: {method} {url} -> {response.status_code} ({elapsed_ms}ms)"
373 )
374 return result
376 except requests.exceptions.Timeout as e:
377 logger.error(f"Request timed out: {method} {url}")
378 raise TimeoutError(f"Request timed out after {timeout} seconds: {str(e)}")
380 except requests.exceptions.ConnectionError as e:
381 logger.error(f"Connection error: {method} {url} - {str(e)}")
382 raise ConnectionError(f"Failed to connect to {url}: {str(e)}")
384 except requests.exceptions.RequestException as e:
385 logger.error(f"Request failed: {method} {url} - {str(e)}")
386 raise Exception(f"Request failed: {str(e)}")
388 except Exception as e:
389 logger.error(f"Unexpected error during request: {method} {url} - {str(e)}")
390 raise Exception(f"Unexpected error: {str(e)}")
393@mcp.tool
394def make_api_request(
395 url: str,
396 method: str = "GET",
397 headers: dict = None,
398 params: dict = None,
399 data: str = None,
400 timeout: int = 30,
401 spec_id: str = None,
402) -> dict:
403 """
404 Make a generic REST API request with full control over method, headers, parameters, and body.
406 This tool enables direct interaction with REST APIs, complementing the OpenAPI exploration tools
407 by allowing you to actually call the endpoints you've discovered.
409 Args:
410 url: The full URL to make the request to
411 method: HTTP method (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS). Defaults to GET.
412 headers: Dictionary of HTTP headers to include in the request (optional)
413 params: Dictionary of URL parameters to append to the URL (optional)
414 data: Request body data as a string (JSON, XML, form data, etc.) (optional)
415 timeout: Request timeout in seconds. Defaults to 30.
416 spec_id: Optional spec ID to automatically apply headers from set_spec_headers.
417 Request headers take precedence over spec headers. (optional)
419 Returns:
420 Dictionary containing:
421 - status_code: HTTP status code
422 - headers: Response headers as a dictionary
423 - body: Raw response body as string
424 - json: Parsed JSON response (if response is valid JSON, otherwise None)
425 - url: Final URL after any redirects
426 - elapsed_ms: Request duration in milliseconds
427 - method: The HTTP method that was used
429 Raises:
430 ValueError: If the URL is invalid or method is not supported, or spec_id doesn't exist
431 ConnectionError: If the request fails due to network issues
432 TimeoutError: If the request times out
434 Example with spec_id:
435 # Load and configure a spec with auth
436 load_spec_from_url("https://api.example.com/openapi.json", "my-api")
437 set_spec_headers("my-api", {"Authorization": "Bearer token123"})
439 # Make requests without repeating auth headers
440 make_api_request("https://api.example.com/users", spec_id="my-api")
442 # Override specific headers while keeping others from spec
443 make_api_request(
444 "https://api.example.com/admin",
445 spec_id="my-api",
446 headers={"X-Admin-Token": "admin456"}
447 )
448 """
449 return _make_api_request_impl(url, method, headers, params, data, timeout, spec_id)
452def _set_spec_headers_impl(spec_id: str, headers: dict = None) -> str:
453 """
454 Internal implementation for setting default headers on a spec.
456 Args:
457 spec_id: ID of the loaded spec to set headers for
458 headers: Dictionary of HTTP headers to use by default
460 Returns:
461 Confirmation message
463 Raises:
464 ValueError: If the spec_id doesn't exist
465 """
466 spec = _spec_manager.get_spec(spec_id)
467 if not spec:
468 raise ValueError(f"No spec found with ID: {spec_id}")
470 if headers is None:
471 headers = {}
473 spec.set_headers(headers)
475 header_count = len(headers)
476 if header_count == 0:
477 return f"Successfully cleared headers for spec: {spec_id}"
478 else:
479 return f"Successfully set {header_count} default header(s) for spec: {spec_id}"
482@mcp.tool
483def set_spec_headers(spec_id: str, headers: dict = None) -> str:
484 """
485 Set default headers for API requests to a loaded OpenAPI spec.
487 This allows you to "mount" authentication or other headers to a spec once,
488 rather than passing them on every make_api_request call. When making requests
489 with make_api_request, you can reference the spec_id to automatically apply
490 these headers.
492 Args:
493 spec_id: ID of the loaded spec to set headers for
494 headers: Dictionary of HTTP headers to use by default (optional, defaults to empty)
496 Returns:
497 Confirmation message
499 Raises:
500 ValueError: If the spec_id doesn't exist
502 Example:
503 # Load a spec
504 load_spec_from_url("https://api.example.com/openapi.json", "my-api")
506 # Mount auth headers once
507 set_spec_headers("my-api", {"Authorization": "Bearer secret-token"})
509 # Make requests without repeating headers
510 make_api_request("https://api.example.com/users", spec_id="my-api")
511 """
512 return _set_spec_headers_impl(spec_id, headers)
515def main():
516 """Main entry point for the OpenAPI MCP server."""
517 logging.basicConfig(level=logging.INFO)
518 # Use the module-level mcp instance
519 mcp.run()
522if __name__ == "__main__":
523 main()