Fetch JSONPath MCP

Instructions

Implementation Reference

• src/jsonrpc_mcp/utils.py:385-487 (handler)

  Core implementation of batch JSON extraction: groups identical requests to fetch unique URLs once, applies JSONPath extraction to each pattern, handles errors per request and per extraction, maintains original order.
  async def batch_extract_json(url_patterns: list[dict[str, Any]]) -> list[dict[str, Any]]: """ Batch extract JSON data from multiple URLs with different patterns. Optimized to fetch each unique URL only once for the same method/data combination. Args: url_patterns: List of dicts with 'url', optional 'pattern', 'method', 'data', 'headers' keys Returns: List of dictionaries with extraction results """ # Group requests by URL and request parameters to minimize HTTP requests request_groups = {} for i, item in enumerate(url_patterns): url = item.get("url", "") pattern = item.get("pattern", "") method = item.get("method", "GET") data = item.get("data") headers = item.get("headers") if not url: # Handle missing URL case immediately continue # Create a unique key for the same URL with same request parameters import hashlib request_key = f"{url}:{method}:{json.dumps(data, sort_keys=True) if data else ''}:{json.dumps(headers, sort_keys=True) if headers else ''}" request_hash = hashlib.md5(request_key.encode()).hexdigest() if request_hash not in request_groups: request_groups[request_hash] = {"url": url, "method": method, "data": data, "headers": headers, "patterns": []} request_groups[request_hash]["patterns"].append((i, pattern)) # Fetch each unique request once async def fetch_and_extract_for_request(request_info: dict[str, Any]) -> list[tuple[int, dict[str, Any]]]: url = request_info["url"] method = request_info["method"] data = request_info["data"] headers = request_info["headers"] patterns_with_indices = request_info["patterns"] try: content = await fetch_url_content(url, as_json=True, method=method, data=data, headers=headers) results = [] for index, pattern in patterns_with_indices: try: extracted = extract_json(content, pattern) results.append((index, { "url": url, "pattern": pattern, "method": method, "success": True, "content": extracted })) except Exception as e: results.append((index, { "url": url, "pattern": pattern, "method": method, "success": False, "error": str(e) })) return results except Exception as e: # If URL fetch fails, all patterns for this request fail results = [] for index, pattern in patterns_with_indices: results.append((index, { "url": url, "pattern": pattern, "method": method, "success": False, "error": str(e) })) return results # Create tasks for each unique request tasks = [fetch_and_extract_for_request(request_info) for request_info in request_groups.values()] request_results = await asyncio.gather(*tasks) # Flatten results and sort by original index to maintain order all_results = [] for request_result_group in request_results: all_results.extend(request_result_group) # Handle missing URLs for i, item in enumerate(url_patterns): url = item.get("url", "") pattern = item.get("pattern", "") method = item.get("method", "GET") if not url: all_results.append((i, { "url": url, "pattern": pattern, "method": method, "success": False, "error": "Missing URL" })) # Sort by index and return just the results all_results.sort(key=lambda x: x[0]) return [result for _, result in all_results]
• src/jsonrpc_mcp/server.py:226-232 (handler)

  MCP @server.call_tool() dispatch handler for 'batch-fetch-json': validates 'requests' input, calls batch_extract_json utility, serializes result to JSON.
  elif tool_name == "batch-fetch-json": requests = args.get("requests", []) if not isinstance(requests, list) or not requests: result = "Failed to call tool, error: Missing or empty 'requests' array" else: response_result = await batch_extract_json(requests) result = json.dumps(response_result)
• src/jsonrpc_mcp/server.py:94-144 (registration)

  Registration of 'batch-fetch-json' tool in @server.list_tools(), including full description and input schema definition.
  types.Tool( name="batch-fetch-json", description=( "Batch extract JSON content from multiple URLs with different extended JSONPath patterns. " "Supports all JSONPath extensions and optimizes by fetching each unique request only once. " "Executes requests concurrently for better performance. Supports different HTTP methods." ), inputSchema={ "type": "object", "properties": { "requests": { "type": "array", "description": "Array of request objects", "items": { "type": "object", "properties": { "url": { "type": "string", "description": "The URL to get JSON from", }, "pattern": { "type": "string", "description": ( "Extended JSONPath pattern (optional) supporting: " "Basic: 'foo[*].baz'; Extensions: '$.data.`len`'; " "Filtering: '$.items[?(@.price > 10)]'; " "Arithmetic: '$.a + $.b'; Text ops: '$.text.`sub(/old/, new)`'" ), }, "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"], }, }, }, "required": ["requests"], }, ),
• src/jsonrpc_mcp/fastmcp_server.py:116-142 (handler)

  Alternative FastMCP @mcp.tool() handler for batch_fetch_json: uses Pydantic BatchRequest for schema validation, delegates to batch_extract_json, ensures JSON serialization.
  @mcp.tool() async def batch_fetch_json(requests: list[BatchRequest]) -> list[dict]: """ Batch extract JSON content from multiple URLs with different JSONPath patterns. Supports different HTTP methods. Executes requests concurrently for better performance. Args: requests: Array of request objects with 'url', optional 'pattern', 'method', 'data', 'headers' Returns: List of results with success/failure information """ # Convert Pydantic models to dicts for utility function request_dicts = [ { "url": req.url, "pattern": req.pattern, "method": req.method, "data": req.data, "headers": req.headers } for req in requests ] results = await batch_extract_json(request_dicts) # Ensure all results are JSON serializable return ensure_json_serializable(results)
• src/jsonrpc_mcp/fastmcp_server.py:30-36 (schema)

  Pydantic model defining input schema for batch_fetch_json tool requests in FastMCP implementation.
  class BatchRequest(BaseModel): url: str pattern: str = "" method: str = "GET" data: dict | str | None = None headers: dict[str, str] | None = None