Fetch JSONPath MCP

Instructions

Implementation Reference

• src/jsonrpc_mcp/server.py:13-59 (registration)

  Registers the 'fetch-json' tool with its description and input schema in the MCP server's list_tools handler.
  @server.list_tools() async def handle_list_tools() -> list[types.Tool]: return [ types.Tool( name="fetch-json", description=( "Extract JSON content from a URL using JSONPath with extended features. " "Supports extensions like len, keys, filtering, arithmetic operations, and more. " "If 'pattern' is omitted or empty, the entire JSON document is returned. " "Supports different HTTP methods (default: GET)." ), inputSchema={ "type": "object", "properties": { "url": { "type": "string", "description": "The URL to get raw JSON from", }, "pattern": { "type": "string", "description": ( "Extended JSONPath pattern supporting: " "Basic: 'foo[*].baz', 'bar.items[*]'; " "Extensions: '$.data.`len`', '$.users.`keys`', '$.field.`str()`'; " "Filtering: '$.items[?(@.price > 10)]', '$.users[?name = \"John\"]'; " "Arithmetic: '$.a + $.b', '$.items[*].price * 1.2'; " "Text ops: '$.text.`sub(/old/, new)`', '$.csv.`split(\",\")'" ), }, "method": { "type": "string", "description": "HTTP method to use (GET, POST, PUT, DELETE, PATCH, etc.). Default is GET.", "default": "GET" }, "data": { "type": ["object", "string", "null"], "description": "Request body data for POST/PUT/PATCH requests. Can be a JSON object or string.", }, "headers": { "type": "object", "description": "Additional HTTP headers to include in the request", "additionalProperties": {"type": "string"} } }, "required": ["url"], }, ),
• src/jsonrpc_mcp/server.py:203-215 (handler)

  Main dispatch logic in call_tool handler that processes arguments and calls the specific fetch-json implementation.
  if tool_name == "fetch-json": url = args.get("url") if not url or not isinstance(url, str): result = "Failed to call tool, error: Missing required property: url" else: method = args.get("method", "GET") data = args.get("data") headers = args.get("headers") pattern = args.get("pattern", "") response_result = await handle_get_json(url, pattern, method, data, headers) result = json.dumps(response_result) elif tool_name == "fetch-text":
• src/jsonrpc_mcp/server.py:251-261 (handler)

  Core handler function for 'fetch-json' tool: fetches JSON content from URL and applies JSONPath pattern extraction.
  async def handle_get_json( url: str, pattern: str = "", method: str = "GET", data: dict | str | None = None, headers: dict[str, str] | None = None ) -> list: """Handle single JSON extraction request.""" content = await fetch_url_content(url, as_json=True, method=method, data=data, headers=headers) return extract_json(content, pattern)
• src/jsonrpc_mcp/utils.py:11-60 (helper)

  Helper function to parse JSON and extract data using extended JSONPath patterns, supporting various operations and filtering.
  def extract_json(json_str: str, pattern: str) -> list: """ Extract JSON values from a JSON string using a JSONPath pattern. Supports both standard JSONPath and extended JSONPath features including: - Extensions: len, keys, str(), sub(), split(), sorted, filter - Arithmetic operations: +, -, *, / - Advanced filtering: [?(@.field > value)] - And more extended features from jsonpath-ng.ext If the pattern is empty or refers to the root ("$", "$.", or "@"), the entire JSON document is returned as a single-element list. Args: json_str: JSON string to parse pattern: JSONPath pattern to extract data (supports extensions) Returns: List of extracted values Raises: json.JSONDecodeError: If json_str is not valid JSON Exception: If JSONPath pattern is invalid """ try: d = json.loads(json_str) except json.JSONDecodeError as e: raise json.JSONDecodeError(f"Invalid JSON: {e.msg}", e.doc, e.pos) if not pattern or pattern.strip() in {"", "$", "$.", "@"}: return [d] # Basic security: limit pattern length to prevent abuse if len(pattern) > 1000: raise ValueError("JSONPath pattern too long (max 1000 characters)") # Try extended parser first (supports all extensions) try: jsonpath_expr = ext_parse(pattern) return [match.value for match in jsonpath_expr.find(d)] except Exception as ext_error: # Fallback to basic parser if extended parsing fails try: jsonpath_expr = parse(pattern) return [match.value for match in jsonpath_expr.find(d)] except Exception as basic_error: # Report the more descriptive error from extended parser if available error_msg = str(ext_error) if ext_error else str(basic_error) raise Exception(f"Invalid JSONPath pattern '{pattern}': {error_msg}")
• src/jsonrpc_mcp/utils.py:232-343 (helper)

  Helper function to fetch HTTP content from URL with support for various methods, headers, data, validation, and JSON checking.
  async def fetch_url_content( url: str, as_json: bool = True, method: str = "GET", data: dict | str | None = None, headers: dict[str, str] | None = None, output_format: str = "markdown" ) -> str: """ Fetch content from a URL using different HTTP methods. Args: url: URL to fetch content from as_json: If True, validates content as JSON; if False, returns text content method: HTTP method (GET, POST, PUT, DELETE, etc.) data: Request body data (for POST/PUT requests) headers: Additional headers to include in the request output_format: If as_json=False, output format - "markdown", "clean_text", or "raw_html" Returns: String content from the URL (JSON, Markdown, clean text, or raw HTML) Raises: httpx.RequestError: For network-related errors json.JSONDecodeError: If as_json=True and content is not valid JSON ValueError: If URL is invalid or unsafe """ # Validate URL first validate_url(url) config = await get_http_client_config() max_size = config.pop("max_size", 10 * 1024 * 1024) # Remove from client config # Merge additional headers with config headers (user headers override defaults) if headers: if config.get("headers"): config["headers"].update(headers) else: config["headers"] = headers async with httpx.AsyncClient(**config) as client: # Handle different HTTP methods method = method.upper() if method == "GET": response = await client.get(url) elif method == "POST": if isinstance(data, dict): response = await client.post(url, json=data) else: response = await client.post(url, content=data) elif method == "PUT": if isinstance(data, dict): response = await client.put(url, json=data) else: response = await client.put(url, content=data) elif method == "DELETE": response = await client.delete(url) elif method == "PATCH": if isinstance(data, dict): response = await client.patch(url, json=data) else: response = await client.patch(url, content=data) elif method == "HEAD": response = await client.head(url) elif method == "OPTIONS": response = await client.options(url) else: # For any other method, use the generic request method if isinstance(data, dict): response = await client.request(method, url, json=data) else: response = await client.request(method, url, content=data) response.raise_for_status() # Check response size content_length = len(response.content) if content_length > max_size: raise ValueError(f"Response size ({content_length} bytes) exceeds maximum allowed ({max_size} bytes)") if as_json: # For JSON responses, use response.text directly (no compression expected) content_to_parse = response.text if not content_to_parse: # If response.text is empty, try decoding content directly try: content_to_parse = response.content.decode('utf-8') except UnicodeDecodeError: content_to_parse = "" if content_to_parse: try: json.loads(content_to_parse) return content_to_parse except json.JSONDecodeError: # If text parsing fails, try content decoding as fallback if content_to_parse == response.text: try: fallback_content = response.content.decode('utf-8') json.loads(fallback_content) return fallback_content except (json.JSONDecodeError, UnicodeDecodeError): pass raise json.JSONDecodeError("Response is not valid JSON", content_to_parse, 0) else: # Empty response return "" else: # For text content, apply format conversion return extract_text_content(response.text, output_format)