mitmproxy-mcp MCP Server

Instructions

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)