extract_json_fields
Extract specific data fields from JSON content in HTTP flows using JSONPath expressions to isolate and retrieve targeted information.
Instructions
Extract specific fields from JSON content in a flow using JSONPath expressions
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| session_id | Yes | The ID of the session | |
| flow_index | Yes | The index of the flow | |
| content_type | Yes | Whether to extract from request or response content | |
| json_paths | Yes | JSONPath expressions to extract (e.g. ['$.data.users', '$.metadata.timestamp']) |
Implementation Reference
- src/mitmproxy_mcp/server.py:263-331 (handler)The core handler function that executes the 'extract_json_fields' tool. It retrieves the flow, parses JSON content from request or response, and extracts specified fields using JSONPath expressions via extract_with_jsonpath.async def extract_json_fields(arguments: dict) -> list[types.TextContent]: """ Extract specific fields from JSON content in a flow using JSONPath expressions. """ session_id = arguments.get("session_id") flow_index = arguments.get("flow_index") content_type = arguments.get("content_type") json_paths = arguments.get("json_paths") if not session_id: return [types.TextContent(type="text", text="Error: Missing session_id")] if flow_index is None: return [types.TextContent(type="text", text="Error: Missing flow_index")] if not content_type: return [types.TextContent(type="text", text="Error: Missing content_type")] if not json_paths: return [types.TextContent(type="text", text="Error: Missing json_paths")] try: flows = await get_flows_from_dump(session_id) try: flow = flows[flow_index] if flow.type != "http": return [types.TextContent(type="text", text=f"Error: Flow {flow_index} is not an HTTP flow")] request = flow.request response = flow.response # Determine which content to extract from content = None headers = None if content_type == "request": content = request.content headers = dict(request.headers) elif content_type == "response": if not response: return [types.TextContent(type="text", text=f"Error: Flow {flow_index} has no response")] content = response.content headers = dict(response.headers) else: return [types.TextContent(type="text", text=f"Error: Invalid content_type. Must be 'request' or 'response'")] # Parse the content json_content = parse_json_content(content, headers) # Only extract from JSON content if not isinstance(json_content, (dict, list)): return [types.TextContent(type="text", text=f"Error: The {content_type} content is not valid JSON")] # Extract fields result = {} for path in json_paths: try: extracted = extract_with_jsonpath(json_content, path) result[path] = extracted except Exception as e: result[path] = f"Error extracting path: {str(e)}" return [types.TextContent(type="text", text=json.dumps(result, indent=2))] except IndexError: return [types.TextContent(type="text", text=f"Error: Flow index {flow_index} out of range")] except FileNotFoundError: return [types.TextContent(type="text", text="Error: Session not found")] except Exception as e: return [types.TextContent(type="text", text=f"Error extracting JSON fields: {str(e)}")]
- src/mitmproxy_mcp/server.py:78-107 (registration)Registration of the 'extract_json_fields' tool in the list_tools handler, including the input schema definition.types.Tool( name="extract_json_fields", description="Extract specific fields from JSON content in a flow using JSONPath expressions", inputSchema={ "type": "object", "properties": { "session_id": { "type": "string", "description": "The ID of the session" }, "flow_index": { "type": "integer", "description": "The index of the flow" }, "content_type": { "type": "string", "enum": ["request", "response"], "description": "Whether to extract from request or response content" }, "json_paths": { "type": "array", "items": { "type": "string" }, "description": "JSONPath expressions to extract (e.g. ['$.data.users', '$.metadata.timestamp'])" } }, "required": ["session_id", "flow_index", "content_type", "json_paths"] } ),
- src/mitmproxy_mcp/server.py:414-415 (registration)Dispatcher in the call_tool handler that routes calls to the extract_json_fields function.elif name == "extract_json_fields": return await extract_json_fields(arguments)