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
`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)
```

mitmproxy-mcp MCP Server