extract_json_fields
Extract specific fields from JSON content in HTTP requests or responses using JSONPath expressions. Specify session, flow index, and content type to retrieve targeted data efficiently.
Instructions
Extract specific fields from JSON content in a flow using JSONPath expressions
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| content_type | Yes | Whether to extract from request or response content | |
| flow_index | Yes | The index of the flow | |
| json_paths | Yes | JSONPath expressions to extract (e.g. ['$.data.users', '$.metadata.timestamp']) | |
| session_id | Yes | The ID of the session |
Input Schema (JSON Schema)
{
"properties": {
"content_type": {
"description": "Whether to extract from request or response content",
"enum": [
"request",
"response"
],
"type": "string"
},
"flow_index": {
"description": "The index of the flow",
"type": "integer"
},
"json_paths": {
"description": "JSONPath expressions to extract (e.g. ['$.data.users', '$.metadata.timestamp'])",
"items": {
"type": "string"
},
"type": "array"
},
"session_id": {
"description": "The ID of the session",
"type": "string"
}
},
"required": [
"session_id",
"flow_index",
"content_type",
"json_paths"
],
"type": "object"
}
Implementation Reference
- src/mitmproxy_mcp/server.py:263-331 (handler)The main handler function that implements the 'extract_json_fields' tool. It retrieves the flow, parses the specified request/response content as JSON, and extracts values using the provided 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() function, including name, description, and detailed input schema for validation.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)Tool dispatch in the @server.call_tool() handler that routes calls to the extract_json_fields function.elif name == "extract_json_fields": return await extract_json_fields(arguments)
- src/mitmproxy_mcp/server.py:81-106 (schema)JSON Schema defining the input parameters for the 'extract_json_fields' tool.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"] }