fetch-json
Extract specific data from JSON APIs using JSONPath patterns to reduce token usage by fetching only required content from URLs with support for HTTP methods and request customization.
Instructions
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).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| url | Yes | The URL to get raw JSON from | |
| pattern | No | 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 | No | HTTP method to use (GET, POST, PUT, DELETE, PATCH, etc.). Default is GET. | GET |
| data | No | Request body data for POST/PUT/PATCH requests. Can be a JSON object or string. | |
| headers | No | Additional HTTP headers to include in the request |
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)