Coverage for src/openapi_navigator/server.py: 67%
137 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-01-22 08:42 +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(
45 url: str, spec_id: Optional[str] = None, verify_ssl: bool = True
46) -> str:
47 """
48 Load an OpenAPI specification from a URL.
50 Args:
51 url: URL to the OpenAPI spec (YAML or JSON)
52 spec_id: Optional custom identifier for the spec. If not provided, will use 'url:{url}'
53 verify_ssl: Whether to verify SSL certificates (default: True).
54 Set to False to ignore invalid or self-signed certificates.
56 Returns:
57 The spec ID that was assigned to the loaded specification
58 """
59 try:
60 return _spec_manager.load_spec_from_url(url, spec_id, verify_ssl)
61 except Exception as e:
62 logger.error(f"Failed to load spec from URL: {e}")
63 raise
66@mcp.tool
67def unload_spec(spec_id: str) -> str:
68 """
69 Unload an OpenAPI specification from memory.
71 Args:
72 spec_id: ID of the loaded spec to unload
74 Returns:
75 Confirmation message
76 """
77 success = _spec_manager.unload_spec(spec_id)
78 if success:
79 return f"Successfully unloaded spec: {spec_id}"
80 else:
81 return f"Spec not found or already unloaded: {spec_id}"
84@mcp.tool
85def list_loaded_specs() -> List[str]:
86 """
87 List all currently loaded OpenAPI specifications.
89 Returns:
90 List of spec IDs that are currently loaded
91 """
92 return _spec_manager.list_loaded_specs()
95@mcp.tool
96def get_endpoint(
97 spec_id: str, path: str, method: str, summary_only: bool = False
98) -> Optional[Dict[str, Any]]:
99 """
100 Get the operation definition for a specific endpoint.
102 Args:
103 spec_id: ID of the loaded spec to query
104 path: API path (e.g., '/users/{id}')
105 method: HTTP method (e.g., 'GET', 'POST')
106 summary_only: If True, return only essential fields to reduce token usage (default: False)
108 Returns:
109 The operation object from the OpenAPI spec (full or summary), or None if not found
110 """
111 spec = _spec_manager.get_spec(spec_id)
112 if not spec:
113 raise ValueError(f"No spec found with ID: {spec_id}")
115 endpoint = spec.get_endpoint(path, method.upper())
116 if not endpoint:
117 return None
119 if summary_only:
120 # Return only essential information to reduce token usage
121 return {
122 "summary": endpoint.get("summary", ""),
123 "description": endpoint.get("description", ""),
124 "operationId": endpoint.get("operationId", ""),
125 "tags": endpoint.get("tags", []),
126 "parameters": [
127 {
128 "name": p.get("name", ""),
129 "in": p.get("in", ""),
130 "required": p.get("required", False),
131 "type": p.get("schema", {}).get("type", p.get("type", "")),
132 "description": p.get("description", ""),
133 }
134 for p in endpoint.get("parameters", [])
135 ],
136 "responses": {
137 code: {
138 "description": resp.get("description", ""),
139 "content_types": (
140 list(resp.get("content", {}).keys())
141 if "content" in resp
142 else []
143 ),
144 }
145 for code, resp in endpoint.get("responses", {}).items()
146 },
147 "requestBody": (
148 {
149 "required": endpoint.get("requestBody", {}).get("required", False),
150 "content_types": list(
151 endpoint.get("requestBody", {}).get("content", {}).keys()
152 ),
153 }
154 if "requestBody" in endpoint
155 else None
156 ),
157 }
159 return endpoint
162@mcp.tool
163def search_endpoints(
164 spec_id: str, query: str, limit: int = 50, offset: int = 0
165) -> Dict[str, Any]:
166 """
167 Search endpoints using fuzzy matching across paths, summaries, and operation IDs with pagination.
169 To get a full list of all endpoints, use an empty string "" or a very short query like "a" as the search term.
170 The search will return all endpoints with a relevance score of 100 when the query is very short.
172 Args:
173 spec_id: ID of the loaded spec to query
174 query: Search query string. Use "" or "a" to get all endpoints.
175 limit: Maximum number of results to return (default 50, max 200)
176 offset: Number of results to skip for pagination (default 0)
178 Returns:
179 Dictionary containing:
180 - endpoints: List of matching endpoints with relevance scores
181 - total: Total number of matches (before pagination)
182 - limit: Applied limit
183 - offset: Applied offset
184 - has_more: Whether there are more results available
185 """
186 spec = _spec_manager.get_spec(spec_id)
187 if not spec:
188 raise ValueError(f"No spec found with ID: {spec_id}")
190 return spec.search_endpoints(query, limit, offset)
193@mcp.tool
194def get_schema(spec_id: str, schema_name: str) -> Optional[Dict[str, Any]]:
195 """
196 Get a specific schema definition from a loaded OpenAPI specification.
198 Args:
199 spec_id: ID of the loaded spec to query
200 schema_name: Name of the schema to retrieve
202 Returns:
203 The raw schema object from the OpenAPI spec, or None if not found
204 """
205 spec = _spec_manager.get_spec(spec_id)
206 if not spec:
207 raise ValueError(f"No spec found with ID: {spec_id}")
209 return spec.get_schema(schema_name)
212@mcp.tool
213def search_schemas(
214 spec_id: str, query: str, limit: int = 50, offset: int = 0
215) -> Dict[str, Any]:
216 """
217 Search schema names using fuzzy matching with pagination.
219 To get a full list of all schemas, use an empty string "" or a very short query like "a" as the search term.
220 The search will return all schemas with a relevance score of 100 when the query is very short.
222 Args:
223 spec_id: ID of the loaded spec to query
224 query: Search query string. Use "" or "a" to get all schemas.
225 limit: Maximum number of results to return (default 50, max 200)
226 offset: Number of results to skip for pagination (default 0)
228 Returns:
229 Dictionary containing:
230 - schemas: List of matching schema names with relevance scores
231 - total: Total number of matches (before pagination)
232 - limit: Applied limit
233 - offset: Applied offset
234 - has_more: Whether there are more results available
235 """
236 spec = _spec_manager.get_spec(spec_id)
237 if not spec:
238 raise ValueError(f"No spec found with ID: {spec_id}")
240 return spec.search_schemas(query, limit, offset)
243@mcp.tool
244def get_spec_metadata(spec_id: str) -> Dict[str, Any]:
245 """
246 Get comprehensive metadata about a loaded OpenAPI specification.
248 This includes information about the spec version, title, description, base path,
249 servers, contact info, license, and counts of endpoints and schemas.
251 Args:
252 spec_id: ID of the loaded spec to query
254 Returns:
255 Dictionary containing spec metadata including base path and help text
256 """
257 spec = _spec_manager.get_spec(spec_id)
258 if not spec:
259 raise ValueError(f"No spec found with ID: {spec_id}")
261 return spec.get_spec_metadata()
264def _make_api_request_impl(
265 url: str,
266 method: str = "GET",
267 headers: Optional[Dict[str, str]] = None,
268 params: Optional[Dict[str, str]] = None,
269 data: Optional[str] = None,
270 timeout: int = 30,
271 spec_id: Optional[str] = None,
272) -> Dict[str, Any]:
273 """
274 Make a generic REST API request with full control over method, headers, parameters, and body.
276 This tool enables direct interaction with REST APIs, complementing the OpenAPI exploration tools
277 by allowing you to actually call the endpoints you've discovered.
279 Args:
280 url: The full URL to make the request to
281 method: HTTP method (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS). Defaults to GET.
282 headers: Optional dictionary of HTTP headers to include in the request
283 params: Optional dictionary of URL parameters to append to the URL
284 data: Optional request body data as a string (JSON, XML, form data, etc.)
285 timeout: Request timeout in seconds. Defaults to 30.
286 spec_id: Optional spec ID to use for default headers. If provided, headers from
287 set_spec_headers will be merged (request headers take precedence).
289 Returns:
290 Dictionary containing:
291 - status_code: HTTP status code
292 - headers: Response headers as a dictionary
293 - body: Raw response body as string
294 - json: Parsed JSON response (if response is valid JSON, otherwise None)
295 - url: Final URL after any redirects
296 - elapsed_ms: Request duration in milliseconds
297 - method: The HTTP method that was used
299 Raises:
300 ValueError: If the URL is invalid or method is not supported, or spec_id doesn't exist
301 ConnectionError: If the request fails due to network issues
302 TimeoutError: If the request times out
303 """
304 # Validate method
305 valid_methods = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
306 method = method.upper()
307 if method not in valid_methods:
308 raise ValueError(
309 f"Invalid HTTP method: {method}. Must be one of {valid_methods}"
310 )
312 # Validate URL
313 if not url or not isinstance(url, str):
314 raise ValueError("URL must be a non-empty string")
316 # Merge headers from spec if spec_id provided
317 merged_headers = {}
318 if spec_id:
319 spec = _spec_manager.get_spec(spec_id)
320 if not spec:
321 raise ValueError(f"No spec found with ID: {spec_id}")
322 # Start with spec's default headers
323 merged_headers = spec.get_headers()
325 # Override with explicit headers (request headers take precedence)
326 if headers:
327 merged_headers.update(headers)
329 try:
330 # Prepare the request
331 request_kwargs = {
332 "method": method,
333 "url": url,
334 "timeout": timeout,
335 }
337 if merged_headers: # Changed from `if headers:`
338 request_kwargs["headers"] = merged_headers
340 if params:
341 request_kwargs["params"] = params
343 if data:
344 request_kwargs["data"] = data
346 # Make the request and time it
347 start_time = time.time()
348 logger.info(f"Making {method} request to {url}")
350 response = requests.request(**request_kwargs)
352 end_time = time.time()
353 elapsed_ms = int((end_time - start_time) * 1000)
355 # Parse JSON response if possible
356 response_json = None
357 try:
358 if response.text.strip(): # Only try to parse if there's content
359 response_json = response.json()
360 except (json_module.JSONDecodeError, ValueError):
361 # Response is not JSON, that's fine
362 pass
364 # Build response dictionary
365 result = {
366 "status_code": response.status_code,
367 "headers": dict(response.headers),
368 "body": response.text,
369 "json": response_json,
370 "url": response.url,
371 "elapsed_ms": elapsed_ms,
372 "method": method,
373 }
375 logger.info(
376 f"Request completed: {method} {url} -> {response.status_code} ({elapsed_ms}ms)"
377 )
378 return result
380 except requests.exceptions.Timeout as e:
381 logger.error(f"Request timed out: {method} {url}")
382 raise TimeoutError(f"Request timed out after {timeout} seconds: {str(e)}")
384 except requests.exceptions.ConnectionError as e:
385 logger.error(f"Connection error: {method} {url} - {str(e)}")
386 raise ConnectionError(f"Failed to connect to {url}: {str(e)}")
388 except requests.exceptions.RequestException as e:
389 logger.error(f"Request failed: {method} {url} - {str(e)}")
390 raise Exception(f"Request failed: {str(e)}")
392 except Exception as e:
393 logger.error(f"Unexpected error during request: {method} {url} - {str(e)}")
394 raise Exception(f"Unexpected error: {str(e)}")
397@mcp.tool
398def make_api_request(
399 url: str,
400 method: str = "GET",
401 headers: dict = None,
402 params: dict = None,
403 data: str = None,
404 timeout: int = 30,
405 spec_id: str = None,
406) -> dict:
407 """
408 Make a generic REST API request with full control over method, headers, parameters, and body.
410 This tool enables direct interaction with REST APIs, complementing the OpenAPI exploration tools
411 by allowing you to actually call the endpoints you've discovered.
413 Args:
414 url: The full URL to make the request to
415 method: HTTP method (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS). Defaults to GET.
416 headers: Dictionary of HTTP headers to include in the request (optional)
417 params: Dictionary of URL parameters to append to the URL (optional)
418 data: Request body data as a string (JSON, XML, form data, etc.) (optional)
419 timeout: Request timeout in seconds. Defaults to 30.
420 spec_id: Optional spec ID to automatically apply headers from set_spec_headers.
421 Request headers take precedence over spec headers. (optional)
423 Returns:
424 Dictionary containing:
425 - status_code: HTTP status code
426 - headers: Response headers as a dictionary
427 - body: Raw response body as string
428 - json: Parsed JSON response (if response is valid JSON, otherwise None)
429 - url: Final URL after any redirects
430 - elapsed_ms: Request duration in milliseconds
431 - method: The HTTP method that was used
433 Raises:
434 ValueError: If the URL is invalid or method is not supported, or spec_id doesn't exist
435 ConnectionError: If the request fails due to network issues
436 TimeoutError: If the request times out
438 Example with spec_id:
439 # Load and configure a spec with auth
440 load_spec_from_url("https://api.example.com/openapi.json", "my-api")
441 set_spec_headers("my-api", {"Authorization": "Bearer token123"})
443 # Make requests without repeating auth headers
444 make_api_request("https://api.example.com/users", spec_id="my-api")
446 # Override specific headers while keeping others from spec
447 make_api_request(
448 "https://api.example.com/admin",
449 spec_id="my-api",
450 headers={"X-Admin-Token": "admin456"}
451 )
452 """
453 return _make_api_request_impl(url, method, headers, params, data, timeout, spec_id)
456def _set_spec_headers_impl(spec_id: str, headers: dict = None) -> str:
457 """
458 Internal implementation for setting default headers on a spec.
460 Args:
461 spec_id: ID of the loaded spec to set headers for
462 headers: Dictionary of HTTP headers to use by default
464 Returns:
465 Confirmation message
467 Raises:
468 ValueError: If the spec_id doesn't exist
469 """
470 spec = _spec_manager.get_spec(spec_id)
471 if not spec:
472 raise ValueError(f"No spec found with ID: {spec_id}")
474 if headers is None:
475 headers = {}
477 spec.set_headers(headers)
479 header_count = len(headers)
480 if header_count == 0:
481 return f"Successfully cleared headers for spec: {spec_id}"
482 else:
483 return f"Successfully set {header_count} default header(s) for spec: {spec_id}"
486@mcp.tool
487def set_spec_headers(spec_id: str, headers: dict = None) -> str:
488 """
489 Set default headers for API requests to a loaded OpenAPI spec.
491 This allows you to "mount" authentication or other headers to a spec once,
492 rather than passing them on every make_api_request call. When making requests
493 with make_api_request, you can reference the spec_id to automatically apply
494 these headers.
496 Args:
497 spec_id: ID of the loaded spec to set headers for
498 headers: Dictionary of HTTP headers to use by default (optional, defaults to empty)
500 Returns:
501 Confirmation message
503 Raises:
504 ValueError: If the spec_id doesn't exist
506 Example:
507 # Load a spec
508 load_spec_from_url("https://api.example.com/openapi.json", "my-api")
510 # Mount auth headers once
511 set_spec_headers("my-api", {"Authorization": "Bearer secret-token"})
513 # Make requests without repeating headers
514 make_api_request("https://api.example.com/users", spec_id="my-api")
515 """
516 return _set_spec_headers_impl(spec_id, headers)
519def main():
520 """Main entry point for the OpenAPI MCP server."""
521 logging.basicConfig(level=logging.INFO)
522 # Use the module-level mcp instance
523 mcp.run()
526if __name__ == "__main__":
527 main()